Viewing file: test_telnet.py (23.66 KB) -rw-r--r-- Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) |
# -*- test-case-name: twisted.conch.test.test_telnet -*- # Copyright (c) 2001-2009 Twisted Matrix Laboratories. # See LICENSE for details.
""" Tests for L{twisted.conch.telnet}. """
from zope.interface import implements
from twisted.internet import defer
from twisted.conch import telnet
from twisted.trial import unittest from twisted.test import proto_helpers
class TestProtocol: implements(telnet.ITelnetProtocol)
localEnableable = () remoteEnableable = ()
def __init__(self): self.bytes = '' self.subcmd = '' self.calls = []
self.enabledLocal = [] self.enabledRemote = [] self.disabledLocal = [] self.disabledRemote = []
def makeConnection(self, transport): d = transport.negotiationMap = {} d['\x12'] = self.neg_TEST_COMMAND
d = transport.commandMap = transport.commandMap.copy() for cmd in ('NOP', 'DM', 'BRK', 'IP', 'AO', 'AYT', 'EC', 'EL', 'GA'): d[getattr(telnet, cmd)] = lambda arg, cmd=cmd: self.calls.append(cmd)
def dataReceived(self, bytes): self.bytes += bytes
def connectionLost(self, reason): pass
def neg_TEST_COMMAND(self, payload): self.subcmd = payload
def enableLocal(self, option): if option in self.localEnableable: self.enabledLocal.append(option) return True return False
def disableLocal(self, option): self.disabledLocal.append(option)
def enableRemote(self, option): if option in self.remoteEnableable: self.enabledRemote.append(option) return True return False
def disableRemote(self, option): self.disabledRemote.append(option)
class TelnetTransportTestCase(unittest.TestCase): """ Tests for L{telnet.TelnetTransport}. """ def setUp(self): self.p = telnet.TelnetTransport(TestProtocol) self.t = proto_helpers.StringTransport() self.p.makeConnection(self.t)
def testRegularBytes(self): # Just send a bunch of bytes. None of these do anything # with telnet. They should pass right through to the # application layer. h = self.p.protocol
L = ["here are some bytes la la la", "some more arrive here", "lots of bytes to play with", "la la la", "ta de da", "dum"] for b in L: self.p.dataReceived(b)
self.assertEquals(h.bytes, ''.join(L))
def testNewlineHandling(self): # Send various kinds of newlines and make sure they get translated # into \n. h = self.p.protocol
L = ["here is the first line\r\n", "here is the second line\r\0", "here is the third line\r\n", "here is the last line\r\0"]
for b in L: self.p.dataReceived(b)
self.assertEquals(h.bytes, L[0][:-2] + '\n' + L[1][:-2] + '\r' + L[2][:-2] + '\n' + L[3][:-2] + '\r')
def testIACEscape(self): # Send a bunch of bytes and a couple quoted \xFFs. Unquoted, # \xFF is a telnet command. Quoted, one of them from each pair # should be passed through to the application layer. h = self.p.protocol
L = ["here are some bytes\xff\xff with an embedded IAC", "and here is a test of a border escape\xff", "\xff did you get that IAC?"]
for b in L: self.p.dataReceived(b)
self.assertEquals(h.bytes, ''.join(L).replace('\xff\xff', '\xff'))
def _simpleCommandTest(self, cmdName): # Send a single simple telnet command and make sure # it gets noticed and the appropriate method gets # called. h = self.p.protocol
cmd = telnet.IAC + getattr(telnet, cmdName) L = ["Here's some bytes, tra la la", "But ono!" + cmd + " an interrupt"]
for b in L: self.p.dataReceived(b)
self.assertEquals(h.calls, [cmdName]) self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))
def testInterrupt(self): self._simpleCommandTest("IP")
def testNoOperation(self): self._simpleCommandTest("NOP")
def testDataMark(self): self._simpleCommandTest("DM")
def testBreak(self): self._simpleCommandTest("BRK")
def testAbortOutput(self): self._simpleCommandTest("AO")
def testAreYouThere(self): self._simpleCommandTest("AYT")
def testEraseCharacter(self): self._simpleCommandTest("EC")
def testEraseLine(self): self._simpleCommandTest("EL")
def testGoAhead(self): self._simpleCommandTest("GA")
def testSubnegotiation(self): # Send a subnegotiation command and make sure it gets # parsed and that the correct method is called. h = self.p.protocol
cmd = telnet.IAC + telnet.SB + '\x12hello world' + telnet.IAC + telnet.SE L = ["These are some bytes but soon" + cmd, "there will be some more"]
for b in L: self.p.dataReceived(b)
self.assertEquals(h.bytes, ''.join(L).replace(cmd, '')) self.assertEquals(h.subcmd, list("hello world"))
def testSubnegotiationWithEmbeddedSE(self): # Send a subnegotiation command with an embedded SE. Make sure # that SE gets passed to the correct method. h = self.p.protocol
cmd = (telnet.IAC + telnet.SB + '\x12' + telnet.SE + telnet.IAC + telnet.SE)
L = ["Some bytes are here" + cmd + "and here", "and here"]
for b in L: self.p.dataReceived(b)
self.assertEquals(h.bytes, ''.join(L).replace(cmd, '')) self.assertEquals(h.subcmd, [telnet.SE])
def testBoundarySubnegotiation(self): # Send a subnegotiation command. Split it at every possible byte boundary # and make sure it always gets parsed and that it is passed to the correct # method. cmd = (telnet.IAC + telnet.SB + '\x12' + telnet.SE + 'hello' + telnet.IAC + telnet.SE)
for i in range(len(cmd)): h = self.p.protocol = TestProtocol() h.makeConnection(self.p)
a, b = cmd[:i], cmd[i:] L = ["first part" + a, b + "last part"]
for bytes in L: self.p.dataReceived(bytes)
self.assertEquals(h.bytes, ''.join(L).replace(cmd, '')) self.assertEquals(h.subcmd, [telnet.SE] + list('hello'))
def _enabledHelper(self, o, eL=[], eR=[], dL=[], dR=[]): self.assertEquals(o.enabledLocal, eL) self.assertEquals(o.enabledRemote, eR) self.assertEquals(o.disabledLocal, dL) self.assertEquals(o.disabledRemote, dR)
def testRefuseWill(self): # Try to enable an option. The server should refuse to enable it. cmd = telnet.IAC + telnet.WILL + '\x12'
bytes = "surrounding bytes" + cmd + "to spice things up" self.p.dataReceived(bytes)
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, '')) self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x12') self._enabledHelper(self.p.protocol)
def testRefuseDo(self): # Try to enable an option. The server should refuse to enable it. cmd = telnet.IAC + telnet.DO + '\x12'
bytes = "surrounding bytes" + cmd + "to spice things up" self.p.dataReceived(bytes)
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, '')) self.assertEquals(self.t.value(), telnet.IAC + telnet.WONT + '\x12') self._enabledHelper(self.p.protocol)
def testAcceptDo(self): # Try to enable an option. The option is in our allowEnable # list, so we will allow it to be enabled. cmd = telnet.IAC + telnet.DO + '\x19' bytes = 'padding' + cmd + 'trailer'
h = self.p.protocol h.localEnableable = ('\x19',) self.p.dataReceived(bytes)
self.assertEquals(self.t.value(), telnet.IAC + telnet.WILL + '\x19') self._enabledHelper(h, eL=['\x19'])
def testAcceptWill(self): # Same as testAcceptDo, but reversed. cmd = telnet.IAC + telnet.WILL + '\x91' bytes = 'header' + cmd + 'padding'
h = self.p.protocol h.remoteEnableable = ('\x91',) self.p.dataReceived(bytes)
self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x91') self._enabledHelper(h, eR=['\x91'])
def testAcceptWont(self): # Try to disable an option. The server must allow any option to # be disabled at any time. Make sure it disables it and sends # back an acknowledgement of this. cmd = telnet.IAC + telnet.WONT + '\x29'
# Jimmy it - after these two lines, the server will be in a state # such that it believes the option to have been previously enabled # via normal negotiation. s = self.p.getOptionState('\x29') s.him.state = 'yes'
bytes = "fiddle dee" + cmd self.p.dataReceived(bytes)
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, '')) self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x29') self.assertEquals(s.him.state, 'no') self._enabledHelper(self.p.protocol, dR=['\x29'])
def testAcceptDont(self): # Try to disable an option. The server must allow any option to # be disabled at any time. Make sure it disables it and sends # back an acknowledgement of this. cmd = telnet.IAC + telnet.DONT + '\x29'
# Jimmy it - after these two lines, the server will be in a state # such that it believes the option to have beenp previously enabled # via normal negotiation. s = self.p.getOptionState('\x29') s.us.state = 'yes'
bytes = "fiddle dum " + cmd self.p.dataReceived(bytes)
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, '')) self.assertEquals(self.t.value(), telnet.IAC + telnet.WONT + '\x29') self.assertEquals(s.us.state, 'no') self._enabledHelper(self.p.protocol, dL=['\x29'])
def testIgnoreWont(self): # Try to disable an option. The option is already disabled. The # server should send nothing in response to this. cmd = telnet.IAC + telnet.WONT + '\x47'
bytes = "dum de dum" + cmd + "tra la la" self.p.dataReceived(bytes)
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, '')) self.assertEquals(self.t.value(), '') self._enabledHelper(self.p.protocol)
def testIgnoreDont(self): # Try to disable an option. The option is already disabled. The # server should send nothing in response to this. Doing so could # lead to a negotiation loop. cmd = telnet.IAC + telnet.DONT + '\x47'
bytes = "dum de dum" + cmd + "tra la la" self.p.dataReceived(bytes)
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, '')) self.assertEquals(self.t.value(), '') self._enabledHelper(self.p.protocol)
def testIgnoreWill(self): # Try to enable an option. The option is already enabled. The # server should send nothing in response to this. Doing so could # lead to a negotiation loop. cmd = telnet.IAC + telnet.WILL + '\x56'
# Jimmy it - after these two lines, the server will be in a state # such that it believes the option to have been previously enabled # via normal negotiation. s = self.p.getOptionState('\x56') s.him.state = 'yes'
bytes = "tra la la" + cmd + "dum de dum" self.p.dataReceived(bytes)
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, '')) self.assertEquals(self.t.value(), '') self._enabledHelper(self.p.protocol)
def testIgnoreDo(self): # Try to enable an option. The option is already enabled. The # server should send nothing in response to this. Doing so could # lead to a negotiation loop. cmd = telnet.IAC + telnet.DO + '\x56'
# Jimmy it - after these two lines, the server will be in a state # such that it believes the option to have been previously enabled # via normal negotiation. s = self.p.getOptionState('\x56') s.us.state = 'yes'
bytes = "tra la la" + cmd + "dum de dum" self.p.dataReceived(bytes)
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, '')) self.assertEquals(self.t.value(), '') self._enabledHelper(self.p.protocol)
def testAcceptedEnableRequest(self): # Try to enable an option through the user-level API. This # returns a Deferred that fires when negotiation about the option # finishes. Make sure it fires, make sure state gets updated # properly, make sure the result indicates the option was enabled. d = self.p.do('\x42')
h = self.p.protocol h.remoteEnableable = ('\x42',)
self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x42')
self.p.dataReceived(telnet.IAC + telnet.WILL + '\x42')
d.addCallback(self.assertEquals, True) d.addCallback(lambda _: self._enabledHelper(h, eR=['\x42'])) return d
def testRefusedEnableRequest(self): # Try to enable an option through the user-level API. This # returns a Deferred that fires when negotiation about the option # finishes. Make sure it fires, make sure state gets updated # properly, make sure the result indicates the option was enabled. d = self.p.do('\x42')
self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x42')
self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')
d = self.assertFailure(d, telnet.OptionRefused) d.addCallback(lambda _: self._enabledHelper(self.p.protocol)) return d
def testAcceptedDisableRequest(self): # Try to disable an option through the user-level API. This # returns a Deferred that fires when negotiation about the option # finishes. Make sure it fires, make sure state gets updated # properly, make sure the result indicates the option was enabled. s = self.p.getOptionState('\x42') s.him.state = 'yes'
d = self.p.dont('\x42')
self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x42')
self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')
d.addCallback(self.assertEquals, True) d.addCallback(lambda _: self._enabledHelper(self.p.protocol, dR=['\x42'])) return d
def testNegotiationBlocksFurtherNegotiation(self): # Try to disable an option, then immediately try to enable it, then # immediately try to disable it. Ensure that the 2nd and 3rd calls # fail quickly with the right exception. s = self.p.getOptionState('\x24') s.him.state = 'yes' d2 = self.p.dont('\x24') # fires after the first line of _final
def _do(x): d = self.p.do('\x24') return self.assertFailure(d, telnet.AlreadyNegotiating)
def _dont(x): d = self.p.dont('\x24') return self.assertFailure(d, telnet.AlreadyNegotiating)
def _final(x): self.p.dataReceived(telnet.IAC + telnet.WONT + '\x24') # an assertion that only passes if d2 has fired self._enabledHelper(self.p.protocol, dR=['\x24']) # Make sure we allow this self.p.protocol.remoteEnableable = ('\x24',) d = self.p.do('\x24') self.p.dataReceived(telnet.IAC + telnet.WILL + '\x24') d.addCallback(self.assertEquals, True) d.addCallback(lambda _: self._enabledHelper(self.p.protocol, eR=['\x24'], dR=['\x24'])) return d
d = _do(None) d.addCallback(_dont) d.addCallback(_final) return d
def testSuperfluousDisableRequestRaises(self): # Try to disable a disabled option. Make sure it fails properly. d = self.p.dont('\xab') return self.assertFailure(d, telnet.AlreadyDisabled)
def testSuperfluousEnableRequestRaises(self): # Try to disable a disabled option. Make sure it fails properly. s = self.p.getOptionState('\xab') s.him.state = 'yes' d = self.p.do('\xab') return self.assertFailure(d, telnet.AlreadyEnabled)
def testLostConnectionFailsDeferreds(self): d1 = self.p.do('\x12') d2 = self.p.do('\x23') d3 = self.p.do('\x34')
class TestException(Exception): pass
self.p.connectionLost(TestException("Total failure!"))
d1 = self.assertFailure(d1, TestException) d2 = self.assertFailure(d2, TestException) d3 = self.assertFailure(d3, TestException) return defer.gatherResults([d1, d2, d3])
class TestTelnet(telnet.Telnet): """ A trivial extension of the telnet protocol class useful to unit tests. """ def __init__(self): telnet.Telnet.__init__(self) self.events = []
def applicationDataReceived(self, bytes): """ Record the given data in C{self.events}. """ self.events.append(('bytes', bytes))
def unhandledCommand(self, command, bytes): """ Record the given command in C{self.events}. """ self.events.append(('command', command, bytes))
def unhandledSubnegotiation(self, command, bytes): """ Record the given subnegotiation command in C{self.events}. """ self.events.append(('negotiate', command, bytes))
class TelnetTests(unittest.TestCase): """ Tests for L{telnet.Telnet}.
L{telnet.Telnet} implements the TELNET protocol (RFC 854), including option and suboption negotiation, and option state tracking. """ def setUp(self): """ Create an unconnected L{telnet.Telnet} to be used by tests. """ self.protocol = TestTelnet()
def test_enableLocal(self): """ L{telnet.Telnet.enableLocal} should reject all options, since L{telnet.Telnet} does not know how to implement any options. """ self.assertFalse(self.protocol.enableLocal('\0'))
def test_enableRemote(self): """ L{telnet.Telnet.enableRemote} should reject all options, since L{telnet.Telnet} does not know how to implement any options. """ self.assertFalse(self.protocol.enableRemote('\0'))
def test_disableLocal(self): """ It is an error for L{telnet.Telnet.disableLocal} to be called, since L{telnet.Telnet.enableLocal} will never allow any options to be enabled locally. If a subclass overrides enableLocal, it must also override disableLocal. """ self.assertRaises(NotImplementedError, self.protocol.disableLocal, '\0')
def test_disableRemote(self): """ It is an error for L{telnet.Telnet.disableRemote} to be called, since L{telnet.Telnet.enableRemote} will never allow any options to be enabled remotely. If a subclass overrides enableRemote, it must also override disableRemote. """ self.assertRaises(NotImplementedError, self.protocol.disableRemote, '\0')
def test_requestNegotiation(self): """ L{telnet.Telnet.requestNegotiation} formats the feature byte and the payload bytes into the subnegotiation format and sends them.
See RFC 855. """ transport = proto_helpers.StringTransport() self.protocol.makeConnection(transport) self.protocol.requestNegotiation('\x01', '\x02\x03') self.assertEqual( transport.value(), # IAC SB feature bytes IAC SE '\xff\xfa\x01\x02\x03\xff\xf0')
def test_requestNegotiationEscapesIAC(self): """ If the payload for a subnegotiation includes I{IAC}, it is escaped by L{telnet.Telnet.requestNegotiation} with another I{IAC}.
See RFC 855. """ transport = proto_helpers.StringTransport() self.protocol.makeConnection(transport) self.protocol.requestNegotiation('\x01', '\xff') self.assertEqual( transport.value(), '\xff\xfa\x01\xff\xff\xff\xf0')
def _deliver(self, bytes, *expected): """ Pass the given bytes to the protocol's C{dataReceived} method and assert that the given events occur. """ received = self.protocol.events = [] self.protocol.dataReceived(bytes) self.assertEqual(received, list(expected))
def test_oneApplicationDataByte(self): """ One application-data byte in the default state gets delivered right away. """ self._deliver('a', ('bytes', 'a'))
def test_twoApplicationDataBytes(self): """ Two application-data bytes in the default state get delivered together. """ self._deliver('bc', ('bytes', 'bc'))
def test_threeApplicationDataBytes(self): """ Three application-data bytes followed by a control byte get delivered, but the control byte doesn't. """ self._deliver('def' + telnet.IAC, ('bytes', 'def'))
def test_escapedControl(self): """ IAC in the escaped state gets delivered and so does another application-data byte following it. """ self._deliver(telnet.IAC) self._deliver(telnet.IAC + 'g', ('bytes', telnet.IAC + 'g'))
def test_carriageReturn(self): """ A carriage return only puts the protocol into the newline state. A linefeed in the newline state causes just the newline to be delivered. A nul in the newline state causes a carriage return to be delivered. An IAC in the newline state causes a carriage return to be delivered and puts the protocol into the escaped state. Anything else causes a carriage return and that thing to be delivered. """ self._deliver('\r') self._deliver('\n', ('bytes', '\n')) self._deliver('\r\n', ('bytes', '\n'))
self._deliver('\r') self._deliver('\0', ('bytes', '\r')) self._deliver('\r\0', ('bytes', '\r'))
self._deliver('\r') self._deliver('a', ('bytes', '\ra')) self._deliver('\ra', ('bytes', '\ra'))
self._deliver('\r') self._deliver( telnet.IAC + telnet.IAC + 'x', ('bytes', '\r' + telnet.IAC + 'x'))
def test_applicationDataBeforeSimpleCommand(self): """ Application bytes received before a command are delivered before the command is processed. """ self._deliver( 'x' + telnet.IAC + telnet.NOP, ('bytes', 'x'), ('command', telnet.NOP, None))
def test_applicationDataBeforeCommand(self): """ Application bytes received before a WILL/WONT/DO/DONT are delivered before the command is processed. """ self.protocol.commandMap = {} self._deliver( 'y' + telnet.IAC + telnet.WILL + '\x00', ('bytes', 'y'), ('command', telnet.WILL, '\x00'))
def test_applicationDataBeforeSubnegotiation(self): """ Application bytes received before a subnegotiation command are delivered before the negotiation is processed. """ self._deliver( 'z' + telnet.IAC + telnet.SB + 'Qx' + telnet.IAC + telnet.SE, ('bytes', 'z'), ('negotiate', 'Q', ['x']))
|