diff options
Diffstat (limited to 'Lib')
| -rw-r--r-- | Lib/telnetlib.py | 146 | ||||
| -rw-r--r-- | Lib/test/test_telnetlib.py | 297 |
2 files changed, 367 insertions, 76 deletions
diff --git a/Lib/telnetlib.py b/Lib/telnetlib.py index 14d05571..11e1a65 100644 --- a/Lib/telnetlib.py +++ b/Lib/telnetlib.py @@ -47,87 +47,87 @@ DEBUGLEVEL = 0 TELNET_PORT = 23 # Telnet protocol characters (don't change) -IAC = 255 # "Interpret As Command" -DONT = 254 -DO = 253 -WONT = 252 -WILL = 251 -theNULL = 0 - -SE = 240 # Subnegotiation End -NOP = 241 # No Operation -DM = 242 # Data Mark -BRK = 243 # Break -IP = 244 # Interrupt process -AO = 245 # Abort output -AYT = 246 # Are You There -EC = 247 # Erase Character -EL = 248 # Erase Line -GA = 249 # Go Ahead -SB = 250 # Subnegotiation Begin +IAC = bytes([255]) # "Interpret As Command" +DONT = bytes([254]) +DO = bytes([253]) +WONT = bytes([252]) +WILL = bytes([251]) +theNULL = bytes([0]) + +SE = bytes([240]) # Subnegotiation End +NOP = bytes([241]) # No Operation +DM = bytes([242]) # Data Mark +BRK = bytes([243]) # Break +IP = bytes([244]) # Interrupt process +AO = bytes([245]) # Abort output +AYT = bytes([246]) # Are You There +EC = bytes([247]) # Erase Character +EL = bytes([248]) # Erase Line +GA = bytes([249]) # Go Ahead +SB = bytes([250]) # Subnegotiation Begin # Telnet protocol options code (don't change) # These ones all come from arpa/telnet.h -BINARY = 0 # 8-bit data path -ECHO = 1 # echo -RCP = 2 # prepare to reconnect -SGA = 3 # suppress go ahead -NAMS = 4 # approximate message size -STATUS = 5 # give status -TM = 6 # timing mark -RCTE = 7 # remote controlled transmission and echo -NAOL = 8 # negotiate about output line width -NAOP = 9 # negotiate about output page size -NAOCRD = 10 # negotiate about CR disposition -NAOHTS = 11 # negotiate about horizontal tabstops -NAOHTD = 12 # negotiate about horizontal tab disposition -NAOFFD = 13 # negotiate about formfeed disposition -NAOVTS = 14 # negotiate about vertical tab stops -NAOVTD = 15 # negotiate about vertical tab disposition -NAOLFD = 16 # negotiate about output LF disposition -XASCII = 17 # extended ascii character set -LOGOUT = 18 # force logout -BM = 19 # byte macro -DET = 20 # data entry terminal -SUPDUP = 21 # supdup protocol -SUPDUPOUTPUT = 22 # supdup output -SNDLOC = 23 # send location -TTYPE = 24 # terminal type -EOR = 25 # end or record -TUID = 26 # TACACS user identification -OUTMRK = 27 # output marking -TTYLOC = 28 # terminal location number -VT3270REGIME = 29 # 3270 regime -X3PAD = 30 # X.3 PAD -NAWS = 31 # window size -TSPEED = 32 # terminal speed -LFLOW = 33 # remote flow control -LINEMODE = 34 # Linemode option -XDISPLOC = 35 # X Display Location -OLD_ENVIRON = 36 # Old - Environment variables -AUTHENTICATION = 37 # Authenticate -ENCRYPT = 38 # Encryption option -NEW_ENVIRON = 39 # New - Environment variables +BINARY = bytes([0]) # 8-bit data path +ECHO = bytes([1]) # echo +RCP = bytes([2]) # prepare to reconnect +SGA = bytes([3]) # suppress go ahead +NAMS = bytes([4]) # approximate message size +STATUS = bytes([5]) # give status +TM = bytes([6]) # timing mark +RCTE = bytes([7]) # remote controlled transmission and echo +NAOL = bytes([8]) # negotiate about output line width +NAOP = bytes([9]) # negotiate about output page size +NAOCRD = bytes([10]) # negotiate about CR disposition +NAOHTS = bytes([11]) # negotiate about horizontal tabstops +NAOHTD = bytes([12]) # negotiate about horizontal tab disposition +NAOFFD = bytes([13]) # negotiate about formfeed disposition +NAOVTS = bytes([14]) # negotiate about vertical tab stops +NAOVTD = bytes([15]) # negotiate about vertical tab disposition +NAOLFD = bytes([16]) # negotiate about output LF disposition +XASCII = bytes([17]) # extended ascii character set +LOGOUT = bytes([18]) # force logout +BM = bytes([19]) # byte macro +DET = bytes([20]) # data entry terminal +SUPDUP = bytes([21]) # supdup protocol +SUPDUPOUTPUT = bytes([22]) # supdup output +SNDLOC = bytes([23]) # send location +TTYPE = bytes([24]) # terminal type +EOR = bytes([25]) # end or record +TUID = bytes([26]) # TACACS user identification +OUTMRK = bytes([27]) # output marking +TTYLOC = bytes([28]) # terminal location number +VT3270REGIME = bytes([29]) # 3270 regime +X3PAD = bytes([30]) # X.3 PAD +NAWS = bytes([31]) # window size +TSPEED = bytes([32]) # terminal speed +LFLOW = bytes([33]) # remote flow control +LINEMODE = bytes([34]) # Linemode option +XDISPLOC = bytes([35]) # X Display Location +OLD_ENVIRON = bytes([36]) # Old - Environment variables +AUTHENTICATION = bytes([37]) # Authenticate +ENCRYPT = bytes([38]) # Encryption option +NEW_ENVIRON = bytes([39]) # New - Environment variables # the following ones come from # http://www.iana.org/assignments/telnet-options # Unfortunately, that document does not assign identifiers # to all of them, so we are making them up -TN3270E = 40 # TN3270E -XAUTH = 41 # XAUTH -CHARSET = 42 # CHARSET -RSP = 43 # Telnet Remote Serial Port -COM_PORT_OPTION = 44 # Com Port Control Option -SUPPRESS_LOCAL_ECHO = 45 # Telnet Suppress Local Echo -TLS = 46 # Telnet Start TLS -KERMIT = 47 # KERMIT -SEND_URL = 48 # SEND-URL -FORWARD_X = 49 # FORWARD_X -PRAGMA_LOGON = 138 # TELOPT PRAGMA LOGON -SSPI_LOGON = 139 # TELOPT SSPI LOGON -PRAGMA_HEARTBEAT = 140 # TELOPT PRAGMA HEARTBEAT -EXOPL = 255 # Extended-Options-List -NOOPT = 0 +TN3270E = bytes([40]) # TN3270E +XAUTH = bytes([41]) # XAUTH +CHARSET = bytes([42]) # CHARSET +RSP = bytes([43]) # Telnet Remote Serial Port +COM_PORT_OPTION = bytes([44]) # Com Port Control Option +SUPPRESS_LOCAL_ECHO = bytes([45]) # Telnet Suppress Local Echo +TLS = bytes([46]) # Telnet Start TLS +KERMIT = bytes([47]) # KERMIT +SEND_URL = bytes([48]) # SEND-URL +FORWARD_X = bytes([49]) # FORWARD_X +PRAGMA_LOGON = bytes([138]) # TELOPT PRAGMA LOGON +SSPI_LOGON = bytes([139]) # TELOPT SSPI LOGON +PRAGMA_HEARTBEAT = bytes([140]) # TELOPT PRAGMA HEARTBEAT +EXOPL = bytes([255]) # Extended-Options-List +NOOPT = bytes([0]) class Telnet: diff --git a/Lib/test/test_telnetlib.py b/Lib/test/test_telnetlib.py index 4ae6222..bad2819 100644 --- a/Lib/test/test_telnetlib.py +++ b/Lib/test/test_telnetlib.py @@ -2,17 +2,38 @@ import socket import threading import telnetlib import time +import queue from unittest import TestCase from test import support HOST = support.HOST +EOF_sigil = object() -def server(evt, serv): +def server(evt, serv, dataq=None): + """ Open a tcp server in three steps + 1) set evt to true to let the parent know we are ready + 2) [optional] if is not False, write the list of data from dataq.get() + to the socket. + 3) set evt to true to let the parent know we're done + """ serv.listen(5) evt.set() try: conn, addr = serv.accept() + if dataq: + data = b'' + new_data = dataq.get(True, 0.5) + dataq.task_done() + for item in new_data: + if item == EOF_sigil: + break + if type(item) in [int, float]: + time.sleep(item) + else: + data += item + written = conn.send(data) + data = data[written:] except socket.timeout: pass finally: @@ -26,13 +47,15 @@ class GeneralTests(TestCase): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(3) self.port = support.bind_port(self.sock) - threading.Thread(target=server, args=(self.evt,self.sock)).start() + self.thread = threading.Thread(target=server, args=(self.evt,self.sock)) + self.thread.start() self.evt.wait() self.evt.clear() time.sleep(.1) def tearDown(self): self.evt.wait() + self.thread.join() def testBasic(self): # connects @@ -71,9 +94,277 @@ class GeneralTests(TestCase): self.assertEqual(telnet.sock.gettimeout(), 30) telnet.sock.close() +def _read_setUp(self): + self.evt = threading.Event() + self.dataq = queue.Queue() + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.settimeout(3) + self.port = support.bind_port(self.sock) + self.thread = threading.Thread(target=server, args=(self.evt,self.sock, self.dataq)) + self.thread.start() + self.evt.wait() + self.evt.clear() + time.sleep(.1) + +def _read_tearDown(self): + self.evt.wait() + self.thread.join() + +class ReadTests(TestCase): + setUp = _read_setUp + tearDown = _read_tearDown + + # use a similar approach to testing timeouts as test_timeout.py + # these will never pass 100% but make the fuzz big enough that it is rare + block_long = 0.6 + block_short = 0.3 + def test_read_until_A(self): + """ + read_until(expected, [timeout]) + Read until the expected string has been seen, or a timeout is + hit (default is no timeout); may block. + """ + want = [b'x' * 10, b'match', b'y' * 10, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + data = telnet.read_until(b'match') + self.assertEqual(data, b''.join(want[:-2])) + + def test_read_until_B(self): + # test the timeout - it does NOT raise socket.timeout + want = [b'hello', self.block_long, b'not seen', EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + data = telnet.read_until(b'not seen', self.block_short) + self.assertEqual(data, want[0]) + self.assertEqual(telnet.read_all(), b'not seen') + + def test_read_all_A(self): + """ + read_all() + Read all data until EOF; may block. + """ + want = [b'x' * 500, b'y' * 500, b'z' * 500, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + data = telnet.read_all() + self.assertEqual(data, b''.join(want[:-1])) + return + + def _test_blocking(self, func): + self.dataq.put([self.block_long, EOF_sigil]) + self.dataq.join() + start = time.time() + data = func() + self.assertTrue(self.block_short <= time.time() - start) + + def test_read_all_B(self): + self._test_blocking(telnetlib.Telnet(HOST, self.port).read_all) + + def test_read_all_C(self): + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + telnet.read_all() + telnet.read_all() # shouldn't raise + + def test_read_some_A(self): + """ + read_some() + Read at least one byte or EOF; may block. + """ + # test 'at least one byte' + want = [b'x' * 500, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + data = telnet.read_all() + self.assertTrue(len(data) >= 1) + + def test_read_some_B(self): + # test EOF + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + self.assertEqual(b'', telnet.read_some()) + + def test_read_some_C(self): + self._test_blocking(telnetlib.Telnet(HOST, self.port).read_some) + + def _test_read_any_eager_A(self, func_name): + """ + read_very_eager() + Read all data available already queued or on the socket, + without blocking. + """ + want = [self.block_long, b'x' * 100, b'y' * 100, EOF_sigil] + expects = want[1] + want[2] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + func = getattr(telnet, func_name) + data = b'' + while True: + try: + data += func() + self.assertTrue(expects.startswith(data)) + except EOFError: + break + self.assertEqual(expects, data) + + def _test_read_any_eager_B(self, func_name): + # test EOF + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + time.sleep(self.block_short) + func = getattr(telnet, func_name) + self.assertRaises(EOFError, func) + + # read_eager and read_very_eager make the same gaurantees + # (they behave differently but we only test the gaurantees) + def test_read_very_eager_A(self): + self._test_read_any_eager_A('read_very_eager') + def test_read_very_eager_B(self): + self._test_read_any_eager_B('read_very_eager') + def test_read_eager_A(self): + self._test_read_any_eager_A('read_eager') + def test_read_eager_B(self): + self._test_read_any_eager_B('read_eager') + # NB -- we need to test the IAC block which is mentioned in the docstring + # but not in the module docs + + def _test_read_any_lazy_B(self, func_name): + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + func = getattr(telnet, func_name) + telnet.fill_rawq() + self.assertRaises(EOFError, func) + + def test_read_lazy_A(self): + want = [b'x' * 100, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + time.sleep(self.block_short) + self.assertEqual(b'', telnet.read_lazy()) + data = b'' + while True: + try: + read_data = telnet.read_lazy() + data += read_data + if not read_data: + telnet.fill_rawq() + except EOFError: + break + self.assertTrue(want[0].startswith(data)) + self.assertEqual(data, want[0]) + + def test_read_lazy_B(self): + self._test_read_any_lazy_B('read_lazy') + + def test_read_very_lazy_A(self): + want = [b'x' * 100, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + time.sleep(self.block_short) + self.assertEqual(b'', telnet.read_very_lazy()) + data = b'' + while True: + try: + read_data = telnet.read_very_lazy() + except EOFError: + break + data += read_data + if not read_data: + telnet.fill_rawq() + self.assertEqual(b'', telnet.cookedq) + telnet.process_rawq() + self.assertTrue(want[0].startswith(data)) + self.assertEqual(data, want[0]) + + def test_read_very_lazy_B(self): + self._test_read_any_lazy_B('read_very_lazy') + +class nego_collector(object): + def __init__(self, sb_getter=None): + self.seen = b'' + self.sb_getter = sb_getter + self.sb_seen = b'' + + def do_nego(self, sock, cmd, opt): + self.seen += cmd + opt + if cmd == tl.SE and self.sb_getter: + sb_data = self.sb_getter() + self.sb_seen += sb_data + +tl = telnetlib +class OptionTests(TestCase): + setUp = _read_setUp + tearDown = _read_tearDown + # RFC 854 commands + cmds = [tl.AO, tl.AYT, tl.BRK, tl.EC, tl.EL, tl.GA, tl.IP, tl.NOP] + + def _test_command(self, data): + """ helper for testing IAC + cmd """ + self.setUp() + self.dataq.put(data) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + nego = nego_collector() + telnet.set_option_negotiation_callback(nego.do_nego) + txt = telnet.read_all() + cmd = nego.seen + self.assertTrue(len(cmd) > 0) # we expect at least one command + self.assertTrue(cmd[:1] in self.cmds) + self.assertEqual(cmd[1:2], tl.NOOPT) + self.assertEqual(len(b''.join(data[:-1])), len(txt + cmd)) + nego.sb_getter = None # break the nego => telnet cycle + self.tearDown() + + def test_IAC_commands(self): + # reset our setup + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + self.tearDown() + + for cmd in self.cmds: + self._test_command([tl.IAC, cmd, EOF_sigil]) + self._test_command([b'x' * 100, tl.IAC, cmd, b'y'*100, EOF_sigil]) + self._test_command([b'x' * 10, tl.IAC, cmd, b'y'*10, EOF_sigil]) + # all at once + self._test_command([tl.IAC + cmd for (cmd) in self.cmds] + [EOF_sigil]) + self.assertEqual(b'', telnet.read_sb_data()) + + def test_SB_commands(self): + # RFC 855, subnegotiations portion + send = [tl.IAC + tl.SB + tl.IAC + tl.SE, + tl.IAC + tl.SB + tl.IAC + tl.IAC + tl.IAC + tl.SE, + tl.IAC + tl.SB + tl.IAC + tl.IAC + b'aa' + tl.IAC + tl.SE, + tl.IAC + tl.SB + b'bb' + tl.IAC + tl.IAC + tl.IAC + tl.SE, + tl.IAC + tl.SB + b'cc' + tl.IAC + tl.IAC + b'dd' + tl.IAC + tl.SE, + EOF_sigil, + ] + self.dataq.put(send) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + nego = nego_collector(telnet.read_sb_data) + telnet.set_option_negotiation_callback(nego.do_nego) + txt = telnet.read_all() + self.assertEqual(txt, b'') + want_sb_data = tl.IAC + tl.IAC + b'aabb' + tl.IAC + b'cc' + tl.IAC + b'dd' + self.assertEqual(nego.sb_seen, want_sb_data) + self.assertEqual(b'', telnet.read_sb_data()) + nego.sb_getter = None # break the nego => telnet cycle def test_main(verbose=None): - support.run_unittest(GeneralTests) + support.run_unittest(GeneralTests, ReadTests, OptionTests) if __name__ == '__main__': test_main() |