1 # -*- test-case-name: twisted.conch.test.test_telnet -*-
2 # Copyright (c) Twisted Matrix Laboratories.
3 # See LICENSE for details.
6 Tests for L{twisted.conch.telnet}.
9 from zope.interface import implements
10 from zope.interface.verify import verifyObject
12 from twisted.internet import defer
14 from twisted.conch import telnet
16 from twisted.trial import unittest
17 from twisted.test import proto_helpers
21 implements(telnet.ITelnetProtocol)
31 self.enabledLocal = []
32 self.enabledRemote = []
33 self.disabledLocal = []
34 self.disabledRemote = []
36 def makeConnection(self, transport):
37 d = transport.negotiationMap = {}
38 d['\x12'] = self.neg_TEST_COMMAND
40 d = transport.commandMap = transport.commandMap.copy()
41 for cmd in ('NOP', 'DM', 'BRK', 'IP', 'AO', 'AYT', 'EC', 'EL', 'GA'):
42 d[getattr(telnet, cmd)] = lambda arg, cmd=cmd: self.calls.append(cmd)
44 def dataReceived(self, bytes):
47 def connectionLost(self, reason):
50 def neg_TEST_COMMAND(self, payload):
53 def enableLocal(self, option):
54 if option in self.localEnableable:
55 self.enabledLocal.append(option)
59 def disableLocal(self, option):
60 self.disabledLocal.append(option)
62 def enableRemote(self, option):
63 if option in self.remoteEnableable:
64 self.enabledRemote.append(option)
68 def disableRemote(self, option):
69 self.disabledRemote.append(option)
73 class TestInterfaces(unittest.TestCase):
74 def test_interface(self):
76 L{telnet.TelnetProtocol} implements L{telnet.ITelnetProtocol}
78 p = telnet.TelnetProtocol()
79 verifyObject(telnet.ITelnetProtocol, p)
83 class TelnetTransportTestCase(unittest.TestCase):
85 Tests for L{telnet.TelnetTransport}.
88 self.p = telnet.TelnetTransport(TestProtocol)
89 self.t = proto_helpers.StringTransport()
90 self.p.makeConnection(self.t)
92 def testRegularBytes(self):
93 # Just send a bunch of bytes. None of these do anything
94 # with telnet. They should pass right through to the
98 L = ["here are some bytes la la la",
99 "some more arrive here",
100 "lots of bytes to play with",
105 self.p.dataReceived(b)
107 self.assertEqual(h.bytes, ''.join(L))
109 def testNewlineHandling(self):
110 # Send various kinds of newlines and make sure they get translated
114 L = ["here is the first line\r\n",
115 "here is the second line\r\0",
116 "here is the third line\r\n",
117 "here is the last line\r\0"]
120 self.p.dataReceived(b)
122 self.assertEqual(h.bytes, L[0][:-2] + '\n' +
127 def testIACEscape(self):
128 # Send a bunch of bytes and a couple quoted \xFFs. Unquoted,
129 # \xFF is a telnet command. Quoted, one of them from each pair
130 # should be passed through to the application layer.
133 L = ["here are some bytes\xff\xff with an embedded IAC",
134 "and here is a test of a border escape\xff",
135 "\xff did you get that IAC?"]
138 self.p.dataReceived(b)
140 self.assertEqual(h.bytes, ''.join(L).replace('\xff\xff', '\xff'))
142 def _simpleCommandTest(self, cmdName):
143 # Send a single simple telnet command and make sure
144 # it gets noticed and the appropriate method gets
148 cmd = telnet.IAC + getattr(telnet, cmdName)
149 L = ["Here's some bytes, tra la la",
150 "But ono!" + cmd + " an interrupt"]
153 self.p.dataReceived(b)
155 self.assertEqual(h.calls, [cmdName])
156 self.assertEqual(h.bytes, ''.join(L).replace(cmd, ''))
158 def testInterrupt(self):
159 self._simpleCommandTest("IP")
161 def testNoOperation(self):
162 self._simpleCommandTest("NOP")
164 def testDataMark(self):
165 self._simpleCommandTest("DM")
168 self._simpleCommandTest("BRK")
170 def testAbortOutput(self):
171 self._simpleCommandTest("AO")
173 def testAreYouThere(self):
174 self._simpleCommandTest("AYT")
176 def testEraseCharacter(self):
177 self._simpleCommandTest("EC")
179 def testEraseLine(self):
180 self._simpleCommandTest("EL")
182 def testGoAhead(self):
183 self._simpleCommandTest("GA")
185 def testSubnegotiation(self):
186 # Send a subnegotiation command and make sure it gets
187 # parsed and that the correct method is called.
190 cmd = telnet.IAC + telnet.SB + '\x12hello world' + telnet.IAC + telnet.SE
191 L = ["These are some bytes but soon" + cmd,
192 "there will be some more"]
195 self.p.dataReceived(b)
197 self.assertEqual(h.bytes, ''.join(L).replace(cmd, ''))
198 self.assertEqual(h.subcmd, list("hello world"))
200 def testSubnegotiationWithEmbeddedSE(self):
201 # Send a subnegotiation command with an embedded SE. Make sure
202 # that SE gets passed to the correct method.
205 cmd = (telnet.IAC + telnet.SB +
207 telnet.IAC + telnet.SE)
209 L = ["Some bytes are here" + cmd + "and here",
213 self.p.dataReceived(b)
215 self.assertEqual(h.bytes, ''.join(L).replace(cmd, ''))
216 self.assertEqual(h.subcmd, [telnet.SE])
218 def testBoundarySubnegotiation(self):
219 # Send a subnegotiation command. Split it at every possible byte boundary
220 # and make sure it always gets parsed and that it is passed to the correct
222 cmd = (telnet.IAC + telnet.SB +
223 '\x12' + telnet.SE + 'hello' +
224 telnet.IAC + telnet.SE)
226 for i in range(len(cmd)):
227 h = self.p.protocol = TestProtocol()
228 h.makeConnection(self.p)
230 a, b = cmd[:i], cmd[i:]
231 L = ["first part" + a,
235 self.p.dataReceived(bytes)
237 self.assertEqual(h.bytes, ''.join(L).replace(cmd, ''))
238 self.assertEqual(h.subcmd, [telnet.SE] + list('hello'))
240 def _enabledHelper(self, o, eL=[], eR=[], dL=[], dR=[]):
241 self.assertEqual(o.enabledLocal, eL)
242 self.assertEqual(o.enabledRemote, eR)
243 self.assertEqual(o.disabledLocal, dL)
244 self.assertEqual(o.disabledRemote, dR)
246 def testRefuseWill(self):
247 # Try to enable an option. The server should refuse to enable it.
248 cmd = telnet.IAC + telnet.WILL + '\x12'
250 bytes = "surrounding bytes" + cmd + "to spice things up"
251 self.p.dataReceived(bytes)
253 self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
254 self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + '\x12')
255 self._enabledHelper(self.p.protocol)
257 def testRefuseDo(self):
258 # Try to enable an option. The server should refuse to enable it.
259 cmd = telnet.IAC + telnet.DO + '\x12'
261 bytes = "surrounding bytes" + cmd + "to spice things up"
262 self.p.dataReceived(bytes)
264 self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
265 self.assertEqual(self.t.value(), telnet.IAC + telnet.WONT + '\x12')
266 self._enabledHelper(self.p.protocol)
268 def testAcceptDo(self):
269 # Try to enable an option. The option is in our allowEnable
270 # list, so we will allow it to be enabled.
271 cmd = telnet.IAC + telnet.DO + '\x19'
272 bytes = 'padding' + cmd + 'trailer'
275 h.localEnableable = ('\x19',)
276 self.p.dataReceived(bytes)
278 self.assertEqual(self.t.value(), telnet.IAC + telnet.WILL + '\x19')
279 self._enabledHelper(h, eL=['\x19'])
281 def testAcceptWill(self):
282 # Same as testAcceptDo, but reversed.
283 cmd = telnet.IAC + telnet.WILL + '\x91'
284 bytes = 'header' + cmd + 'padding'
287 h.remoteEnableable = ('\x91',)
288 self.p.dataReceived(bytes)
290 self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + '\x91')
291 self._enabledHelper(h, eR=['\x91'])
293 def testAcceptWont(self):
294 # Try to disable an option. The server must allow any option to
295 # be disabled at any time. Make sure it disables it and sends
296 # back an acknowledgement of this.
297 cmd = telnet.IAC + telnet.WONT + '\x29'
299 # Jimmy it - after these two lines, the server will be in a state
300 # such that it believes the option to have been previously enabled
301 # via normal negotiation.
302 s = self.p.getOptionState('\x29')
305 bytes = "fiddle dee" + cmd
306 self.p.dataReceived(bytes)
308 self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
309 self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + '\x29')
310 self.assertEqual(s.him.state, 'no')
311 self._enabledHelper(self.p.protocol, dR=['\x29'])
313 def testAcceptDont(self):
314 # Try to disable an option. The server must allow any option to
315 # be disabled at any time. Make sure it disables it and sends
316 # back an acknowledgement of this.
317 cmd = telnet.IAC + telnet.DONT + '\x29'
319 # Jimmy it - after these two lines, the server will be in a state
320 # such that it believes the option to have beenp previously enabled
321 # via normal negotiation.
322 s = self.p.getOptionState('\x29')
325 bytes = "fiddle dum " + cmd
326 self.p.dataReceived(bytes)
328 self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
329 self.assertEqual(self.t.value(), telnet.IAC + telnet.WONT + '\x29')
330 self.assertEqual(s.us.state, 'no')
331 self._enabledHelper(self.p.protocol, dL=['\x29'])
333 def testIgnoreWont(self):
334 # Try to disable an option. The option is already disabled. The
335 # server should send nothing in response to this.
336 cmd = telnet.IAC + telnet.WONT + '\x47'
338 bytes = "dum de dum" + cmd + "tra la la"
339 self.p.dataReceived(bytes)
341 self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
342 self.assertEqual(self.t.value(), '')
343 self._enabledHelper(self.p.protocol)
345 def testIgnoreDont(self):
346 # Try to disable an option. The option is already disabled. The
347 # server should send nothing in response to this. Doing so could
348 # lead to a negotiation loop.
349 cmd = telnet.IAC + telnet.DONT + '\x47'
351 bytes = "dum de dum" + cmd + "tra la la"
352 self.p.dataReceived(bytes)
354 self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
355 self.assertEqual(self.t.value(), '')
356 self._enabledHelper(self.p.protocol)
358 def testIgnoreWill(self):
359 # Try to enable an option. The option is already enabled. The
360 # server should send nothing in response to this. Doing so could
361 # lead to a negotiation loop.
362 cmd = telnet.IAC + telnet.WILL + '\x56'
364 # Jimmy it - after these two lines, the server will be in a state
365 # such that it believes the option to have been previously enabled
366 # via normal negotiation.
367 s = self.p.getOptionState('\x56')
370 bytes = "tra la la" + cmd + "dum de dum"
371 self.p.dataReceived(bytes)
373 self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
374 self.assertEqual(self.t.value(), '')
375 self._enabledHelper(self.p.protocol)
377 def testIgnoreDo(self):
378 # Try to enable an option. The option is already enabled. The
379 # server should send nothing in response to this. Doing so could
380 # lead to a negotiation loop.
381 cmd = telnet.IAC + telnet.DO + '\x56'
383 # Jimmy it - after these two lines, the server will be in a state
384 # such that it believes the option to have been previously enabled
385 # via normal negotiation.
386 s = self.p.getOptionState('\x56')
389 bytes = "tra la la" + cmd + "dum de dum"
390 self.p.dataReceived(bytes)
392 self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
393 self.assertEqual(self.t.value(), '')
394 self._enabledHelper(self.p.protocol)
396 def testAcceptedEnableRequest(self):
397 # Try to enable an option through the user-level API. This
398 # returns a Deferred that fires when negotiation about the option
399 # finishes. Make sure it fires, make sure state gets updated
400 # properly, make sure the result indicates the option was enabled.
401 d = self.p.do('\x42')
404 h.remoteEnableable = ('\x42',)
406 self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + '\x42')
408 self.p.dataReceived(telnet.IAC + telnet.WILL + '\x42')
410 d.addCallback(self.assertEqual, True)
411 d.addCallback(lambda _: self._enabledHelper(h, eR=['\x42']))
415 def test_refusedEnableRequest(self):
417 If the peer refuses to enable an option we request it to enable, the
418 L{Deferred} returned by L{TelnetProtocol.do} fires with an
419 L{OptionRefused} L{Failure}.
421 # Try to enable an option through the user-level API. This returns a
422 # Deferred that fires when negotiation about the option finishes. Make
423 # sure it fires, make sure state gets updated properly, make sure the
424 # result indicates the option was enabled.
425 self.p.protocol.remoteEnableable = ('\x42',)
426 d = self.p.do('\x42')
428 self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + '\x42')
430 s = self.p.getOptionState('\x42')
431 self.assertEqual(s.him.state, 'no')
432 self.assertEqual(s.us.state, 'no')
433 self.assertEqual(s.him.negotiating, True)
434 self.assertEqual(s.us.negotiating, False)
436 self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')
438 d = self.assertFailure(d, telnet.OptionRefused)
439 d.addCallback(lambda ignored: self._enabledHelper(self.p.protocol))
441 lambda ignored: self.assertEqual(s.him.negotiating, False))
445 def test_refusedEnableOffer(self):
447 If the peer refuses to allow us to enable an option, the L{Deferred}
448 returned by L{TelnetProtocol.will} fires with an L{OptionRefused}
451 # Try to offer an option through the user-level API. This returns a
452 # Deferred that fires when negotiation about the option finishes. Make
453 # sure it fires, make sure state gets updated properly, make sure the
454 # result indicates the option was enabled.
455 self.p.protocol.localEnableable = ('\x42',)
456 d = self.p.will('\x42')
458 self.assertEqual(self.t.value(), telnet.IAC + telnet.WILL + '\x42')
460 s = self.p.getOptionState('\x42')
461 self.assertEqual(s.him.state, 'no')
462 self.assertEqual(s.us.state, 'no')
463 self.assertEqual(s.him.negotiating, False)
464 self.assertEqual(s.us.negotiating, True)
466 self.p.dataReceived(telnet.IAC + telnet.DONT + '\x42')
468 d = self.assertFailure(d, telnet.OptionRefused)
469 d.addCallback(lambda ignored: self._enabledHelper(self.p.protocol))
471 lambda ignored: self.assertEqual(s.us.negotiating, False))
475 def testAcceptedDisableRequest(self):
476 # Try to disable an option through the user-level API. This
477 # returns a Deferred that fires when negotiation about the option
478 # finishes. Make sure it fires, make sure state gets updated
479 # properly, make sure the result indicates the option was enabled.
480 s = self.p.getOptionState('\x42')
483 d = self.p.dont('\x42')
485 self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + '\x42')
487 self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')
489 d.addCallback(self.assertEqual, True)
490 d.addCallback(lambda _: self._enabledHelper(self.p.protocol,
494 def testNegotiationBlocksFurtherNegotiation(self):
495 # Try to disable an option, then immediately try to enable it, then
496 # immediately try to disable it. Ensure that the 2nd and 3rd calls
497 # fail quickly with the right exception.
498 s = self.p.getOptionState('\x24')
500 d2 = self.p.dont('\x24') # fires after the first line of _final
503 d = self.p.do('\x24')
504 return self.assertFailure(d, telnet.AlreadyNegotiating)
507 d = self.p.dont('\x24')
508 return self.assertFailure(d, telnet.AlreadyNegotiating)
511 self.p.dataReceived(telnet.IAC + telnet.WONT + '\x24')
512 # an assertion that only passes if d2 has fired
513 self._enabledHelper(self.p.protocol, dR=['\x24'])
514 # Make sure we allow this
515 self.p.protocol.remoteEnableable = ('\x24',)
516 d = self.p.do('\x24')
517 self.p.dataReceived(telnet.IAC + telnet.WILL + '\x24')
518 d.addCallback(self.assertEqual, True)
519 d.addCallback(lambda _: self._enabledHelper(self.p.protocol,
526 d.addCallback(_final)
529 def testSuperfluousDisableRequestRaises(self):
530 # Try to disable a disabled option. Make sure it fails properly.
531 d = self.p.dont('\xab')
532 return self.assertFailure(d, telnet.AlreadyDisabled)
534 def testSuperfluousEnableRequestRaises(self):
535 # Try to disable a disabled option. Make sure it fails properly.
536 s = self.p.getOptionState('\xab')
538 d = self.p.do('\xab')
539 return self.assertFailure(d, telnet.AlreadyEnabled)
541 def testLostConnectionFailsDeferreds(self):
542 d1 = self.p.do('\x12')
543 d2 = self.p.do('\x23')
544 d3 = self.p.do('\x34')
546 class TestException(Exception):
549 self.p.connectionLost(TestException("Total failure!"))
551 d1 = self.assertFailure(d1, TestException)
552 d2 = self.assertFailure(d2, TestException)
553 d3 = self.assertFailure(d3, TestException)
554 return defer.gatherResults([d1, d2, d3])
557 class TestTelnet(telnet.Telnet):
559 A trivial extension of the telnet protocol class useful to unit tests.
562 telnet.Telnet.__init__(self)
566 def applicationDataReceived(self, bytes):
568 Record the given data in C{self.events}.
570 self.events.append(('bytes', bytes))
573 def unhandledCommand(self, command, bytes):
575 Record the given command in C{self.events}.
577 self.events.append(('command', command, bytes))
580 def unhandledSubnegotiation(self, command, bytes):
582 Record the given subnegotiation command in C{self.events}.
584 self.events.append(('negotiate', command, bytes))
588 class TelnetTests(unittest.TestCase):
590 Tests for L{telnet.Telnet}.
592 L{telnet.Telnet} implements the TELNET protocol (RFC 854), including option
593 and suboption negotiation, and option state tracking.
597 Create an unconnected L{telnet.Telnet} to be used by tests.
599 self.protocol = TestTelnet()
602 def test_enableLocal(self):
604 L{telnet.Telnet.enableLocal} should reject all options, since
605 L{telnet.Telnet} does not know how to implement any options.
607 self.assertFalse(self.protocol.enableLocal('\0'))
610 def test_enableRemote(self):
612 L{telnet.Telnet.enableRemote} should reject all options, since
613 L{telnet.Telnet} does not know how to implement any options.
615 self.assertFalse(self.protocol.enableRemote('\0'))
618 def test_disableLocal(self):
620 It is an error for L{telnet.Telnet.disableLocal} to be called, since
621 L{telnet.Telnet.enableLocal} will never allow any options to be enabled
622 locally. If a subclass overrides enableLocal, it must also override
625 self.assertRaises(NotImplementedError, self.protocol.disableLocal, '\0')
628 def test_disableRemote(self):
630 It is an error for L{telnet.Telnet.disableRemote} to be called, since
631 L{telnet.Telnet.enableRemote} will never allow any options to be
632 enabled remotely. If a subclass overrides enableRemote, it must also
633 override disableRemote.
635 self.assertRaises(NotImplementedError, self.protocol.disableRemote, '\0')
638 def test_requestNegotiation(self):
640 L{telnet.Telnet.requestNegotiation} formats the feature byte and the
641 payload bytes into the subnegotiation format and sends them.
645 transport = proto_helpers.StringTransport()
646 self.protocol.makeConnection(transport)
647 self.protocol.requestNegotiation('\x01', '\x02\x03')
650 # IAC SB feature bytes IAC SE
651 '\xff\xfa\x01\x02\x03\xff\xf0')
654 def test_requestNegotiationEscapesIAC(self):
656 If the payload for a subnegotiation includes I{IAC}, it is escaped by
657 L{telnet.Telnet.requestNegotiation} with another I{IAC}.
661 transport = proto_helpers.StringTransport()
662 self.protocol.makeConnection(transport)
663 self.protocol.requestNegotiation('\x01', '\xff')
666 '\xff\xfa\x01\xff\xff\xff\xf0')
669 def _deliver(self, bytes, *expected):
671 Pass the given bytes to the protocol's C{dataReceived} method and
672 assert that the given events occur.
674 received = self.protocol.events = []
675 self.protocol.dataReceived(bytes)
676 self.assertEqual(received, list(expected))
679 def test_oneApplicationDataByte(self):
681 One application-data byte in the default state gets delivered right
684 self._deliver('a', ('bytes', 'a'))
687 def test_twoApplicationDataBytes(self):
689 Two application-data bytes in the default state get delivered
692 self._deliver('bc', ('bytes', 'bc'))
695 def test_threeApplicationDataBytes(self):
697 Three application-data bytes followed by a control byte get
698 delivered, but the control byte doesn't.
700 self._deliver('def' + telnet.IAC, ('bytes', 'def'))
703 def test_escapedControl(self):
705 IAC in the escaped state gets delivered and so does another
706 application-data byte following it.
708 self._deliver(telnet.IAC)
709 self._deliver(telnet.IAC + 'g', ('bytes', telnet.IAC + 'g'))
712 def test_carriageReturn(self):
714 A carriage return only puts the protocol into the newline state. A
715 linefeed in the newline state causes just the newline to be
716 delivered. A nul in the newline state causes a carriage return to
717 be delivered. An IAC in the newline state causes a carriage return
718 to be delivered and puts the protocol into the escaped state.
719 Anything else causes a carriage return and that thing to be
723 self._deliver('\n', ('bytes', '\n'))
724 self._deliver('\r\n', ('bytes', '\n'))
727 self._deliver('\0', ('bytes', '\r'))
728 self._deliver('\r\0', ('bytes', '\r'))
731 self._deliver('a', ('bytes', '\ra'))
732 self._deliver('\ra', ('bytes', '\ra'))
736 telnet.IAC + telnet.IAC + 'x', ('bytes', '\r' + telnet.IAC + 'x'))
739 def test_applicationDataBeforeSimpleCommand(self):
741 Application bytes received before a command are delivered before the
742 command is processed.
745 'x' + telnet.IAC + telnet.NOP,
746 ('bytes', 'x'), ('command', telnet.NOP, None))
749 def test_applicationDataBeforeCommand(self):
751 Application bytes received before a WILL/WONT/DO/DONT are delivered
752 before the command is processed.
754 self.protocol.commandMap = {}
756 'y' + telnet.IAC + telnet.WILL + '\x00',
757 ('bytes', 'y'), ('command', telnet.WILL, '\x00'))
760 def test_applicationDataBeforeSubnegotiation(self):
762 Application bytes received before a subnegotiation command are
763 delivered before the negotiation is processed.
766 'z' + telnet.IAC + telnet.SB + 'Qx' + telnet.IAC + telnet.SE,
767 ('bytes', 'z'), ('negotiate', 'Q', ['x']))