r46944 - port telnet
hawkowl-TA+aISz0psMTMxyoc4vAAJOcrHinNvQL0E9HWUfgJXw@public.gmane.org
| Newsgroups | gmane.comp.python.twisted.commits |
|---|---|
| Message-ID | <[email protected]> |
Author: hawkowl
Date: Mon Mar 7 22:34:45 2016
New Revision: 46944
Modified:
branches/telnet-py3-8228/twisted/conch/telnet.py
branches/telnet-py3-8228/twisted/conch/test/test_telnet.py
branches/telnet-py3-8228/twisted/protocols/telnet.py
branches/telnet-py3-8228/twisted/python/compat.py
branches/telnet-py3-8228/twisted/python/dist3.py
Log:
port telnet
Modified: branches/telnet-py3-8228/twisted/conch/telnet.py
==============================================================================
--- branches/telnet-py3-8228/twisted/conch/telnet.py (original)
+++ branches/telnet-py3-8228/twisted/conch/telnet.py Mon Mar 7 22:34:45 2016
@@ -8,12 +8,15 @@
@author: Jean-Paul Calderone
"""
+from __future__ import absolute_import, division
+
import struct
from zope.interface import implementer
from twisted.internet import protocol, interfaces as iinternet, defer
from twisted.python import log
+from twisted.python.compat import _bytesChr as chr, iterbytes
MODE = chr(1)
EDIT = 1
@@ -510,11 +513,11 @@
def dataReceived(self, data):
appDataBuffer = []
- for b in data:
+ for b in iterbytes(data):
if self.state == 'data':
if b == IAC:
self.state = 'escaped'
- elif b == '\r':
+ elif b == b'\r':
self.state = 'newline'
else:
appDataBuffer.append(b)
@@ -528,7 +531,7 @@
elif b in (NOP, DM, BRK, IP, AO, AYT, EC, EL, GA):
self.state = 'data'
if appDataBuffer:
- self.applicationDataReceived(''.join(appDataBuffer))
+ self.applicationDataReceived(b''.join(appDataBuffer))
del appDataBuffer[:]
self.commandReceived(b, None)
elif b in (WILL, WONT, DO, DONT):
@@ -541,19 +544,19 @@
command = self.command
del self.command
if appDataBuffer:
- self.applicationDataReceived(''.join(appDataBuffer))
+ self.applicationDataReceived(b''.join(appDataBuffer))
del appDataBuffer[:]
self.commandReceived(command, b)
elif self.state == 'newline':
self.state = 'data'
- if b == '\n':
- appDataBuffer.append('\n')
- elif b == '\0':
- appDataBuffer.append('\r')
+ if b == b'\n':
+ appDataBuffer.append(b'\n')
+ elif b == b'\0':
+ appDataBuffer.append(b'\r')
elif b == IAC:
# IAC isn't really allowed after \r, according to the
# RFC, but handling it this way is less surprising than
- # delivering the IAC to the app as application data.
+ # delivering the IAC to the app as application data.
# The purpose of the restriction is to allow terminals
# to unambiguously interpret the behavior of the CR
# after reading only one more byte. CR LF is supposed
@@ -561,10 +564,10 @@
# CR NUL another (cursor to first column). Absent the
# NUL, it still makes sense to interpret this as CR and
# then apply all the usual interpretation to the IAC.
- appDataBuffer.append('\r')
+ appDataBuffer.append(b'\r')
self.state = 'escaped'
else:
- appDataBuffer.append('\r' + b)
+ appDataBuffer.append(b'\r' + b)
elif self.state == 'subnegotiation':
if b == IAC:
self.state = 'subnegotiation-escaped'
@@ -576,7 +579,7 @@
commands = self.commands
del self.commands
if appDataBuffer:
- self.applicationDataReceived(''.join(appDataBuffer))
+ self.applicationDataReceived(b''.join(appDataBuffer))
del appDataBuffer[:]
self.negotiate(commands)
else:
@@ -586,7 +589,7 @@
raise ValueError("How'd you do this?")
if appDataBuffer:
- self.applicationDataReceived(''.join(appDataBuffer))
+ self.applicationDataReceived(b''.join(appDataBuffer))
def connectionLost(self, reason):
@@ -812,7 +815,7 @@
class ProtocolTransportMixin:
def write(self, bytes):
- self.transport.write(bytes.replace('\n', '\r\n'))
+ self.transport.write(bytes.replace(b'\n', b'\r\n'))
def writeSequence(self, seq):
self.transport.writeSequence(seq)
@@ -898,7 +901,7 @@
self.protocol.dataReceived(bytes)
def write(self, data):
- ProtocolTransportMixin.write(self, data.replace('\xff','\xff\xff'))
+ ProtocolTransportMixin.write(self, data.replace(b'\xff', b'\xff\xff'))
class TelnetBootstrapProtocol(TelnetProtocol, ProtocolTransportMixin):
@@ -964,7 +967,7 @@
# attribute of which will be an ITerminalProtocol. Maybe.
# You know what, XXX TODO clean this up.
if len(bytes) == 4:
- width, height = struct.unpack('!HH', ''.join(bytes))
+ width, height = struct.unpack('!HH', b''.join(bytes))
self.protocol.terminalProtocol.terminalSize(width, height)
else:
log.msg("Wrong number of NAWS bytes")
@@ -987,7 +990,7 @@
from twisted.protocols import basic
class StatefulTelnetProtocol(basic.LineReceiver, TelnetProtocol):
- delimiter = '\n'
+ delimiter = b'\n'
state = 'Discard'
@@ -1026,7 +1029,7 @@
self.portal = portal
def connectionMade(self):
- self.transport.write("Username: ")
+ self.transport.write(b"Username: ")
def connectionLost(self, reason):
StatefulTelnetProtocol.connectionLost(self, reason)
@@ -1040,7 +1043,7 @@
def telnet_User(self, line):
self.username = line
self.transport.will(ECHO)
- self.transport.write("Password: ")
+ self.transport.write(b"Password: ")
return 'Password'
def telnet_Password(self, line):
@@ -1065,8 +1068,8 @@
self.transport.protocol = protocol
def _ebLogin(self, failure):
- self.transport.write("\nAuthentication failed\n")
- self.transport.write("Username: ")
+ self.transport.write(b"\nAuthentication failed\n")
+ self.transport.write(b"Username: ")
self.state = "User"
__all__ = [
Modified: branches/telnet-py3-8228/twisted/conch/test/test_telnet.py
==============================================================================
--- branches/telnet-py3-8228/twisted/conch/test/test_telnet.py (original)
+++ branches/telnet-py3-8228/twisted/conch/test/test_telnet.py Mon Mar 7 22:34:45 2016
@@ -15,7 +15,7 @@
from twisted.trial import unittest
from twisted.test import proto_helpers
-
+from twisted.python.compat import iterbytes
@implementer(telnet.ITelnetProtocol)
@@ -24,8 +24,8 @@
remoteEnableable = ()
def __init__(self):
- self.bytes = ''
- self.subcmd = ''
+ self.bytes = b''
+ self.subcmd = []
self.calls = []
self.enabledLocal = []
@@ -35,7 +35,7 @@
def makeConnection(self, transport):
d = transport.negotiationMap = {}
- d['\x12'] = self.neg_TEST_COMMAND
+ d[b'\x12'] = self.neg_TEST_COMMAND
d = transport.commandMap = transport.commandMap.copy()
for cmd in ('NOP', 'DM', 'BRK', 'IP', 'AO', 'AYT', 'EC', 'EL', 'GA'):
@@ -95,34 +95,34 @@
# 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"]
+ L = [b"here are some bytes la la la",
+ b"some more arrive here",
+ b"lots of bytes to play with",
+ b"la la la",
+ b"ta de da",
+ b"dum"]
for b in L:
self.p.dataReceived(b)
- self.assertEqual(h.bytes, ''.join(L))
+ self.assertEqual(h.bytes, b''.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"]
+ L = [b"here is the first line\r\n",
+ b"here is the second line\r\0",
+ b"here is the third line\r\n",
+ b"here is the last line\r\0"]
for b in L:
self.p.dataReceived(b)
- self.assertEqual(h.bytes, L[0][:-2] + '\n' +
- L[1][:-2] + '\r' +
- L[2][:-2] + '\n' +
- L[3][:-2] + '\r')
+ self.assertEqual(h.bytes, L[0][:-2] + b'\n' +
+ L[1][:-2] + b'\r' +
+ L[2][:-2] + b'\n' +
+ L[3][:-2] + b'\r')
def testIACEscape(self):
# Send a bunch of bytes and a couple quoted \xFFs. Unquoted,
@@ -130,14 +130,14 @@
# 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?"]
+ L = [b"here are some bytes\xff\xff with an embedded IAC",
+ b"and here is a test of a border escape\xff",
+ b"\xff did you get that IAC?"]
for b in L:
self.p.dataReceived(b)
- self.assertEqual(h.bytes, ''.join(L).replace('\xff\xff', '\xff'))
+ self.assertEqual(h.bytes, b''.join(L).replace(b'\xff\xff', b'\xff'))
def _simpleCommandTest(self, cmdName):
# Send a single simple telnet command and make sure
@@ -146,14 +146,14 @@
h = self.p.protocol
cmd = telnet.IAC + getattr(telnet, cmdName)
- L = ["Here's some bytes, tra la la",
- "But ono!" + cmd + " an interrupt"]
+ L = [b"Here's some bytes, tra la la",
+ b"But ono!" + cmd + b" an interrupt"]
for b in L:
self.p.dataReceived(b)
self.assertEqual(h.calls, [cmdName])
- self.assertEqual(h.bytes, ''.join(L).replace(cmd, ''))
+ self.assertEqual(h.bytes, b''.join(L).replace(cmd, b''))
def testInterrupt(self):
self._simpleCommandTest("IP")
@@ -187,15 +187,15 @@
# 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"]
+ cmd = telnet.IAC + telnet.SB + b'\x12hello world' + telnet.IAC + telnet.SE
+ L = [b"These are some bytes but soon" + cmd,
+ b"there will be some more"]
for b in L:
self.p.dataReceived(b)
- self.assertEqual(h.bytes, ''.join(L).replace(cmd, ''))
- self.assertEqual(h.subcmd, list("hello world"))
+ self.assertEqual(h.bytes, b''.join(L).replace(cmd, b''))
+ self.assertEqual(h.subcmd, list(iterbytes(b"hello world")))
def testSubnegotiationWithEmbeddedSE(self):
# Send a subnegotiation command with an embedded SE. Make sure
@@ -203,16 +203,16 @@
h = self.p.protocol
cmd = (telnet.IAC + telnet.SB +
- '\x12' + telnet.SE +
+ b'\x12' + telnet.SE +
telnet.IAC + telnet.SE)
- L = ["Some bytes are here" + cmd + "and here",
- "and here"]
+ L = [b"Some bytes are here" + cmd + b"and here",
+ b"and here"]
for b in L:
self.p.dataReceived(b)
- self.assertEqual(h.bytes, ''.join(L).replace(cmd, ''))
+ self.assertEqual(h.bytes, b''.join(L).replace(cmd, b''))
self.assertEqual(h.subcmd, [telnet.SE])
def testBoundarySubnegotiation(self):
@@ -220,7 +220,7 @@
# 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' +
+ b'\x12' + telnet.SE + b'hello' +
telnet.IAC + telnet.SE)
for i in range(len(cmd)):
@@ -228,14 +228,14 @@
h.makeConnection(self.p)
a, b = cmd[:i], cmd[i:]
- L = ["first part" + a,
- b + "last part"]
+ L = [b"first part" + a,
+ b + b"last part"]
for bytes in L:
self.p.dataReceived(bytes)
- self.assertEqual(h.bytes, ''.join(L).replace(cmd, ''))
- self.assertEqual(h.subcmd, [telnet.SE] + list('hello'))
+ self.assertEqual(h.bytes, b''.join(L).replace(cmd, b''))
+ self.assertEqual(h.subcmd, [telnet.SE] + list(iterbytes(b'hello')))
def _enabledHelper(self, o, eL=[], eR=[], dL=[], dR=[]):
self.assertEqual(o.enabledLocal, eL)
@@ -245,152 +245,152 @@
def testRefuseWill(self):
# Try to enable an option. The server should refuse to enable it.
- cmd = telnet.IAC + telnet.WILL + '\x12'
+ cmd = telnet.IAC + telnet.WILL + b'\x12'
- bytes = "surrounding bytes" + cmd + "to spice things up"
+ bytes = b"surrounding bytes" + cmd + b"to spice things up"
self.p.dataReceived(bytes)
- self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
- self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + '\x12')
+ self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, b''))
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + b'\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'
+ cmd = telnet.IAC + telnet.DO + b'\x12'
- bytes = "surrounding bytes" + cmd + "to spice things up"
+ bytes = b"surrounding bytes" + cmd + b"to spice things up"
self.p.dataReceived(bytes)
- self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
- self.assertEqual(self.t.value(), telnet.IAC + telnet.WONT + '\x12')
+ self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, b''))
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.WONT + b'\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'
+ cmd = telnet.IAC + telnet.DO + b'\x19'
+ bytes = b'padding' + cmd + b'trailer'
h = self.p.protocol
- h.localEnableable = ('\x19',)
+ h.localEnableable = (b'\x19',)
self.p.dataReceived(bytes)
- self.assertEqual(self.t.value(), telnet.IAC + telnet.WILL + '\x19')
- self._enabledHelper(h, eL=['\x19'])
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.WILL + b'\x19')
+ self._enabledHelper(h, eL=[b'\x19'])
def testAcceptWill(self):
# Same as testAcceptDo, but reversed.
- cmd = telnet.IAC + telnet.WILL + '\x91'
- bytes = 'header' + cmd + 'padding'
+ cmd = telnet.IAC + telnet.WILL + b'\x91'
+ bytes = b'header' + cmd + b'padding'
h = self.p.protocol
- h.remoteEnableable = ('\x91',)
+ h.remoteEnableable = (b'\x91',)
self.p.dataReceived(bytes)
- self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + '\x91')
- self._enabledHelper(h, eR=['\x91'])
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + b'\x91')
+ self._enabledHelper(h, eR=[b'\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'
+ cmd = telnet.IAC + telnet.WONT + b'\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 = self.p.getOptionState(b'\x29')
s.him.state = 'yes'
- bytes = "fiddle dee" + cmd
+ bytes = b"fiddle dee" + cmd
self.p.dataReceived(bytes)
- self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
- self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + '\x29')
+ self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, b''))
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + b'\x29')
self.assertEqual(s.him.state, 'no')
- self._enabledHelper(self.p.protocol, dR=['\x29'])
+ self._enabledHelper(self.p.protocol, dR=[b'\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'
+ cmd = telnet.IAC + telnet.DONT + b'\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 = self.p.getOptionState(b'\x29')
s.us.state = 'yes'
- bytes = "fiddle dum " + cmd
+ bytes = b"fiddle dum " + cmd
self.p.dataReceived(bytes)
- self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
- self.assertEqual(self.t.value(), telnet.IAC + telnet.WONT + '\x29')
+ self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, b''))
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.WONT + b'\x29')
self.assertEqual(s.us.state, 'no')
- self._enabledHelper(self.p.protocol, dL=['\x29'])
+ self._enabledHelper(self.p.protocol, dL=[b'\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'
+ cmd = telnet.IAC + telnet.WONT + b'\x47'
- bytes = "dum de dum" + cmd + "tra la la"
+ bytes = b"dum de dum" + cmd + b"tra la la"
self.p.dataReceived(bytes)
- self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
- self.assertEqual(self.t.value(), '')
+ self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, b''))
+ self.assertEqual(self.t.value(), b'')
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'
+ cmd = telnet.IAC + telnet.DONT + b'\x47'
- bytes = "dum de dum" + cmd + "tra la la"
+ bytes = b"dum de dum" + cmd + b"tra la la"
self.p.dataReceived(bytes)
- self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
- self.assertEqual(self.t.value(), '')
+ self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, b''))
+ self.assertEqual(self.t.value(), b'')
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'
+ cmd = telnet.IAC + telnet.WILL + b'\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 = self.p.getOptionState(b'\x56')
s.him.state = 'yes'
- bytes = "tra la la" + cmd + "dum de dum"
+ bytes = b"tra la la" + cmd + b"dum de dum"
self.p.dataReceived(bytes)
- self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
- self.assertEqual(self.t.value(), '')
+ self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, b''))
+ self.assertEqual(self.t.value(), b'')
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'
+ cmd = telnet.IAC + telnet.DO + b'\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 = self.p.getOptionState(b'\x56')
s.us.state = 'yes'
- bytes = "tra la la" + cmd + "dum de dum"
+ bytes = b"tra la la" + cmd + b"dum de dum"
self.p.dataReceived(bytes)
- self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, ''))
- self.assertEqual(self.t.value(), '')
+ self.assertEqual(self.p.protocol.bytes, bytes.replace(cmd, b''))
+ self.assertEqual(self.t.value(), b'')
self._enabledHelper(self.p.protocol)
def testAcceptedEnableRequest(self):
@@ -398,17 +398,17 @@
# 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')
+ d = self.p.do(b'\x42')
h = self.p.protocol
- h.remoteEnableable = ('\x42',)
+ h.remoteEnableable = (b'\x42',)
- self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + '\x42')
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + b'\x42')
- self.p.dataReceived(telnet.IAC + telnet.WILL + '\x42')
+ self.p.dataReceived(telnet.IAC + telnet.WILL + b'\x42')
d.addCallback(self.assertEqual, True)
- d.addCallback(lambda _: self._enabledHelper(h, eR=['\x42']))
+ d.addCallback(lambda _: self._enabledHelper(h, eR=[b'\x42']))
return d
@@ -422,18 +422,18 @@
# 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.
- self.p.protocol.remoteEnableable = ('\x42',)
- d = self.p.do('\x42')
+ self.p.protocol.remoteEnableable = (b'\x42',)
+ d = self.p.do(b'\x42')
- self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + '\x42')
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + b'\x42')
- s = self.p.getOptionState('\x42')
+ s = self.p.getOptionState(b'\x42')
self.assertEqual(s.him.state, 'no')
self.assertEqual(s.us.state, 'no')
self.assertEqual(s.him.negotiating, True)
self.assertEqual(s.us.negotiating, False)
- self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')
+ self.p.dataReceived(telnet.IAC + telnet.WONT + b'\x42')
d = self.assertFailure(d, telnet.OptionRefused)
d.addCallback(lambda ignored: self._enabledHelper(self.p.protocol))
@@ -452,18 +452,18 @@
# 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.
- self.p.protocol.localEnableable = ('\x42',)
- d = self.p.will('\x42')
+ self.p.protocol.localEnableable = (b'\x42',)
+ d = self.p.will(b'\x42')
- self.assertEqual(self.t.value(), telnet.IAC + telnet.WILL + '\x42')
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.WILL + b'\x42')
- s = self.p.getOptionState('\x42')
+ s = self.p.getOptionState(b'\x42')
self.assertEqual(s.him.state, 'no')
self.assertEqual(s.us.state, 'no')
self.assertEqual(s.him.negotiating, False)
self.assertEqual(s.us.negotiating, True)
- self.p.dataReceived(telnet.IAC + telnet.DONT + '\x42')
+ self.p.dataReceived(telnet.IAC + telnet.DONT + b'\x42')
d = self.assertFailure(d, telnet.OptionRefused)
d.addCallback(lambda ignored: self._enabledHelper(self.p.protocol))
@@ -477,48 +477,48 @@
# 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 = self.p.getOptionState(b'\x42')
s.him.state = 'yes'
- d = self.p.dont('\x42')
+ d = self.p.dont(b'\x42')
- self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + '\x42')
+ self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + b'\x42')
- self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')
+ self.p.dataReceived(telnet.IAC + telnet.WONT + b'\x42')
d.addCallback(self.assertEqual, True)
d.addCallback(lambda _: self._enabledHelper(self.p.protocol,
- dR=['\x42']))
+ dR=[b'\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 = self.p.getOptionState(b'\x24')
s.him.state = 'yes'
- self.p.dont('\x24') # fires after the first line of _final
+ self.p.dont(b'\x24') # fires after the first line of _final
def _do(x):
- d = self.p.do('\x24')
+ d = self.p.do(b'\x24')
return self.assertFailure(d, telnet.AlreadyNegotiating)
def _dont(x):
- d = self.p.dont('\x24')
+ d = self.p.dont(b'\x24')
return self.assertFailure(d, telnet.AlreadyNegotiating)
def _final(x):
- self.p.dataReceived(telnet.IAC + telnet.WONT + '\x24')
+ self.p.dataReceived(telnet.IAC + telnet.WONT + b'\x24')
# an assertion that only passes if d2 has fired
- self._enabledHelper(self.p.protocol, dR=['\x24'])
+ self._enabledHelper(self.p.protocol, dR=[b'\x24'])
# Make sure we allow this
- self.p.protocol.remoteEnableable = ('\x24',)
- d = self.p.do('\x24')
- self.p.dataReceived(telnet.IAC + telnet.WILL + '\x24')
+ self.p.protocol.remoteEnableable = (b'\x24',)
+ d = self.p.do(b'\x24')
+ self.p.dataReceived(telnet.IAC + telnet.WILL + b'\x24')
d.addCallback(self.assertEqual, True)
d.addCallback(lambda _: self._enabledHelper(self.p.protocol,
- eR=['\x24'],
- dR=['\x24']))
+ eR=[b'\x24'],
+ dR=[b'\x24']))
return d
d = _do(None)
@@ -528,20 +528,20 @@
def testSuperfluousDisableRequestRaises(self):
# Try to disable a disabled option. Make sure it fails properly.
- d = self.p.dont('\xab')
+ d = self.p.dont(b'\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 = self.p.getOptionState(b'\xab')
s.him.state = 'yes'
- d = self.p.do('\xab')
+ d = self.p.do(b'\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')
+ d1 = self.p.do(b'\x12')
+ d2 = self.p.do(b'\x23')
+ d3 = self.p.do(b'\x34')
class TestException(Exception):
pass
@@ -604,7 +604,7 @@
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'))
+ self.assertFalse(self.protocol.enableLocal(b'\0'))
def test_enableRemote(self):
@@ -612,7 +612,7 @@
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'))
+ self.assertFalse(self.protocol.enableRemote(b'\0'))
def test_disableLocal(self):
@@ -622,7 +622,7 @@
locally. If a subclass overrides enableLocal, it must also override
disableLocal.
"""
- self.assertRaises(NotImplementedError, self.protocol.disableLocal, '\0')
+ self.assertRaises(NotImplementedError, self.protocol.disableLocal, b'\0')
def test_disableRemote(self):
@@ -632,7 +632,7 @@
enabled remotely. If a subclass overrides enableRemote, it must also
override disableRemote.
"""
- self.assertRaises(NotImplementedError, self.protocol.disableRemote, '\0')
+ self.assertRaises(NotImplementedError, self.protocol.disableRemote, b'\0')
def test_requestNegotiation(self):
@@ -644,11 +644,11 @@
"""
transport = proto_helpers.StringTransport()
self.protocol.makeConnection(transport)
- self.protocol.requestNegotiation('\x01', '\x02\x03')
+ self.protocol.requestNegotiation(b'\x01', b'\x02\x03')
self.assertEqual(
transport.value(),
# IAC SB feature bytes IAC SE
- '\xff\xfa\x01\x02\x03\xff\xf0')
+ b'\xff\xfa\x01\x02\x03\xff\xf0')
def test_requestNegotiationEscapesIAC(self):
@@ -660,19 +660,19 @@
"""
transport = proto_helpers.StringTransport()
self.protocol.makeConnection(transport)
- self.protocol.requestNegotiation('\x01', '\xff')
+ self.protocol.requestNegotiation(b'\x01', b'\xff')
self.assertEqual(
transport.value(),
- '\xff\xfa\x01\xff\xff\xff\xf0')
+ b'\xff\xfa\x01\xff\xff\xff\xf0')
- def _deliver(self, bytes, *expected):
+ def _deliver(self, b, *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.protocol.dataReceived(b)
self.assertEqual(received, list(expected))
@@ -681,7 +681,7 @@
One application-data byte in the default state gets delivered right
away.
"""
- self._deliver('a', ('bytes', 'a'))
+ self._deliver(b'a', ('bytes', b'a'))
def test_twoApplicationDataBytes(self):
@@ -689,7 +689,7 @@
Two application-data bytes in the default state get delivered
together.
"""
- self._deliver('bc', ('bytes', 'bc'))
+ self._deliver(b'bc', ('bytes', b'bc'))
def test_threeApplicationDataBytes(self):
@@ -697,7 +697,7 @@
Three application-data bytes followed by a control byte get
delivered, but the control byte doesn't.
"""
- self._deliver('def' + telnet.IAC, ('bytes', 'def'))
+ self._deliver(b'def' + telnet.IAC, ('bytes', b'def'))
def test_escapedControl(self):
@@ -706,7 +706,7 @@
application-data byte following it.
"""
self._deliver(telnet.IAC)
- self._deliver(telnet.IAC + 'g', ('bytes', telnet.IAC + 'g'))
+ self._deliver(telnet.IAC + b'g', ('bytes', telnet.IAC + b'g'))
def test_carriageReturn(self):
@@ -715,25 +715,25 @@
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.
+ 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(b'\r')
+ self._deliver(b'\n', ('bytes', b'\n'))
+ self._deliver(b'\r\n', ('bytes', b'\n'))
+
+ self._deliver(b'\r')
+ self._deliver(b'\0', ('bytes', b'\r'))
+ self._deliver(b'\r\0', ('bytes', b'\r'))
+
+ self._deliver(b'\r')
+ self._deliver(b'a', ('bytes', b'\ra'))
+ self._deliver(b'\ra', ('bytes', b'\ra'))
- self._deliver('\r')
+ self._deliver(b'\r')
self._deliver(
- telnet.IAC + telnet.IAC + 'x', ('bytes', '\r' + telnet.IAC + 'x'))
+ telnet.IAC + telnet.IAC + b'x', ('bytes', b'\r' + telnet.IAC + b'x'))
def test_applicationDataBeforeSimpleCommand(self):
@@ -742,8 +742,8 @@
command is processed.
"""
self._deliver(
- 'x' + telnet.IAC + telnet.NOP,
- ('bytes', 'x'), ('command', telnet.NOP, None))
+ b'x' + telnet.IAC + telnet.NOP,
+ ('bytes', b'x'), ('command', telnet.NOP, None))
def test_applicationDataBeforeCommand(self):
@@ -753,8 +753,8 @@
"""
self.protocol.commandMap = {}
self._deliver(
- 'y' + telnet.IAC + telnet.WILL + '\x00',
- ('bytes', 'y'), ('command', telnet.WILL, '\x00'))
+ b'y' + telnet.IAC + telnet.WILL + b'\x00',
+ ('bytes', b'y'), ('command', telnet.WILL, b'\x00'))
def test_applicationDataBeforeSubnegotiation(self):
@@ -763,5 +763,5 @@
delivered before the negotiation is processed.
"""
self._deliver(
- 'z' + telnet.IAC + telnet.SB + 'Qx' + telnet.IAC + telnet.SE,
- ('bytes', 'z'), ('negotiate', 'Q', ['x']))
+ b'z' + telnet.IAC + telnet.SB + b'Qx' + telnet.IAC + telnet.SE,
+ ('bytes', b'z'), ('negotiate', b'Q', [b'x']))
Modified: branches/telnet-py3-8228/twisted/protocols/telnet.py
==============================================================================
--- branches/telnet-py3-8228/twisted/protocols/telnet.py (original)
+++ branches/telnet-py3-8228/twisted/protocols/telnet.py Mon Mar 7 22:34:45 2016
@@ -1,10 +1,11 @@
-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
-
-"""TELNET implementation, with line-oriented command handling.
"""
+TELNET implementation, with line-oriented command handling.
+"""
+
+from __future__ import absolute_import, division
import warnings
warnings.warn(
@@ -13,21 +14,16 @@
DeprecationWarning,
stacklevel=2)
+from io import BytesIO
-# System Imports
-try:
- from cStringIO import StringIO
-except ImportError:
- from StringIO import StringIO
-
-# Twisted Imports
from twisted import copyright
from twisted.internet import protocol
+from twisted.python.compat import networkString, iterbytes, _bytesChr as chr
# Some utility chars.
ESC = chr(27) # ESC for doing fanciness
-BOLD_MODE_ON = ESC+"[1m" # turn bold on
-BOLD_MODE_OFF= ESC+"[m" # no char attributes
+BOLD_MODE_ON = ESC + b"[1m" # turn bold on
+BOLD_MODE_OFF= ESC + b"[m" # no char attributes
# Characters gleaned from the various (and conflicting) RFCs. Not all of these are correct.
@@ -122,7 +118,7 @@
NOECHO= chr(131) # User-to-Server: Asks the server not to
# return Echos of the transmitted data.
- #
+ #
# Server-to-User: States that the server is
# not sending echos of the transmitted data.
# Sent only as a reply to ECHO or NO ECHO,
@@ -136,7 +132,7 @@
WILL: 'WILL',
WONT: 'WONT',
IP: 'IP'
- }
+}
def multireplace(st, dct):
for k, v in dct.items():
@@ -144,7 +140,8 @@
return st
class Telnet(protocol.Protocol):
- """I am a Protocol for handling Telnet connections. I have two
+ """
+ I am a Protocol for handling Telnet connections. I have two
sets of special methods, telnet_* and iac_*.
telnet_* methods get called on every line sent to me. The method
@@ -165,9 +162,9 @@
gotIAC = 0
iacByte = None
lastLine = None
- buffer = ''
+ buffer = b''
echo = 0
- delimiters = ['\r\n', '\r\000']
+ delimiters = [b'\r\n', b'\r\000']
mode = "User"
def write(self, data):
@@ -183,13 +180,13 @@
"""Override me to return a string which will be sent to the client
before login."""
x = self.factory.__class__
- return ("\r\n" + x.__module__ + '.' + x.__name__ +
+ return networkString("\r\n" + x.__module__ + '.' + x.__name__ +
'\r\nTwisted %s\r\n' % copyright.version
)
def loginPrompt(self):
"""Override me to return a 'login:'-type prompt."""
- return "username: "
+ return b"username: "
def iacSBchunk(self, chunk):
pass
@@ -214,7 +211,7 @@
in by the current mode. telnet_* methods should return a string which
will become the new mode. If None is returned, the mode will not change.
"""
- mode = getattr(self, "telnet_"+self.mode)(line)
+ mode = getattr(self, "telnet_" + self.mode)(line)
if mode is not None:
self.mode = mode
@@ -224,14 +221,14 @@
you want to do something else when the username is received (ie,
create a new user if the user doesn't exist), override me."""
self.username = user
- self.write(IAC+WILL+ECHO+"password: ")
+ self.write(IAC + WILL + ECHO + b"password: ")
return "Password"
def telnet_Password(self, paswd):
"""I accept a password as an argument, and check it with the
checkUserAndPass method. If the login is successful, I call
loggedIn()."""
- self.write(IAC+WONT+ECHO+"*****\r\n")
+ self.write(IAC + WONT + ECHO + b"*****\r\n")
try:
checked = self.checkUserAndPass(self.username, paswd)
except:
@@ -257,7 +254,7 @@
idx = self.buffer.find(delim)
if idx != -1:
break
-
+
while idx != -1:
buf, self.buffer = self.buffer[:idx], self.buffer[idx+2:]
self.processLine(buf)
@@ -270,9 +267,9 @@
break
def dataReceived(self, data):
- chunk = StringIO()
+ chunk = BytesIO()
# silly little IAC state-machine
- for char in data:
+ for char in iterbytes(data):
if self.gotIAC:
# working on an IAC request state
if self.iacByte:
@@ -280,7 +277,7 @@
if self.iacByte == SB:
if char == SE:
self.iacSBchunk(chunk.getvalue())
- chunk = StringIO()
+ chunk = BytesIO()
del self.iacByte
del self.gotIAC
else:
@@ -306,7 +303,7 @@
why = self.processChunk(c)
if why:
return why
- chunk = StringIO()
+ chunk = BytesIO()
self.gotIAC = 1
else:
chunk.write(char)
@@ -319,7 +316,7 @@
def loggedIn(self):
"""Called after the user succesfully logged in.
-
+
Override in subclasses.
"""
pass
Modified: branches/telnet-py3-8228/twisted/python/compat.py
==============================================================================
--- branches/telnet-py3-8228/twisted/python/compat.py (original)
+++ branches/telnet-py3-8228/twisted/python/compat.py Mon Mar 7 22:34:45 2016
@@ -685,6 +685,18 @@
from base64 import decodestring as _b64decodebytes
+
+def _bytesChr(i):
+ """
+ Like L{chr} but always returns L{bytes}.
+ """
+ if _PY3:
+ return bytes([i])
+ else:
+ return chr(i)
+
+
+
__all__ = [
"reraise",
"execfile",
@@ -717,4 +729,5 @@
"_keys",
"_b64encodebytes",
"_b64decodebytes",
+ "_bytesChr",
]
Modified: branches/telnet-py3-8228/twisted/python/dist3.py
==============================================================================
--- branches/telnet-py3-8228/twisted/python/dist3.py (original)
+++ branches/telnet-py3-8228/twisted/python/dist3.py Mon Mar 7 22:34:45 2016
@@ -52,6 +52,7 @@
"twisted.conch.ssh.common",
"twisted.conch.ssh.keys",
"twisted.conch.ssh.sexpy",
+ "twisted.conch.telnet",
"twisted.conch.test.__init__",
"twisted.copyright",
"twisted.cred.__init__",
@@ -155,6 +156,7 @@
"twisted.protocols.amp",
"twisted.protocols.basic",
"twisted.protocols.policies",
+ "twisted.protocols.telnet",
"twisted.protocols.test.__init__",
"twisted.protocols.tls",
"twisted.python.__init__",
@@ -255,8 +257,9 @@
testModules = [
"twisted.application.test.test_internet",
"twisted.application.test.test_service",
- "twisted.conch.test.test_keys",
"twisted.conch.test.test_checkers",
+ "twisted.conch.test.test_keys",
+ "twisted.conch.test.test_telnet",
"twisted.cred.test.test_cramauth",
"twisted.cred.test.test_cred",
"twisted.cred.test.test_digestauth",