1
# -*- test-case-name: twisted.conch.test.test_telnet -*-
2
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
3
# See LICENSE for details.
5
from zope.interface import implements
7
from twisted.internet import defer
9
from twisted.conch import telnet
11
from twisted.trial import unittest
12
from twisted.test import proto_helpers
15
implements(telnet.ITelnetProtocol)
25
self.enabledLocal = []
26
self.enabledRemote = []
27
self.disabledLocal = []
28
self.disabledRemote = []
30
def makeConnection(self, transport):
31
d = transport.negotiationMap = {}
32
d['\x12'] = self.neg_TEST_COMMAND
34
d = transport.commandMap = transport.commandMap.copy()
35
for cmd in ('NOP', 'DM', 'BRK', 'IP', 'AO', 'AYT', 'EC', 'EL', 'GA'):
36
d[getattr(telnet, cmd)] = lambda arg, cmd=cmd: self.calls.append(cmd)
38
def dataReceived(self, bytes):
41
def connectionLost(self, reason):
44
def neg_TEST_COMMAND(self, payload):
47
def enableLocal(self, option):
48
if option in self.localEnableable:
49
self.enabledLocal.append(option)
53
def disableLocal(self, option):
54
self.disabledLocal.append(option)
56
def enableRemote(self, option):
57
if option in self.remoteEnableable:
58
self.enabledRemote.append(option)
62
def disableRemote(self, option):
63
self.disabledRemote.append(option)
65
class TelnetTestCase(unittest.TestCase):
67
self.p = telnet.TelnetTransport(TestProtocol)
68
self.t = proto_helpers.StringTransport()
69
self.p.makeConnection(self.t)
71
def testRegularBytes(self):
72
# Just send a bunch of bytes. None of these do anything
73
# with telnet. They should pass right through to the
77
L = ["here are some bytes la la la",
78
"some more arrive here",
79
"lots of bytes to play with",
84
self.p.dataReceived(b)
86
self.assertEquals(h.bytes, ''.join(L))
88
def testNewlineHandling(self):
89
# Send various kinds of newlines and make sure they get translated
93
L = ["here is the first line\r\n",
94
"here is the second line\r\0",
95
"here is the third line\r\n",
96
"here is the last line\r\0"]
99
self.p.dataReceived(b)
101
self.assertEquals(h.bytes, L[0][:-2] + '\n' +
106
def testIACEscape(self):
107
# Send a bunch of bytes and a couple quoted \xFFs. Unquoted,
108
# \xFF is a telnet command. Quoted, one of them from each pair
109
# should be passed through to the application layer.
112
L = ["here are some bytes\xff\xff with an embedded IAC",
113
"and here is a test of a border escape\xff",
114
"\xff did you get that IAC?"]
117
self.p.dataReceived(b)
119
self.assertEquals(h.bytes, ''.join(L).replace('\xff\xff', '\xff'))
121
def _simpleCommandTest(self, cmdName):
122
# Send a single simple telnet command and make sure
123
# it gets noticed and the appropriate method gets
127
cmd = telnet.IAC + getattr(telnet, cmdName)
128
L = ["Here's some bytes, tra la la",
129
"But ono!" + cmd + " an interrupt"]
132
self.p.dataReceived(b)
134
self.assertEquals(h.calls, [cmdName])
135
self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))
137
def testInterrupt(self):
138
self._simpleCommandTest("IP")
140
def testNoOperation(self):
141
self._simpleCommandTest("NOP")
143
def testDataMark(self):
144
self._simpleCommandTest("DM")
147
self._simpleCommandTest("BRK")
149
def testAbortOutput(self):
150
self._simpleCommandTest("AO")
152
def testAreYouThere(self):
153
self._simpleCommandTest("AYT")
155
def testEraseCharacter(self):
156
self._simpleCommandTest("EC")
158
def testEraseLine(self):
159
self._simpleCommandTest("EL")
161
def testGoAhead(self):
162
self._simpleCommandTest("GA")
164
def testSubnegotiation(self):
165
# Send a subnegotiation command and make sure it gets
166
# parsed and that the correct method is called.
169
cmd = telnet.IAC + telnet.SB + '\x12hello world' + telnet.IAC + telnet.SE
170
L = ["These are some bytes but soon" + cmd,
171
"there will be some more"]
174
self.p.dataReceived(b)
176
self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))
177
self.assertEquals(h.subcmd, list("hello world"))
179
def testSubnegotiationWithEmbeddedSE(self):
180
# Send a subnegotiation command with an embedded SE. Make sure
181
# that SE gets passed to the correct method.
184
cmd = (telnet.IAC + telnet.SB +
186
telnet.IAC + telnet.SE)
188
L = ["Some bytes are here" + cmd + "and here",
192
self.p.dataReceived(b)
194
self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))
195
self.assertEquals(h.subcmd, [telnet.SE])
197
def testBoundarySubnegotiation(self):
198
# Send a subnegotiation command. Split it at every possible byte boundary
199
# and make sure it always gets parsed and that it is passed to the correct
201
cmd = (telnet.IAC + telnet.SB +
202
'\x12' + telnet.SE + 'hello' +
203
telnet.IAC + telnet.SE)
205
for i in range(len(cmd)):
206
h = self.p.protocol = TestProtocol()
207
h.makeConnection(self.p)
209
a, b = cmd[:i], cmd[i:]
210
L = ["first part" + a,
214
self.p.dataReceived(bytes)
216
self.assertEquals(h.bytes, ''.join(L).replace(cmd, ''))
217
self.assertEquals(h.subcmd, [telnet.SE] + list('hello'))
219
def _enabledHelper(self, o, eL=[], eR=[], dL=[], dR=[]):
220
self.assertEquals(o.enabledLocal, eL)
221
self.assertEquals(o.enabledRemote, eR)
222
self.assertEquals(o.disabledLocal, dL)
223
self.assertEquals(o.disabledRemote, dR)
225
def testRefuseWill(self):
226
# Try to enable an option. The server should refuse to enable it.
227
cmd = telnet.IAC + telnet.WILL + '\x12'
229
bytes = "surrounding bytes" + cmd + "to spice things up"
230
self.p.dataReceived(bytes)
232
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))
233
self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x12')
234
self._enabledHelper(self.p.protocol)
236
def testRefuseDo(self):
237
# Try to enable an option. The server should refuse to enable it.
238
cmd = telnet.IAC + telnet.DO + '\x12'
240
bytes = "surrounding bytes" + cmd + "to spice things up"
241
self.p.dataReceived(bytes)
243
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))
244
self.assertEquals(self.t.value(), telnet.IAC + telnet.WONT + '\x12')
245
self._enabledHelper(self.p.protocol)
247
def testAcceptDo(self):
248
# Try to enable an option. The option is in our allowEnable
249
# list, so we will allow it to be enabled.
250
cmd = telnet.IAC + telnet.DO + '\x19'
251
bytes = 'padding' + cmd + 'trailer'
254
h.localEnableable = ('\x19',)
255
self.p.dataReceived(bytes)
257
self.assertEquals(self.t.value(), telnet.IAC + telnet.WILL + '\x19')
258
self._enabledHelper(h, eL=['\x19'])
260
def testAcceptWill(self):
261
# Same as testAcceptDo, but reversed.
262
cmd = telnet.IAC + telnet.WILL + '\x91'
263
bytes = 'header' + cmd + 'padding'
266
h.remoteEnableable = ('\x91',)
267
self.p.dataReceived(bytes)
269
self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x91')
270
self._enabledHelper(h, eR=['\x91'])
272
def testAcceptWont(self):
273
# Try to disable an option. The server must allow any option to
274
# be disabled at any time. Make sure it disables it and sends
275
# back an acknowledgement of this.
276
cmd = telnet.IAC + telnet.WONT + '\x29'
278
# Jimmy it - after these two lines, the server will be in a state
279
# such that it believes the option to have been previously enabled
280
# via normal negotiation.
281
s = self.p.getOptionState('\x29')
284
bytes = "fiddle dee" + cmd
285
self.p.dataReceived(bytes)
287
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))
288
self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x29')
289
self.assertEquals(s.him.state, 'no')
290
self._enabledHelper(self.p.protocol, dR=['\x29'])
292
def testAcceptDont(self):
293
# Try to disable an option. The server must allow any option to
294
# be disabled at any time. Make sure it disables it and sends
295
# back an acknowledgement of this.
296
cmd = telnet.IAC + telnet.DONT + '\x29'
298
# Jimmy it - after these two lines, the server will be in a state
299
# such that it believes the option to have beenp previously enabled
300
# via normal negotiation.
301
s = self.p.getOptionState('\x29')
304
bytes = "fiddle dum " + cmd
305
self.p.dataReceived(bytes)
307
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))
308
self.assertEquals(self.t.value(), telnet.IAC + telnet.WONT + '\x29')
309
self.assertEquals(s.us.state, 'no')
310
self._enabledHelper(self.p.protocol, dL=['\x29'])
312
def testIgnoreWont(self):
313
# Try to disable an option. The option is already disabled. The
314
# server should send nothing in response to this.
315
cmd = telnet.IAC + telnet.WONT + '\x47'
317
bytes = "dum de dum" + cmd + "tra la la"
318
self.p.dataReceived(bytes)
320
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))
321
self.assertEquals(self.t.value(), '')
322
self._enabledHelper(self.p.protocol)
324
def testIgnoreDont(self):
325
# Try to disable an option. The option is already disabled. The
326
# server should send nothing in response to this. Doing so could
327
# lead to a negotiation loop.
328
cmd = telnet.IAC + telnet.DONT + '\x47'
330
bytes = "dum de dum" + cmd + "tra la la"
331
self.p.dataReceived(bytes)
333
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))
334
self.assertEquals(self.t.value(), '')
335
self._enabledHelper(self.p.protocol)
337
def testIgnoreWill(self):
338
# Try to enable an option. The option is already enabled. The
339
# server should send nothing in response to this. Doing so could
340
# lead to a negotiation loop.
341
cmd = telnet.IAC + telnet.WILL + '\x56'
343
# Jimmy it - after these two lines, the server will be in a state
344
# such that it believes the option to have been previously enabled
345
# via normal negotiation.
346
s = self.p.getOptionState('\x56')
349
bytes = "tra la la" + cmd + "dum de dum"
350
self.p.dataReceived(bytes)
352
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))
353
self.assertEquals(self.t.value(), '')
354
self._enabledHelper(self.p.protocol)
356
def testIgnoreDo(self):
357
# Try to enable an option. The option is already enabled. The
358
# server should send nothing in response to this. Doing so could
359
# lead to a negotiation loop.
360
cmd = telnet.IAC + telnet.DO + '\x56'
362
# Jimmy it - after these two lines, the server will be in a state
363
# such that it believes the option to have been previously enabled
364
# via normal negotiation.
365
s = self.p.getOptionState('\x56')
368
bytes = "tra la la" + cmd + "dum de dum"
369
self.p.dataReceived(bytes)
371
self.assertEquals(self.p.protocol.bytes, bytes.replace(cmd, ''))
372
self.assertEquals(self.t.value(), '')
373
self._enabledHelper(self.p.protocol)
375
def testAcceptedEnableRequest(self):
376
# Try to enable an option through the user-level API. This
377
# returns a Deferred that fires when negotiation about the option
378
# finishes. Make sure it fires, make sure state gets updated
379
# properly, make sure the result indicates the option was enabled.
380
d = self.p.do('\x42')
383
h.remoteEnableable = ('\x42',)
385
self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x42')
387
self.p.dataReceived(telnet.IAC + telnet.WILL + '\x42')
389
d.addCallback(self.assertEquals, True)
390
d.addCallback(lambda _: self._enabledHelper(h, eR=['\x42']))
393
def testRefusedEnableRequest(self):
394
# Try to enable an option through the user-level API. This
395
# returns a Deferred that fires when negotiation about the option
396
# finishes. Make sure it fires, make sure state gets updated
397
# properly, make sure the result indicates the option was enabled.
398
d = self.p.do('\x42')
400
self.assertEquals(self.t.value(), telnet.IAC + telnet.DO + '\x42')
402
self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')
404
d = self.assertFailure(d, telnet.OptionRefused)
405
d.addCallback(lambda _: self._enabledHelper(self.p.protocol))
408
def testAcceptedDisableRequest(self):
409
# Try to disable an option through the user-level API. This
410
# returns a Deferred that fires when negotiation about the option
411
# finishes. Make sure it fires, make sure state gets updated
412
# properly, make sure the result indicates the option was enabled.
413
s = self.p.getOptionState('\x42')
416
d = self.p.dont('\x42')
418
self.assertEquals(self.t.value(), telnet.IAC + telnet.DONT + '\x42')
420
self.p.dataReceived(telnet.IAC + telnet.WONT + '\x42')
422
d.addCallback(self.assertEquals, True)
423
d.addCallback(lambda _: self._enabledHelper(self.p.protocol,
427
def testNegotiationBlocksFurtherNegotiation(self):
428
# Try to disable an option, then immediately try to enable it, then
429
# immediately try to disable it. Ensure that the 2nd and 3rd calls
430
# fail quickly with the right exception.
431
s = self.p.getOptionState('\x24')
433
d2 = self.p.dont('\x24') # fires after the first line of _final
436
d = self.p.do('\x24')
437
return self.assertFailure(d, telnet.AlreadyNegotiating)
440
d = self.p.dont('\x24')
441
return self.assertFailure(d, telnet.AlreadyNegotiating)
444
self.p.dataReceived(telnet.IAC + telnet.WONT + '\x24')
445
# an assertion that only passes if d2 has fired
446
self._enabledHelper(self.p.protocol, dR=['\x24'])
447
# Make sure we allow this
448
self.p.protocol.remoteEnableable = ('\x24',)
449
d = self.p.do('\x24')
450
self.p.dataReceived(telnet.IAC + telnet.WILL + '\x24')
451
d.addCallback(self.assertEquals, True)
452
d.addCallback(lambda _: self._enabledHelper(self.p.protocol,
459
d.addCallback(_final)
462
def testSuperfluousDisableRequestRaises(self):
463
# Try to disable a disabled option. Make sure it fails properly.
464
d = self.p.dont('\xab')
465
return self.assertFailure(d, telnet.AlreadyDisabled)
467
def testSuperfluousEnableRequestRaises(self):
468
# Try to disable a disabled option. Make sure it fails properly.
469
s = self.p.getOptionState('\xab')
471
d = self.p.do('\xab')
472
return self.assertFailure(d, telnet.AlreadyEnabled)
474
def testLostConnectionFailsDeferreds(self):
475
d1 = self.p.do('\x12')
476
d2 = self.p.do('\x23')
477
d3 = self.p.do('\x34')
479
class TestException(Exception):
482
self.p.connectionLost(TestException("Total failure!"))
484
d1 = self.assertFailure(d1, TestException)
485
d2 = self.assertFailure(d2, TestException)
486
d3 = self.assertFailure(d3, TestException)
487
return defer.gatherResults([d1, d2, d3])
added
removed
twisted/conch/test/test_telnet.py
