r47020 - tests
hawkowl-TA+aISz0psMTMxyoc4vAAJOcrHinNvQL0E9HWUfgJXw@public.gmane.org Fri, 18 Mar 2016 03:19:07 -0600 (MDT)
| Newsgroups | gmane.comp.python.twisted.commits |
|---|---|
| Message-ID | <[email protected]> |
Author: hawkowl
Date: Fri Mar 18 03:19:02 2016
New Revision: 47020
Modified:
branches/clarity-sshtransport-8237/twisted/conch/ssh/transport.py
branches/clarity-sshtransport-8237/twisted/conch/test/test_transport.py
branches/clarity-sshtransport-8237/twisted/conch/test/test_userauth.py
Log:
tests
Modified: branches/clarity-sshtransport-8237/twisted/conch/ssh/transport.py
==============================================================================
--- branches/clarity-sshtransport-8237/twisted/conch/ssh/transport.py (original)
+++ branches/clarity-sshtransport-8237/twisted/conch/ssh/transport.py Fri Mar 18 03:19:02 2016
@@ -12,11 +12,11 @@
from __future__ import absolute_import, division
+import binascii
+import hmac
+import string
import struct
import zlib
-import string
-import hmac
-import binascii
from hashlib import md5, sha1, sha256, sha512
@@ -27,7 +27,6 @@
from twisted.internet import protocol, defer
from twisted.python import log, randbytes
-
from twisted.conch.ssh import address, keys, _kex
from twisted.conch.ssh.common import (
NS, getNS, MP, getMP, _MPpow, ffs, int_from_bytes
Modified: branches/clarity-sshtransport-8237/twisted/conch/test/test_transport.py
==============================================================================
--- branches/clarity-sshtransport-8237/twisted/conch/test/test_transport.py (original)
+++ branches/clarity-sshtransport-8237/twisted/conch/test/test_transport.py Fri Mar 18 03:19:02 2016
@@ -5,7 +5,10 @@
Tests for ssh/transport.py and the classes therein.
"""
+from __future__ import absolute_import, division
+
import struct
+import binascii
try:
import pyasn1
@@ -125,17 +128,17 @@
"""
A mocked-up version of twisted.conch.ssh.transport.SSHCiphers.
"""
- outCipType = 'test'
+ outCipType = b'test'
encBlockSize = 6
- inCipType = 'test'
+ inCipType = b'test'
decBlockSize = 6
- inMACType = 'test'
- outMACType = 'test'
+ inMACType = b'test'
+ outMACType = b'test'
verifyDigestSize = 1
usedEncrypt = False
usedDecrypt = False
- outMAC = (None, '', '', 1)
- inMAC = (None, '', '', 1)
+ outMAC = (None, b'', b'', 1)
+ inMAC = (None, b'', b'', 1)
keys = ()
@@ -203,7 +206,7 @@
def flush(self, kind):
- return '\x66'
+ return b'\x66'
@@ -258,8 +261,8 @@
Return the public keys that authenticate this server.
"""
return {
- 'ssh-rsa': keys.Key.fromString(keydata.publicRSA_openssh),
- 'ssh-dsa': keys.Key.fromString(keydata.publicDSA_openssh)}
+ b'ssh-rsa': keys.Key.fromString(keydata.publicRSA_openssh),
+ b'ssh-dsa': keys.Key.fromString(keydata.publicDSA_openssh)}
def getPrivateKeys(self):
@@ -267,8 +270,8 @@
Return the private keys that authenticate this server.
"""
return {
- 'ssh-rsa': keys.Key.fromString(keydata.privateRSA_openssh),
- 'ssh-dsa': keys.Key.fromString(keydata.privateDSA_openssh)}
+ b'ssh-rsa': keys.Key.fromString(keydata.privateRSA_openssh),
+ b'ssh-dsa': keys.Key.fromString(keydata.privateDSA_openssh)}
def getPrimes(self):
@@ -286,9 +289,9 @@
# See OpenSSHFactory.getPrimes.
return {
1024: ((2, _kex.getDHGeneratorAndPrime(
- 'diffie-hellman-group1-sha1')[1]),),
+ b'diffie-hellman-group1-sha1')[1]),),
2048: ((3, _kex.getDHGeneratorAndPrime(
- 'diffie-hellman-group1-sha1')[1]),),
+ b'diffie-hellman-group1-sha1')[1]),),
4096: ((5, 7),)}
@@ -345,7 +348,7 @@
"""
Return a consistent entropy value
"""
- return '\x99' * len
+ return b'\x99' * len
self.patch(randbytes, 'secureRandom', secureRandom)
def stubSendPacket(messageType, payload):
self.packets.append((messageType, payload))
@@ -360,10 +363,10 @@
which is started in L{SSHTransportBase.connectionMade} completes and
non-key exchange messages can be sent and received.
"""
- proto.dataReceived("SSH-2.0-BogoClient-1.2i\r\n")
+ proto.dataReceived(b"SSH-2.0-BogoClient-1.2i\r\n")
proto.dispatchMessage(
transport.MSG_KEXINIT, self._A_KEXINIT_MESSAGE)
- proto._keySetup("foo", "bar")
+ proto._keySetup(b"foo", b"bar")
# SSHTransportBase can't handle MSG_NEWKEYS, or it would be the right
# thing to deliver next. _newKeys won't work either, because
# sendKexInit (probably) hasn't been called. sendKexInit is
@@ -390,7 +393,7 @@
Mixin for diffie-hellman-group-exchange-sha1 tests.
"""
- kexAlgorithm = 'diffie-hellman-group-exchange-sha1'
+ kexAlgorithm = b'diffie-hellman-group-exchange-sha1'
hashProcessor = sha1
@@ -400,7 +403,7 @@
Mixin for diffie-hellman-group-exchange-sha256 tests.
"""
- kexAlgorithm = 'diffie-hellman-group-exchange-sha256'
+ kexAlgorithm = b'diffie-hellman-group-exchange-sha256'
hashProcessor = sha256
@@ -421,18 +424,18 @@
"""
_A_KEXINIT_MESSAGE = (
- "\xAA" * 16 +
- common.NS('diffie-hellman-group1-sha1') +
- common.NS('ssh-rsa') +
- common.NS('aes256-ctr') +
- common.NS('aes256-ctr') +
- common.NS('hmac-sha1') +
- common.NS('hmac-sha1') +
- common.NS('none') +
- common.NS('none') +
- common.NS('') +
- common.NS('') +
- '\x00' + '\x00\x00\x00\x00')
+ b"\xAA" * 16 +
+ common.NS(b'diffie-hellman-group1-sha1') +
+ common.NS(b'ssh-rsa') +
+ common.NS(b'aes256-ctr') +
+ common.NS(b'aes256-ctr') +
+ common.NS(b'hmac-sha1') +
+ common.NS(b'hmac-sha1') +
+ common.NS(b'none') +
+ common.NS(b'none') +
+ common.NS(b'') +
+ common.NS(b'') +
+ b'\x00' + b'\x00\x00\x00\x00')
def test_sendVersion(self):
"""
@@ -440,8 +443,8 @@
string.
"""
# the other setup was done in the setup method
- self.assertEqual(self.transport.value().split('\r\n', 1)[0],
- "SSH-2.0-Twisted")
+ self.assertEqual(self.transport.value().split(b'\r\n', 1)[0],
+ b"SSH-2.0-Twisted")
def test_sendPacketPlain(self):
@@ -458,11 +461,11 @@
proto.makeConnection(self.transport)
self.finishKeyExchange(proto)
self.transport.clear()
- message = ord('A')
- payload = 'BCDEFG'
+ message = ord(b'A')
+ payload = b'BCDEFG'
proto.sendPacket(message, payload)
value = self.transport.value()
- self.assertEqual(value, '\x00\x00\x00\x0c\x04ABCDEFG\x99\x99\x99\x99')
+ self.assertEqual(value, b'\x00\x00\x00\x0c\x04ABCDEFG\x99\x99\x99\x99')
def test_sendPacketEncrypted(self):
@@ -475,7 +478,7 @@
self.finishKeyExchange(proto)
proto.currentEncryptions = testCipher = MockCipher()
message = ord('A')
- payload = 'BC'
+ payload = b'BC'
self.transport.clear()
proto.sendPacket(message, payload)
self.assertTrue(testCipher.usedEncrypt)
@@ -483,15 +486,15 @@
self.assertEqual(
value,
# Four byte length prefix
- '\x00\x00\x00\x08'
+ b'\x00\x00\x00\x08'
# One byte padding length
- '\x04'
+ b'\x04'
# The actual application data
- 'ABC'
+ b'ABC'
# "Random" padding - see the secureRandom monkeypatch in setUp
- '\x99\x99\x99\x99'
+ b'\x99\x99\x99\x99'
# The MAC
- '\x02')
+ b'\x02')
def test_sendPacketCompressed(self):
@@ -504,11 +507,11 @@
self.finishKeyExchange(proto)
proto.outgoingCompression = MockCompression()
self.transport.clear()
- proto.sendPacket(ord('A'), 'B')
+ proto.sendPacket(ord('A'), b'B')
value = self.transport.value()
self.assertEqual(
value,
- '\x00\x00\x00\x0c\x08BA\x66\x99\x99\x99\x99\x99\x99\x99\x99')
+ b'\x00\x00\x00\x0c\x08BA\x66\x99\x99\x99\x99\x99\x99\x99\x99')
def test_sendPacketBoth(self):
@@ -523,7 +526,7 @@
proto.currentEncryptions = testCipher = MockCipher()
proto.outgoingCompression = MockCompression()
message = ord('A')
- payload = 'BC'
+ payload = b'BC'
self.transport.clear()
proto.sendPacket(message, payload)
self.assertTrue(testCipher.usedEncrypt)
@@ -531,15 +534,15 @@
self.assertEqual(
value,
# Four byte length prefix
- '\x00\x00\x00\x0e'
+ b'\x00\x00\x00\x0e'
# One byte padding length
- '\x09'
+ b'\x09'
# Compressed application data
- 'CBA\x66'
+ b'CBA\x66'
# "Random" padding - see the secureRandom monkeypatch in setUp
- '\x99\x99\x99\x99\x99\x99\x99\x99\x99'
+ b'\x99\x99\x99\x99\x99\x99\x99\x99\x99'
# The MAC
- '\x02')
+ b'\x02')
def test_getPacketPlain(self):
@@ -551,10 +554,10 @@
proto.makeConnection(self.transport)
self.finishKeyExchange(proto)
self.transport.clear()
- proto.sendPacket(ord('A'), 'BC')
- proto.buf = self.transport.value() + 'extra'
- self.assertEqual(proto.getPacket(), 'ABC')
- self.assertEqual(proto.buf, 'extra')
+ proto.sendPacket(ord('A'), b'BC')
+ proto.buf = self.transport.value() + b'extra'
+ self.assertEqual(proto.getPacket(), b'ABC')
+ self.assertEqual(proto.buf, b'extra')
def test_getPacketEncrypted(self):
@@ -567,15 +570,15 @@
proto.makeConnection(self.transport)
self.transport.clear()
proto.currentEncryptions = testCipher = MockCipher()
- proto.sendPacket(ord('A'), 'BCD')
+ proto.sendPacket(ord('A'), b'BCD')
value = self.transport.value()
proto.buf = value[:MockCipher.decBlockSize]
self.assertEqual(proto.getPacket(), None)
self.assertTrue(testCipher.usedDecrypt)
- self.assertEqual(proto.first, '\x00\x00\x00\x0e\x09A')
+ self.assertEqual(proto.first, b'\x00\x00\x00\x0e\x09A')
proto.buf += value[MockCipher.decBlockSize:]
- self.assertEqual(proto.getPacket(), 'ABCD')
- self.assertEqual(proto.buf, '')
+ self.assertEqual(proto.getPacket(), b'ABCD')
+ self.assertEqual(proto.buf, b'')
def test_getPacketCompressed(self):
@@ -589,9 +592,9 @@
self.transport.clear()
proto.outgoingCompression = MockCompression()
proto.incomingCompression = proto.outgoingCompression
- proto.sendPacket(ord('A'), 'BCD')
+ proto.sendPacket(ord('A'), b'BCD')
proto.buf = self.transport.value()
- self.assertEqual(proto.getPacket(), 'ABCD')
+ self.assertEqual(proto.getPacket(), b'ABCD')
def test_getPacketBoth(self):
@@ -606,17 +609,17 @@
proto.currentEncryptions = MockCipher()
proto.outgoingCompression = MockCompression()
proto.incomingCompression = proto.outgoingCompression
- proto.sendPacket(ord('A'), 'BCDEFG')
+ proto.sendPacket(ord('A'), b'BCDEFG')
proto.buf = self.transport.value()
- self.assertEqual(proto.getPacket(), 'ABCDEFG')
+ self.assertEqual(proto.getPacket(), b'ABCDEFG')
def test_ciphersAreValid(self):
"""
Test that all the supportedCiphers are valid.
"""
- ciphers = transport.SSHCiphers('A', 'B', 'C', 'D')
- iv = key = '\x00' * 16
+ ciphers = transport.SSHCiphers(b'A', b'B', b'C', b'D')
+ iv = key = b'\x00' * 16
for cipName in self.proto.supportedCiphers:
self.assertTrue(ciphers._getCipher(cipName, iv, key))
@@ -637,29 +640,29 @@
bool first packet follows
uint32 0
"""
- value = self.transport.value().split('\r\n', 1)[1]
+ value = self.transport.value().split(b'\r\n', 1)[1]
self.proto.buf = value
packet = self.proto.getPacket()
- self.assertEqual(packet[0], chr(transport.MSG_KEXINIT))
- self.assertEqual(packet[1:17], '\x99' * 16)
+ self.assertEqual(packet[0:1], chr(transport.MSG_KEXINIT))
+ self.assertEqual(packet[1:17], b'\x99' * 16)
(keyExchanges, pubkeys, ciphers1, ciphers2, macs1, macs2,
compressions1, compressions2, languages1, languages2,
buf) = common.getNS(packet[17:], 10)
self.assertEqual(
- keyExchanges, ','.join(self.proto.supportedKeyExchanges))
- self.assertEqual(pubkeys, ','.join(self.proto.supportedPublicKeys))
- self.assertEqual(ciphers1, ','.join(self.proto.supportedCiphers))
- self.assertEqual(ciphers2, ','.join(self.proto.supportedCiphers))
- self.assertEqual(macs1, ','.join(self.proto.supportedMACs))
- self.assertEqual(macs2, ','.join(self.proto.supportedMACs))
+ keyExchanges, b','.join(self.proto.supportedKeyExchanges))
+ self.assertEqual(pubkeys, b','.join(self.proto.supportedPublicKeys))
+ self.assertEqual(ciphers1, b','.join(self.proto.supportedCiphers))
+ self.assertEqual(ciphers2, b','.join(self.proto.supportedCiphers))
+ self.assertEqual(macs1, b','.join(self.proto.supportedMACs))
+ self.assertEqual(macs2, b','.join(self.proto.supportedMACs))
self.assertEqual(compressions1,
- ','.join(self.proto.supportedCompressions))
+ b','.join(self.proto.supportedCompressions))
self.assertEqual(compressions2,
- ','.join(self.proto.supportedCompressions))
- self.assertEqual(languages1, ','.join(self.proto.supportedLanguages))
- self.assertEqual(languages2, ','.join(self.proto.supportedLanguages))
- self.assertEqual(buf, '\x00' * 5)
+ b','.join(self.proto.supportedCompressions))
+ self.assertEqual(languages1, b','.join(self.proto.supportedLanguages))
+ self.assertEqual(languages2, b','.join(self.proto.supportedLanguages))
+ self.assertEqual(buf, b'\x00' * 5)
def test_receiveKEXINITReply(self):
@@ -720,8 +723,8 @@
del self.proto.sendPacket
for messageType in disallowedMessageTypes:
- self.proto.sendPacket(messageType, 'foo')
- self.assertEqual(self.transport.value(), "")
+ self.proto.sendPacket(messageType, b'foo')
+ self.assertEqual(self.transport.value(), b"")
self.finishKeyExchange(self.proto)
# Make the bytes written to the transport cleartext so it's easier to
@@ -731,7 +734,7 @@
# Pseudo-deliver the peer's NEWKEYS message, which should flush the
# messages which were queued above.
self.proto._newKeys()
- self.assertEqual(self.transport.value().count("foo"), 2)
+ self.assertEqual(self.transport.value().count(b"foo"), 2)
def test_sendDebug(self):
@@ -741,11 +744,11 @@
string debug message
string language
"""
- self.proto.sendDebug("test", True, 'en')
+ self.proto.sendDebug(b"test", True, b'en')
self.assertEqual(
self.packets,
[(transport.MSG_DEBUG,
- "\x01\x00\x00\x00\x04test\x00\x00\x00\x02en")])
+ b"\x01\x00\x00\x00\x04test\x00\x00\x00\x02en")])
def test_receiveDebug(self):
@@ -754,8 +757,8 @@
"""
self.proto.dispatchMessage(
transport.MSG_DEBUG,
- '\x01\x00\x00\x00\x04test\x00\x00\x00\x02en')
- self.assertEqual(self.proto.debugs, [(True, 'test', 'en')])
+ b'\x01\x00\x00\x00\x04test\x00\x00\x00\x02en')
+ self.assertEqual(self.proto.debugs, [(True, b'test', b'en')])
def test_sendIgnore(self):
@@ -763,10 +766,10 @@
Test that ignored messages are sent correctly. Payload::
string ignored data
"""
- self.proto.sendIgnore("test")
+ self.proto.sendIgnore(b"test")
self.assertEqual(
self.packets, [(transport.MSG_IGNORE,
- '\x00\x00\x00\x04test')])
+ b'\x00\x00\x00\x04test')])
def test_receiveIgnore(self):
@@ -774,8 +777,8 @@
Test that ignored messages are received correctly. See
test_sendIgnore.
"""
- self.proto.dispatchMessage(transport.MSG_IGNORE, 'test')
- self.assertEqual(self.proto.ignoreds, ['test'])
+ self.proto.dispatchMessage(transport.MSG_IGNORE, b'test')
+ self.assertEqual(self.proto.ignoreds, [b'test'])
def test_sendUnimplemented(self):
@@ -786,7 +789,7 @@
self.proto.sendUnimplemented()
self.assertEqual(
self.packets, [(transport.MSG_UNIMPLEMENTED,
- '\x00\x00\x00\x00')])
+ b'\x00\x00\x00\x00')])
def test_receiveUnimplemented(self):
@@ -795,7 +798,7 @@
test_sendUnimplemented.
"""
self.proto.dispatchMessage(transport.MSG_UNIMPLEMENTED,
- '\x00\x00\x00\xff')
+ b'\x00\x00\x00\xff')
self.assertEqual(self.proto.unimplementeds, [255])
@@ -810,11 +813,11 @@
def stubLoseConnection():
disconnected[0] = True
self.transport.loseConnection = stubLoseConnection
- self.proto.sendDisconnect(0xff, "test")
+ self.proto.sendDisconnect(0xff, b"test")
self.assertEqual(
self.packets,
[(transport.MSG_DISCONNECT,
- "\x00\x00\x00\xff\x00\x00\x00\x04test\x00\x00\x00\x00")])
+ b"\x00\x00\x00\xff\x00\x00\x00\x04test\x00\x00\x00\x00")])
self.assertTrue(disconnected[0])
@@ -828,8 +831,8 @@
disconnected[0] = True
self.transport.loseConnection = stubLoseConnection
self.proto.dispatchMessage(transport.MSG_DISCONNECT,
- '\x00\x00\x00\xff\x00\x00\x00\x04test')
- self.assertEqual(self.proto.errors, [(255, 'test')])
+ b'\x00\x00\x00\xff\x00\x00\x00\x04test')
+ self.assertEqual(self.proto.errors, [(255, b'test')])
self.assertTrue(disconnected[0])
@@ -858,8 +861,8 @@
self.proto.setService(service)
self.assertEqual(self.proto.service, service)
self.assertTrue(service.started)
- self.proto.dispatchMessage(0xff, "test")
- self.assertEqual(self.packets, [(0xff, "test")])
+ self.proto.dispatchMessage(0xff, b"test")
+ self.assertEqual(self.packets, [(0xff, b"test")])
service2 = MockService()
self.proto.setService(service2)
@@ -895,8 +898,8 @@
self.assertTrue(self.proto.isEncrypted('in'))
self.assertTrue(self.proto.isEncrypted('out'))
self.assertTrue(self.proto.isEncrypted('both'))
- self.proto.currentEncryptions = transport.SSHCiphers('none', 'none',
- 'none', 'none')
+ self.proto.currentEncryptions = transport.SSHCiphers(b'none', b'none',
+ b'none', b'none')
self.assertFalse(self.proto.isEncrypted('in'))
self.assertFalse(self.proto.isEncrypted('out'))
self.assertFalse(self.proto.isEncrypted('both'))
@@ -915,8 +918,8 @@
self.assertTrue(self.proto.isVerified('in'))
self.assertTrue(self.proto.isVerified('out'))
self.assertTrue(self.proto.isVerified('both'))
- self.proto.currentEncryptions = transport.SSHCiphers('none', 'none',
- 'none', 'none')
+ self.proto.currentEncryptions = transport.SSHCiphers(b'none', b'none',
+ b'none', b'none')
self.assertFalse(self.proto.isVerified('in'))
self.assertFalse(self.proto.isVerified('out'))
self.assertFalse(self.proto.isVerified('both'))
@@ -950,16 +953,16 @@
def stubLoseConnection():
disconnected[0] = True
self.transport.loseConnection = stubLoseConnection
- for c in version + '\r\n':
+ for c in version + b'\r\n':
self.proto.dataReceived(c)
self.assertTrue(disconnected[0])
self.assertEqual(self.packets[0][0], transport.MSG_DISCONNECT)
self.assertEqual(
self.packets[0][1][3],
chr(transport.DISCONNECT_PROTOCOL_VERSION_NOT_SUPPORTED))
- testBad('SSH-1.5-OpenSSH')
- testBad('SSH-3.0-Twisted')
- testBad('GET / HTTP/1.1')
+ testBad(b'SSH-1.5-OpenSSH')
+ testBad(b'SSH-3.0-Twisted')
+ testBad(b'GET / HTTP/1.1')
def test_dataBeforeVersion(self):
@@ -968,9 +971,9 @@
"""
proto = MockTransportBase()
proto.makeConnection(proto_helpers.StringTransport())
- data = ("""here's some stuff beforehand
+ data = (b"""here's some stuff beforehand
here's some other stuff
-""" + proto.ourVersionString + "\r\n")
+""" + proto.ourVersionString + b"\r\n")
[proto.dataReceived(c) for c in data]
self.assertTrue(proto.gotVersion)
self.assertEqual(proto.otherVersionString, proto.ourVersionString)
@@ -983,9 +986,9 @@
"""
proto = MockTransportBase()
proto.makeConnection(proto_helpers.StringTransport())
- proto.dataReceived("SSH-1.99-OpenSSH\n")
+ proto.dataReceived(b"SSH-1.99-OpenSSH\n")
self.assertTrue(proto.gotVersion)
- self.assertEqual(proto.otherVersionString, "SSH-1.99-OpenSSH")
+ self.assertEqual(proto.otherVersionString, b"SSH-1.99-OpenSSH")
def test_supportedVersionsAreAllowed(self):
@@ -994,9 +997,9 @@
C{supportedVersions}, an unsupported version error is not emitted.
"""
proto = MockTransportBase()
- proto.supportedVersions = ("9.99", )
+ proto.supportedVersions = (b"9.99", )
proto.makeConnection(proto_helpers.StringTransport())
- proto.dataReceived("SSH-9.99-OpenSSH\n")
+ proto.dataReceived(b"SSH-9.99-OpenSSH\n")
self.assertFalse(proto.gotUnsupportedVersion)
@@ -1006,10 +1009,10 @@
C{supportedVersions}, an unsupported version error is emitted.
"""
proto = MockTransportBase()
- proto.supportedVersions = ("2.0", )
+ proto.supportedVersions = (b"2.0", )
proto.makeConnection(proto_helpers.StringTransport())
- proto.dataReceived("SSH-9.99-OpenSSH\n")
- self.assertEqual("9.99", proto.gotUnsupportedVersion)
+ proto.dataReceived(b"SSH-9.99-OpenSSH\n")
+ self.assertEqual(b"9.99", proto.gotUnsupportedVersion)
def test_badPackets(self):
@@ -1025,20 +1028,20 @@
self.assertEqual(self.packets[0][0], transport.MSG_DISCONNECT)
self.assertEqual(self.packets[0][1][3], chr(error))
- testBad('\xff' * 8) # big packet
- testBad('\x00\x00\x00\x05\x00BCDE') # length not modulo blocksize
+ testBad(b'\xff' * 8) # big packet
+ testBad(b'\x00\x00\x00\x05\x00BCDE') # length not modulo blocksize
oldEncryptions = self.proto.currentEncryptions
self.proto.currentEncryptions = MockCipher()
- testBad('\x00\x00\x00\x08\x06AB123456', # bad MAC
+ testBad(b'\x00\x00\x00\x08\x06AB123456', # bad MAC
transport.DISCONNECT_MAC_ERROR)
self.proto.currentEncryptions.decrypt = lambda x: x[:-1]
- testBad('\x00\x00\x00\x08\x06BCDEFGHIJK') # bad decryption
+ testBad(b'\x00\x00\x00\x08\x06BCDEFGHIJK') # bad decryption
self.proto.currentEncryptions = oldEncryptions
self.proto.incomingCompression = MockCompression()
def stubDecompress(payload):
raise Exception('bad compression')
self.proto.incomingCompression.decompress = stubDecompress
- testBad('\x00\x00\x00\x04\x00BCDE', # bad decompression
+ testBad(b'\x00\x00\x00\x04\x00BCDE', # bad decompression
transport.DISCONNECT_COMPRESSION_ERROR)
self.flushLoggedErrors()
@@ -1056,17 +1059,17 @@
self.proto.packets = []
seqnum += 1
- self.proto.dispatchMessage(40, '')
+ self.proto.dispatchMessage(40, b'')
checkUnimplemented()
- transport.messages[41] = 'MSG_fiction'
- self.proto.dispatchMessage(41, '')
+ transport.messages[41] = b'MSG_fiction'
+ self.proto.dispatchMessage(41, b'')
checkUnimplemented()
- self.proto.dispatchMessage(60, '')
+ self.proto.dispatchMessage(60, b'')
checkUnimplemented()
self.proto.setService(MockService())
- self.proto.dispatchMessage(70, '')
+ self.proto.dispatchMessage(70, b'')
checkUnimplemented()
- self.proto.dispatchMessage(71, '')
+ self.proto.dispatchMessage(71, b'')
checkUnimplemented()
@@ -1082,7 +1085,7 @@
proto.setService(MockService())
proto2 = MockTransportBase()
proto2.makeConnection(proto_helpers.StringTransport())
- proto2.sendIgnore('')
+ proto2.sendIgnore(b'')
self.assertNotEqual(proto.gotVersion, proto2.gotVersion)
self.assertNotEqual(proto.transport, proto2.transport)
self.assertNotEqual(proto.outgoingPacketSequence,
@@ -1105,12 +1108,12 @@
Test that _getKey generates the correct keys.
"""
self.proto.kexAlg = self.kexAlgorithm
- self.proto.sessionID = 'EF'
+ self.proto.sessionID = b'EF'
k1 = self.hashProcessor(
- 'AB' + 'CD' + 'K' + self.proto.sessionID).digest()
- k2 = self.hashProcessor('ABCD' + k1).digest()
- self.assertEqual(self.proto._getKey('K', 'AB', 'CD'), k1 + k2)
+ b'AB' + b'CD' + b'K' + self.proto.sessionID).digest()
+ k2 = self.hashProcessor(b'ABCD' + k1).digest()
+ self.assertEqual(self.proto._getKey(b'K', b'AB', b'CD'), k1 + k2)
@@ -1262,35 +1265,35 @@
set up the first common algorithm found in the client's preference
list.
"""
- self.proto.dataReceived( 'SSH-2.0-Twisted\r\n\x00\x00\x01\xf4\x04\x14'
- '\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99'
- '\x99\x00\x00\x00bdiffie-hellman-group1-sha1,diffie-hellman-g'
- 'roup-exchange-sha1,diffie-hellman-group-exchange-sha256\x00'
- '\x00\x00\x0fssh-dss,ssh-rsa\x00\x00\x00\x85aes128-ctr,aes128-'
- 'cbc,aes192-ctr,aes192-cbc,aes256-ctr,aes256-cbc,cast128-ctr,c'
- 'ast128-cbc,blowfish-ctr,blowfish-cbc,3des-ctr,3des-cbc\x00'
- '\x00\x00\x85aes128-ctr,aes128-cbc,aes192-ctr,aes192-cbc,aes25'
- '6-ctr,aes256-cbc,cast128-ctr,cast128-cbc,blowfish-ctr,blowfis'
- 'h-cbc,3des-ctr,3des-cbc\x00\x00\x00\x12hmac-md5,hmac-sha1\x00'
- '\x00\x00\x12hmac-md5,hmac-sha1\x00\x00\x00\tnone,zlib\x00\x00'
- '\x00\tnone,zlib\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
- '\x00\x00\x99\x99\x99\x99')
+ self.proto.dataReceived(b'SSH-2.0-Twisted\r\n\x00\x00\x01\xf4\x04\x14'
+ b'\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99'
+ b'\x99\x00\x00\x00bdiffie-hellman-group1-sha1,diffie-hellman-g'
+ b'roup-exchange-sha1,diffie-hellman-group-exchange-sha256\x00'
+ b'\x00\x00\x0fssh-dss,ssh-rsa\x00\x00\x00\x85aes128-ctr,aes128-'
+ b'cbc,aes192-ctr,aes192-cbc,aes256-ctr,aes256-cbc,cast128-ctr,c'
+ b'ast128-cbc,blowfish-ctr,blowfish-cbc,3des-ctr,3des-cbc\x00'
+ b'\x00\x00\x85aes128-ctr,aes128-cbc,aes192-ctr,aes192-cbc,aes25'
+ b'6-ctr,aes256-cbc,cast128-ctr,cast128-cbc,blowfish-ctr,blowfis'
+ b'h-cbc,3des-ctr,3des-cbc\x00\x00\x00\x12hmac-md5,hmac-sha1\x00'
+ b'\x00\x00\x12hmac-md5,hmac-sha1\x00\x00\x00\tnone,zlib\x00\x00'
+ b'\x00\tnone,zlib\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+ b'\x00\x00\x99\x99\x99\x99')
# Even if as server we prefer diffie-hellman-group-exchange-sha256 the
# client preference is used.
self.assertEqual(self.proto.kexAlg,
- 'diffie-hellman-group1-sha1')
+ b'diffie-hellman-group1-sha1')
self.assertEqual(self.proto.keyAlg,
- 'ssh-dss')
+ b'ssh-dss')
self.assertEqual(self.proto.outgoingCompressionType,
- 'none')
+ b'none')
self.assertEqual(self.proto.incomingCompressionType,
- 'none')
+ b'none')
ne = self.proto.nextEncryptions
- self.assertEqual(ne.outCipType, 'aes128-ctr')
- self.assertEqual(ne.inCipType, 'aes128-ctr')
- self.assertEqual(ne.outMACType, 'hmac-md5')
- self.assertEqual(ne.inMACType, 'hmac-md5')
+ self.assertEqual(ne.outCipType, b'aes128-ctr')
+ self.assertEqual(ne.inCipType, b'aes128-ctr')
+ self.assertEqual(ne.outMACType, b'hmac-md5')
+ self.assertEqual(ne.inMACType, b'hmac-md5')
def test_ignoreGuessPacketKex(self):
@@ -1301,9 +1304,9 @@
the packet is ignored in the case of the key exchange method not
matching.
"""
- kexInitPacket = '\x00' * 16 + (
- ''.join([common.NS(x) for x in
- [','.join(y) for y in
+ kexInitPacket = b'\x00' * 16 + (
+ b''.join([common.NS(x) for x in
+ [b','.join(y) for y in
[self.proto.supportedKeyExchanges[::-1],
self.proto.supportedPublicKeys,
self.proto.supportedCiphers,
@@ -1314,19 +1317,19 @@
self.proto.supportedCompressions,
self.proto.supportedLanguages,
self.proto.supportedLanguages]]])) + (
- '\xff\x00\x00\x00\x00')
+ b'\xff\x00\x00\x00\x00')
self.proto.ssh_KEXINIT(kexInitPacket)
self.assertTrue(self.proto.ignoreNextPacket)
- self.proto.ssh_DEBUG("\x01\x00\x00\x00\x04test\x00\x00\x00\x00")
+ self.proto.ssh_DEBUG(b"\x01\x00\x00\x00\x04test\x00\x00\x00\x00")
self.assertTrue(self.proto.ignoreNextPacket)
- self.proto.ssh_KEX_DH_GEX_REQUEST_OLD('\x00\x00\x08\x00')
+ self.proto.ssh_KEX_DH_GEX_REQUEST_OLD(b'\x00\x00\x08\x00')
self.assertFalse(self.proto.ignoreNextPacket)
self.assertEqual(self.packets, [])
self.proto.ignoreNextPacket = True
- self.proto.ssh_KEX_DH_GEX_REQUEST('\x00\x00\x08\x00' * 3)
+ self.proto.ssh_KEX_DH_GEX_REQUEST(b'\x00\x00\x08\x00' * 3)
self.assertFalse(self.proto.ignoreNextPacket)
self.assertEqual(self.packets, [])
@@ -1336,9 +1339,9 @@
Like test_ignoreGuessPacketKex, but for an incorrectly guessed
public key format.
"""
- kexInitPacket = '\x00' * 16 + (
- ''.join([common.NS(x) for x in
- [','.join(y) for y in
+ kexInitPacket = b'\x00' * 16 + (
+ b''.join([common.NS(x) for x in
+ [b','.join(y) for y in
[self.proto.supportedKeyExchanges,
self.proto.supportedPublicKeys[::-1],
self.proto.supportedCiphers,
@@ -1349,18 +1352,18 @@
self.proto.supportedCompressions,
self.proto.supportedLanguages,
self.proto.supportedLanguages]]])) + (
- '\xff\x00\x00\x00\x00')
+ b'\xff\x00\x00\x00\x00')
self.proto.ssh_KEXINIT(kexInitPacket)
self.assertTrue(self.proto.ignoreNextPacket)
- self.proto.ssh_DEBUG("\x01\x00\x00\x00\x04test\x00\x00\x00\x00")
+ self.proto.ssh_DEBUG(b"\x01\x00\x00\x00\x04test\x00\x00\x00\x00")
self.assertTrue(self.proto.ignoreNextPacket)
- self.proto.ssh_KEX_DH_GEX_REQUEST_OLD('\x00\x00\x08\x00')
+ self.proto.ssh_KEX_DH_GEX_REQUEST_OLD(b'\x00\x00\x08\x00')
self.assertFalse(self.proto.ignoreNextPacket)
self.assertEqual(self.packets, [])
self.proto.ignoreNextPacket = True
- self.proto.ssh_KEX_DH_GEX_REQUEST('\x00\x00\x08\x00' * 3)
+ self.proto.ssh_KEX_DH_GEX_REQUEST(b'\x00\x00\x08\x00' * 3)
self.assertFalse(self.proto.ignoreNextPacket)
self.assertEqual(self.packets, [])
@@ -1374,35 +1377,35 @@
@type kexAlgorithm: C{str}
"""
self.proto.supportedKeyExchanges = [kexAlgorithm]
- self.proto.supportedPublicKeys = ['ssh-rsa']
+ self.proto.supportedPublicKeys = [b'ssh-rsa']
self.proto.dataReceived(self.transport.value())
g, p = _kex.getDHGeneratorAndPrime(kexAlgorithm)
e = pow(g, 5000, p)
self.proto.ssh_KEX_DH_GEX_REQUEST_OLD(common.MP(e))
- y = common.getMP('\x00\x00\x00\x40' + '\x99' * 64)[0]
+ y = common.getMP(b'\x00\x00\x00\x40' + b'\x99' * 64)[0]
f = common._MPpow(self.proto.g, y, self.proto.p)
sharedSecret = common._MPpow(e, y, self.proto.p)
h = sha1()
h.update(common.NS(self.proto.ourVersionString) * 2)
h.update(common.NS(self.proto.ourKexInitPayload) * 2)
- h.update(common.NS(self.proto.factory.publicKeys['ssh-rsa'].blob()))
+ h.update(common.NS(self.proto.factory.publicKeys[b'ssh-rsa'].blob()))
h.update(common.MP(e))
h.update(f)
h.update(sharedSecret)
exchangeHash = h.digest()
- signature = self.proto.factory.privateKeys['ssh-rsa'].sign(
+ signature = self.proto.factory.privateKeys[b'ssh-rsa'].sign(
exchangeHash)
self.assertEqual(
self.packets,
[(transport.MSG_KEXDH_REPLY,
- common.NS(self.proto.factory.publicKeys['ssh-rsa'].blob())
+ common.NS(self.proto.factory.publicKeys[b'ssh-rsa'].blob())
+ f + common.NS(signature)),
- (transport.MSG_NEWKEYS, '')])
+ (transport.MSG_NEWKEYS, b'')])
def test_KEXDH_INIT_GROUP1(self):
@@ -1410,7 +1413,7 @@
KEXDH_INIT messages are processed when the
diffie-hellman-group1-sha1 key exchange algorithm is requested.
"""
- self.assertKexDHInitResponse('diffie-hellman-group1-sha1')
+ self.assertKexDHInitResponse(b'diffie-hellman-group1-sha1')
def test_KEXDH_INIT_GROUP14(self):
@@ -1418,21 +1421,22 @@
KEXDH_INIT messages are processed when the
diffie-hellman-group14-sha1 key exchange algorithm is requested.
"""
- self.assertKexDHInitResponse('diffie-hellman-group14-sha1')
+ self.assertKexDHInitResponse(b'diffie-hellman-group14-sha1')
def test_keySetup(self):
"""
Test that _keySetup sets up the next encryption keys.
"""
- self.proto.kexAlg = 'diffie-hellman-group1-sha1'
+ self.proto.kexAlg = b'diffie-hellman-group1-sha1'
self.proto.nextEncryptions = MockCipher()
- self.simulateKeyExchange('AB', 'CD')
- self.assertEqual(self.proto.sessionID, 'CD')
- self.simulateKeyExchange('AB', 'EF')
- self.assertEqual(self.proto.sessionID, 'CD')
- self.assertEqual(self.packets[-1], (transport.MSG_NEWKEYS, ''))
- newKeys = [self.proto._getKey(c, 'AB', 'EF') for c in 'ABCDEF']
+ self.simulateKeyExchange(b'AB', b'CD')
+ self.assertEqual(self.proto.sessionID, b'CD')
+ self.simulateKeyExchange(b'AB', b'EF')
+ self.assertEqual(self.proto.sessionID, b'CD')
+ self.assertEqual(self.packets[-1], (transport.MSG_NEWKEYS, b''))
+ newKeys = [self.proto._getKey(c, b'AB', b'EF')
+ for c in b'ABCDEF']
self.assertEqual(
self.proto.nextEncryptions.keys,
(newKeys[1], newKeys[3], newKeys[0], newKeys[2], newKeys[5],
@@ -1446,20 +1450,20 @@
"""
self.test_KEXINITMultipleAlgorithms()
- self.proto.nextEncryptions = transport.SSHCiphers('none', 'none',
- 'none', 'none')
- self.proto.ssh_NEWKEYS('')
+ self.proto.nextEncryptions = transport.SSHCiphers(b'none', b'none',
+ b'none', b'none')
+ self.proto.ssh_NEWKEYS(b'')
self.assertIs(self.proto.currentEncryptions,
self.proto.nextEncryptions)
self.assertIs(self.proto.outgoingCompression, None)
self.assertIs(self.proto.incomingCompression, None)
- self.proto.outgoingCompressionType = 'zlib'
- self.simulateKeyExchange('AB', 'CD')
- self.proto.ssh_NEWKEYS('')
+ self.proto.outgoingCompressionType = b'zlib'
+ self.simulateKeyExchange(b'AB', b'CD')
+ self.proto.ssh_NEWKEYS(b'')
self.assertIsNot(self.proto.outgoingCompression, None)
- self.proto.incomingCompressionType = 'zlib'
- self.simulateKeyExchange('AB', 'EF')
- self.proto.ssh_NEWKEYS('')
+ self.proto.incomingCompressionType = b'zlib'
+ self.simulateKeyExchange(b'AB', b'EF')
+ self.proto.ssh_NEWKEYS(b'')
self.assertIsNot(self.proto.incomingCompression, None)
@@ -1468,9 +1472,9 @@
Test that the SERVICE_REQUEST message requests and starts a
service.
"""
- self.proto.ssh_SERVICE_REQUEST(common.NS('ssh-userauth'))
+ self.proto.ssh_SERVICE_REQUEST(common.NS(b'ssh-userauth'))
self.assertEqual(self.packets, [(transport.MSG_SERVICE_ACCEPT,
- common.NS('ssh-userauth'))])
+ common.NS(b'ssh-userauth'))])
self.assertEqual(self.proto.service.name, 'MockService')
@@ -1478,7 +1482,7 @@
"""
Test that NEWKEYS disconnects if it receives data.
"""
- self.proto.ssh_NEWKEYS("bad packet")
+ self.proto.ssh_NEWKEYS(b"bad packet")
self.checkDisconnected()
@@ -1487,7 +1491,7 @@
Test that SERVICE_REQUESTS disconnects if an unknown service is
requested.
"""
- self.proto.ssh_SERVICE_REQUEST(common.NS('no service'))
+ self.proto.ssh_SERVICE_REQUEST(common.NS(b'no service'))
self.checkDisconnected(transport.DISCONNECT_SERVICE_NOT_AVAILABLE)
@@ -1504,14 +1508,14 @@
Diffie-Hellman group.
"""
self.proto.supportedKeyExchanges = [self.kexAlgorithm]
- self.proto.supportedPublicKeys = ['ssh-rsa']
+ self.proto.supportedPublicKeys = [b'ssh-rsa']
self.proto.dataReceived(self.transport.value())
- self.proto.ssh_KEX_DH_GEX_REQUEST_OLD('\x00\x00\x04\x00')
+ self.proto.ssh_KEX_DH_GEX_REQUEST_OLD(b'\x00\x00\x04\x00')
dhGenerator, dhPrime = self.proto.factory.getPrimes().get(1024)[0]
self.assertEqual(
self.packets,
[(transport.MSG_KEX_DH_GEX_GROUP,
- common.MP(dhPrime) + '\x00\x00\x00\x01\x02')])
+ common.MP(dhPrime) + b'\x00\x00\x00\x01\x02')])
self.assertEqual(self.proto.g, 2)
self.assertEqual(self.proto.p, dhPrime)
@@ -1533,15 +1537,15 @@
group.
"""
self.proto.supportedKeyExchanges = [self.kexAlgorithm]
- self.proto.supportedPublicKeys = ['ssh-rsa']
+ self.proto.supportedPublicKeys = [b'ssh-rsa']
self.proto.dataReceived(self.transport.value())
- self.proto.ssh_KEX_DH_GEX_REQUEST('\x00\x00\x04\x00\x00\x00\x08\x00' +
- '\x00\x00\x0c\x00')
+ self.proto.ssh_KEX_DH_GEX_REQUEST(b'\x00\x00\x04\x00\x00\x00\x08\x00' +
+ b'\x00\x00\x0c\x00')
dhGenerator, dhPrime = self.proto.factory.getPrimes().get(1024)[0]
self.assertEqual(
self.packets,
[(transport.MSG_KEX_DH_GEX_GROUP,
- common.MP(dhPrime) + '\x00\x00\x00\x01\x03')])
+ common.MP(dhPrime) + b'\x00\x00\x00\x01\x03')])
self.assertEqual(self.proto.g, 3)
self.assertEqual(self.proto.p, dhPrime)
@@ -1554,14 +1558,14 @@
"""
self.test_KEX_DH_GEX_REQUEST_OLD()
e = pow(self.proto.g, 3, self.proto.p)
- y = common.getMP('\x00\x00\x00\x80' + '\x99' * 128)[0]
+ y = common.getMP(b'\x00\x00\x00\x80' + b'\x99' * 128)[0]
f = common._MPpow(self.proto.g, y, self.proto.p)
sharedSecret = common._MPpow(e, y, self.proto.p)
h = self.hashProcessor()
h.update(common.NS(self.proto.ourVersionString) * 2)
h.update(common.NS(self.proto.ourKexInitPayload) * 2)
- h.update(common.NS(self.proto.factory.publicKeys['ssh-rsa'].blob()))
- h.update('\x00\x00\x04\x00')
+ h.update(common.NS(self.proto.factory.publicKeys[b'ssh-rsa'].blob()))
+ h.update(b'\x00\x00\x04\x00')
h.update(common.MP(self.proto.p))
h.update(common.MP(self.proto.g))
h.update(common.MP(e))
@@ -1572,10 +1576,10 @@
self.assertEqual(
self.packets[1:],
[(transport.MSG_KEX_DH_GEX_REPLY,
- common.NS(self.proto.factory.publicKeys['ssh-rsa'].blob()) +
- f + common.NS(self.proto.factory.privateKeys['ssh-rsa'].sign(
+ common.NS(self.proto.factory.publicKeys[b'ssh-rsa'].blob()) +
+ f + common.NS(self.proto.factory.privateKeys[b'ssh-rsa'].sign(
exchangeHash))),
- (transport.MSG_NEWKEYS, '')])
+ (transport.MSG_NEWKEYS, b'')])
def test_KEX_DH_GEX_INIT_after_REQUEST(self):
@@ -1586,14 +1590,14 @@
"""
self.test_KEX_DH_GEX_REQUEST()
e = pow(self.proto.g, 3, self.proto.p)
- y = common.getMP('\x00\x00\x00\x80' + '\x99' * 128)[0]
+ y = common.getMP(b'\x00\x00\x00\x80' + b'\x99' * 128)[0]
f = common._MPpow(self.proto.g, y, self.proto.p)
sharedSecret = common._MPpow(e, y, self.proto.p)
h = self.hashProcessor()
h.update(common.NS(self.proto.ourVersionString) * 2)
h.update(common.NS(self.proto.ourKexInitPayload) * 2)
- h.update(common.NS(self.proto.factory.publicKeys['ssh-rsa'].blob()))
- h.update('\x00\x00\x04\x00\x00\x00\x08\x00\x00\x00\x0c\x00')
+ h.update(common.NS(self.proto.factory.publicKeys[b'ssh-rsa'].blob()))
+ h.update(b'\x00\x00\x04\x00\x00\x00\x08\x00\x00\x00\x0c\x00')
h.update(common.MP(self.proto.p))
h.update(common.MP(self.proto.g))
h.update(common.MP(e))
@@ -1604,8 +1608,8 @@
self.assertEqual(
self.packets[1],
(transport.MSG_KEX_DH_GEX_REPLY,
- common.NS(self.proto.factory.publicKeys['ssh-rsa'].blob()) +
- f + common.NS(self.proto.factory.privateKeys['ssh-rsa'].sign(
+ common.NS(self.proto.factory.publicKeys[b'ssh-rsa'].blob()) +
+ f + common.NS(self.proto.factory.privateKeys[b'ssh-rsa'].sign(
exchangeHash))))
@@ -1642,8 +1646,8 @@
"""
self.calledVerifyHostKey = True
self.assertEqual(pubKey, self.blob)
- self.assertEqual(fingerprint.replace(':', ''),
- md5(pubKey).hexdigest())
+ self.assertEqual(fingerprint.replace(b':', b''),
+ binascii.hexlify(md5(pubKey).digest()))
return defer.succeed(True)
@@ -1667,35 +1671,35 @@
algorithms will set up the first common algorithm, ordered after our
preference.
"""
- self.proto.dataReceived( 'SSH-2.0-Twisted\r\n\x00\x00\x01\xf4\x04\x14'
- '\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99'
- '\x99\x00\x00\x00bdiffie-hellman-group1-sha1,diffie-hellman-g'
- 'roup-exchange-sha1,diffie-hellman-group-exchange-sha256\x00'
- '\x00\x00\x0fssh-dss,ssh-rsa\x00\x00\x00\x85aes128-ctr,aes128-'
- 'cbc,aes192-ctr,aes192-cbc,aes256-ctr,aes256-cbc,cast128-ctr,c'
- 'ast128-cbc,blowfish-ctr,blowfish-cbc,3des-ctr,3des-cbc\x00'
- '\x00\x00\x85aes128-ctr,aes128-cbc,aes192-ctr,aes192-cbc,aes25'
- '6-ctr,aes256-cbc,cast128-ctr,cast128-cbc,blowfish-ctr,blowfis'
- 'h-cbc,3des-ctr,3des-cbc\x00\x00\x00\x12hmac-md5,hmac-sha1\x00'
- '\x00\x00\x12hmac-md5,hmac-sha1\x00\x00\x00\tzlib,none\x00\x00'
- '\x00\tzlib,none\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
- '\x00\x00\x99\x99\x99\x99')
+ self.proto.dataReceived(b'SSH-2.0-Twisted\r\n\x00\x00\x01\xf4\x04\x14'
+ b'\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99\x99'
+ b'\x99\x00\x00\x00bdiffie-hellman-group1-sha1,diffie-hellman-g'
+ b'roup-exchange-sha1,diffie-hellman-group-exchange-sha256\x00'
+ b'\x00\x00\x0fssh-dss,ssh-rsa\x00\x00\x00\x85aes128-ctr,aes128-'
+ b'cbc,aes192-ctr,aes192-cbc,aes256-ctr,aes256-cbc,cast128-ctr,c'
+ b'ast128-cbc,blowfish-ctr,blowfish-cbc,3des-ctr,3des-cbc\x00'
+ b'\x00\x00\x85aes128-ctr,aes128-cbc,aes192-ctr,aes192-cbc,aes25'
+ b'6-ctr,aes256-cbc,cast128-ctr,cast128-cbc,blowfish-ctr,blowfis'
+ b'h-cbc,3des-ctr,3des-cbc\x00\x00\x00\x12hmac-md5,hmac-sha1\x00'
+ b'\x00\x00\x12hmac-md5,hmac-sha1\x00\x00\x00\tzlib,none\x00\x00'
+ b'\x00\tzlib,none\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+ b'\x00\x00\x99\x99\x99\x99')
# Even if client prefer diffie-hellman-group1-sha1, we will go for
# diffie-hellman-group-exchange-sha256 as this what we prefer and is
# also supported by the server.
self.assertEqual(self.proto.kexAlg,
- 'diffie-hellman-group-exchange-sha256')
+ b'diffie-hellman-group-exchange-sha256')
self.assertEqual(self.proto.keyAlg,
- 'ssh-rsa')
+ b'ssh-rsa')
self.assertEqual(self.proto.outgoingCompressionType,
- 'none')
+ b'none')
self.assertEqual(self.proto.incomingCompressionType,
- 'none')
+ b'none')
ne = self.proto.nextEncryptions
- self.assertEqual(ne.outCipType, 'aes256-ctr')
- self.assertEqual(ne.inCipType, 'aes256-ctr')
- self.assertEqual(ne.outMACType, 'hmac-sha1')
- self.assertEqual(ne.inMACType, 'hmac-sha1')
+ self.assertEqual(ne.outCipType, b'aes256-ctr')
+ self.assertEqual(ne.inCipType, b'aes256-ctr')
+ self.assertEqual(ne.outMACType, b'hmac-sha1')
+ self.assertEqual(ne.inMACType, b'hmac-sha1')
def test_notImplementedClientMethods(self):
@@ -1725,7 +1729,7 @@
# in data returned by self.transport.value()
self.proto.dataReceived(self.transport.value())
- self.assertEqual(common.MP(self.proto.x)[5:], '\x99' * 64)
+ self.assertEqual(common.MP(self.proto.x)[5:], b'\x99' * 64)
# Data sent to server should be a transport.MSG_KEXDH_INIT
# message containing our public key.
@@ -1738,7 +1742,7 @@
KEXINIT messages requesting diffie-hellman-group14-sha1 result in
KEXDH_INIT responses.
"""
- self.assertKexInitResponseForDH('diffie-hellman-group14-sha1')
+ self.assertKexInitResponseForDH(b'diffie-hellman-group14-sha1')
def test_KEXINIT_group1(self):
@@ -1746,7 +1750,7 @@
KEXINIT messages requesting diffie-hellman-group1-sha1 result in
KEXDH_INIT responses.
"""
- self.assertKexInitResponseForDH('diffie-hellman-group1-sha1')
+ self.assertKexInitResponseForDH(b'diffie-hellman-group1-sha1')
def test_KEXINIT_badKexAlg(self):
@@ -1755,8 +1759,8 @@
KEXINIT message but doesn't have a key exchange algorithm that we
understand.
"""
- self.proto.supportedKeyExchanges = ['diffie-hellman-group2-sha1']
- data = self.transport.value().replace('group1', 'group2')
+ self.proto.supportedKeyExchanges = [b'diffie-hellman-group2-sha1']
+ data = self.transport.value().replace(b'group1', b'group2')
self.assertRaises(ConchError, self.proto.dataReceived, data)
@@ -1773,7 +1777,7 @@
h.update(common.NS(self.proto.ourKexInitPayload) * 2)
h.update(common.NS(self.blob))
h.update(self.proto.e)
- h.update('\x00\x00\x00\x01\x02') # f
+ h.update(b'\x00\x00\x00\x01\x02') # f
h.update(sharedSecret)
exchangeHash = h.digest()
@@ -1785,7 +1789,7 @@
signature = self.privObj.sign(exchangeHash)
d = self.proto.ssh_KEX_DH_GEX_GROUP(
- (common.NS(self.blob) + '\x00\x00\x00\x01\x02' +
+ (common.NS(self.blob) + b'\x00\x00\x00\x01\x02' +
common.NS(signature)))
d.addCallback(_cbTestKEXDH_REPLY)
@@ -1796,14 +1800,14 @@
"""
Test that _keySetup sets up the next encryption keys.
"""
- self.proto.kexAlg = 'diffie-hellman-group1-sha1'
+ self.proto.kexAlg = b'diffie-hellman-group1-sha1'
self.proto.nextEncryptions = MockCipher()
- self.simulateKeyExchange('AB', 'CD')
- self.assertEqual(self.proto.sessionID, 'CD')
- self.simulateKeyExchange('AB', 'EF')
- self.assertEqual(self.proto.sessionID, 'CD')
- self.assertEqual(self.packets[-1], (transport.MSG_NEWKEYS, ''))
- newKeys = [self.proto._getKey(c, 'AB', 'EF') for c in 'ABCDEF']
+ self.simulateKeyExchange(b'AB', b'CD')
+ self.assertEqual(self.proto.sessionID, b'CD')
+ self.simulateKeyExchange(b'AB', b'EF')
+ self.assertEqual(self.proto.sessionID, b'CD')
+ self.assertEqual(self.packets[-1], (transport.MSG_NEWKEYS, b''))
+ newKeys = [self.proto._getKey(c, b'AB', b'EF') for c in b'ABCDEF']
self.assertEqual(self.proto.nextEncryptions.keys,
(newKeys[0], newKeys[2], newKeys[1], newKeys[3],
newKeys[4], newKeys[5]))
@@ -1821,25 +1825,25 @@
self.proto.connectionSecure = stubConnectionSecure
self.proto.nextEncryptions = transport.SSHCiphers(
- 'none', 'none', 'none', 'none')
- self.simulateKeyExchange('AB', 'CD')
+ b'none', b'none', b'none', b'none')
+ self.simulateKeyExchange(b'AB', b'CD')
self.assertIsNot(self.proto.currentEncryptions,
self.proto.nextEncryptions)
self.proto.nextEncryptions = MockCipher()
- self.proto.ssh_NEWKEYS('')
+ self.proto.ssh_NEWKEYS(b'')
self.assertIs(self.proto.outgoingCompression, None)
self.assertIs(self.proto.incomingCompression, None)
self.assertIs(self.proto.currentEncryptions,
self.proto.nextEncryptions)
self.assertTrue(secure[0])
- self.proto.outgoingCompressionType = 'zlib'
- self.simulateKeyExchange('AB', 'GH')
- self.proto.ssh_NEWKEYS('')
+ self.proto.outgoingCompressionType = b'zlib'
+ self.simulateKeyExchange(b'AB', b'GH')
+ self.proto.ssh_NEWKEYS(b'')
self.assertIsNot(self.proto.outgoingCompression, None)
- self.proto.incomingCompressionType = 'zlib'
- self.simulateKeyExchange('AB', 'IJ')
- self.proto.ssh_NEWKEYS('')
+ self.proto.incomingCompressionType = b'zlib'
+ self.simulateKeyExchange(b'AB', b'IJ')
+ self.proto.ssh_NEWKEYS(b'')
self.assertIsNot(self.proto.incomingCompression, None)
@@ -1848,7 +1852,7 @@
Test that the SERVICE_ACCEPT packet starts the requested service.
"""
self.proto.instance = MockService()
- self.proto.ssh_SERVICE_ACCEPT('\x00\x00\x00\x0bMockService')
+ self.proto.ssh_SERVICE_ACCEPT(b'\x00\x00\x00\x0bMockService')
self.assertTrue(self.proto.instance.started)
@@ -1858,7 +1862,7 @@
"""
self.proto.requestService(MockService())
self.assertEqual(self.packets, [(transport.MSG_SERVICE_REQUEST,
- '\x00\x00\x00\x0bMockService')])
+ b'\x00\x00\x00\x0bMockService')])
def test_disconnectKEXDH_REPLYBadSignature(self):
@@ -1866,7 +1870,7 @@
Test that KEXDH_REPLY disconnects if the signature is bad.
"""
self.test_KEXDH_REPLY()
- self.proto._continueKEXDH_REPLY(None, self.blob, 3, "bad signature")
+ self.proto._continueKEXDH_REPLY(None, self.blob, 3, b"bad signature")
self.checkDisconnected(transport.DISCONNECT_KEY_EXCHANGE_FAILED)
@@ -1874,7 +1878,7 @@
"""
Test that NEWKEYS disconnects if it receives data.
"""
- self.proto.ssh_NEWKEYS("bad packet")
+ self.proto.ssh_NEWKEYS(b"bad packet")
self.checkDisconnected()
@@ -1884,7 +1888,7 @@
differet from the asked-for protocol.
"""
self.proto.instance = MockService()
- self.proto.ssh_SERVICE_ACCEPT('\x00\x00\x00\x03bad')
+ self.proto.ssh_SERVICE_ACCEPT(b'\x00\x00\x00\x03bad')
self.checkDisconnected()
@@ -1895,7 +1899,7 @@
name of the service.
"""
self.proto.instance = MockService()
- self.proto.ssh_SERVICE_ACCEPT('') # no payload
+ self.proto.ssh_SERVICE_ACCEPT(b'') # no payload
self.assertTrue(self.proto.instance.started)
self.assertEqual(len(self.packets), 0) # not disconnected
@@ -1916,7 +1920,7 @@
# The response will include our advertised group sizes.
self.assertEqual(self.packets, [(
transport.MSG_KEX_DH_GEX_REQUEST,
- '\x00\x00\x04\x00\x00\x00\x08\x00\x00\x00\x20\x00')])
+ b'\x00\x00\x04\x00\x00\x00\x08\x00\x00\x00\x20\x00')])
def test_KEX_DH_GEX_GROUP(self):
@@ -1926,10 +1930,10 @@
"""
self.test_KEXINIT_groupexchange()
self.proto.ssh_KEX_DH_GEX_GROUP(
- '\x00\x00\x00\x01\x0f\x00\x00\x00\x01\x02')
+ b'\x00\x00\x00\x01\x0f\x00\x00\x00\x01\x02')
self.assertEqual(self.proto.p, 15)
self.assertEqual(self.proto.g, 2)
- self.assertEqual(common.MP(self.proto.x)[5:], '\x99' * 40)
+ self.assertEqual(common.MP(self.proto.x)[5:], b'\x99' * 40)
self.assertEqual(self.proto.e,
common.MP(pow(2, self.proto.x, 15)))
self.assertEqual(self.packets[1:], [(transport.MSG_KEX_DH_GEX_INIT,
@@ -1948,10 +1952,10 @@
h.update(common.NS(self.proto.ourKexInitPayload) * 2)
h.update(common.NS(self.blob))
# Here is the wire format for advertised min, pref and max DH sizes.
- h.update('\x00\x00\x04\x00\x00\x00\x08\x00\x00\x00\x20\x00')
- h.update('\x00\x00\x00\x01\x0f\x00\x00\x00\x01\x02')
+ h.update(b'\x00\x00\x04\x00\x00\x00\x08\x00\x00\x00\x20\x00')
+ h.update(b'\x00\x00\x00\x01\x0f\x00\x00\x00\x01\x02')
h.update(self.proto.e)
- h.update('\x00\x00\x00\x01\x03') # f
+ h.update(b'\x00\x00\x00\x01\x03') # f
h.update(sharedSecret)
exchangeHash = h.digest()
@@ -1964,7 +1968,7 @@
d = self.proto.ssh_KEX_DH_GEX_REPLY(
common.NS(self.blob) +
- '\x00\x00\x00\x01\x03' +
+ b'\x00\x00\x00\x01\x03' +
common.NS(signature))
d.addCallback(_cbTestKEX_DH_GEX_REPLY)
return d
@@ -1975,7 +1979,7 @@
Test that KEX_DH_GEX_REPLY disconnects if the signature is bad.
"""
self.test_KEX_DH_GEX_REPLY()
- self.proto._continueGEX_REPLY(None, self.blob, 3, "bad signature")
+ self.proto._continueGEX_REPLY(None, self.blob, 3, b"bad signature")
self.checkDisconnected(transport.DISCONNECT_KEY_EXCHANGE_FAILED)
@@ -2043,7 +2047,7 @@
params = self.ciphers._getMAC(hmacName, secret)
- key = secret[:digestSize] + '\x00' * blockPadSize
+ key = secret[:digestSize] + b'\x00' * blockPadSize
innerPad = b''.join(chr(ord(b) ^ 0x36) for b in key)
outerPad = b''.join(chr(ord(b) ^ 0x5c) for b in key)
self.assertEqual(
@@ -2117,22 +2121,22 @@
"""
Test that the initializer sets up the SSHCiphers object.
"""
- ciphers = transport.SSHCiphers('A', 'B', 'C', 'D')
- self.assertEqual(ciphers.outCipType, 'A')
- self.assertEqual(ciphers.inCipType, 'B')
- self.assertEqual(ciphers.outMACType, 'C')
- self.assertEqual(ciphers.inMACType, 'D')
+ ciphers = transport.SSHCiphers(b'A', b'B', b'C', b'D')
+ self.assertEqual(ciphers.outCipType, b'A')
+ self.assertEqual(ciphers.inCipType, b'B')
+ self.assertEqual(ciphers.outMACType, b'C')
+ self.assertEqual(ciphers.inMACType, b'D')
def test_getCipher(self):
"""
Test that the _getCipher method returns the correct cipher.
"""
- ciphers = transport.SSHCiphers('A', 'B', 'C', 'D')
- iv = key = '\x00' * 16
+ ciphers = transport.SSHCiphers(b'A', b'B', b'C', b'D')
+ iv = key = b'\x00' * 16
for cipName, (algClass, keySize, counter) in ciphers.cipherMap.items():
cip = ciphers._getCipher(cipName, iv, key)
- if cipName == 'none':
+ if cipName == b'none':
self.assertIsInstance(cip, transport._DummyCipher)
else:
self.assertIsInstance(cip.algorithm, algClass)
@@ -2142,15 +2146,15 @@
"""
Test that setKeys sets up the ciphers.
"""
- key = '\x00' * 64
+ key = b'\x00' * 64
for cipName in transport.SSHTransportBase.supportedCiphers:
modName, keySize, counter = transport.SSHCiphers.cipherMap[cipName]
- encCipher = transport.SSHCiphers(cipName, 'none', 'none', 'none')
- decCipher = transport.SSHCiphers('none', cipName, 'none', 'none')
+ encCipher = transport.SSHCiphers(cipName, b'none', b'none', b'none')
+ decCipher = transport.SSHCiphers(b'none', cipName, b'none', b'none')
cip = encCipher._getCipher(cipName, key, key)
bs = cip.algorithm.block_size // 8
- encCipher.setKeys(key, key, '', '', '', '')
- decCipher.setKeys('', '', key, key, '', '')
+ encCipher.setKeys(key, key, b'', b'', b'', b'')
+ decCipher.setKeys(b'', b'', key, key, b'', b'')
self.assertEqual(encCipher.encBlockSize, bs)
self.assertEqual(decCipher.decBlockSize, bs)
encryptor = cip.encryptor()
@@ -2166,12 +2170,12 @@
"""
Test that setKeys sets up the MACs.
"""
- key = '\x00' * 64
+ key = b'\x00' * 64
for macName, mod in transport.SSHCiphers.macMap.items():
- outMac = transport.SSHCiphers('none', 'none', macName, 'none')
- inMac = transport.SSHCiphers('none', 'none', 'none', macName)
- outMac.setKeys('', '', '', '', key, '')
- inMac.setKeys('', '', '', '', '', key)
+ outMac = transport.SSHCiphers(b'none', b'none', macName, b'none')
+ inMac = transport.SSHCiphers(b'none', b'none', b'none', macName)
+ outMac.setKeys(b'', b'', b'', b'', key, b'')
+ inMac.setKeys(b'', b'', b'', b'', b'', key)
if mod:
ds = mod().digest_size
else:
@@ -2181,11 +2185,11 @@
mod, i, o, ds = outMac._getMAC(macName, key)
seqid = 0
data = key
- packet = '\x00' * 4 + key
+ packet = b'\x00' * 4 + key
if mod:
mac = mod(o + mod(i + packet).digest()).digest()
else:
- mac = ''
+ mac = b''
self.assertEqual(outMac.makeMAC(seqid, data), mac)
self.assertTrue(inMac.verify(seqid, data, mac))
@@ -2206,12 +2210,12 @@
]
for key, data, mac in vectors:
- outMAC = transport.SSHCiphers('none', 'none', 'hmac-md5', 'none')
- outMAC.outMAC = outMAC._getMAC("hmac-md5", key)
+ outMAC = transport.SSHCiphers(b'none', b'none', b'hmac-md5', b'none')
+ outMAC.outMAC = outMAC._getMAC(b"hmac-md5", key)
(seqid,) = struct.unpack('>L', data[:4])
shortened = data[4:]
self.assertEqual(
- mac, outMAC.makeMAC(seqid, shortened).encode("hex"),
+ mac, binascii.hexlify(outMAC.makeMAC(seqid, shortened)),
"Failed HMAC test vector; key=%r data=%r" % (key, data))
@@ -2255,14 +2259,14 @@
self.assertEqual(client.errors, [])
self.assertEqual(server.errors, [(
transport.DISCONNECT_CONNECTION_LOST,
- "user closed connection")])
- if server.supportedCiphers[0] == 'none':
+ b"user closed connection")])
+ if server.supportedCiphers[0] == b'none':
self.assertFalse(server.isEncrypted(), name)
self.assertFalse(client.isEncrypted(), name)
else:
self.assertTrue(server.isEncrypted(), name)
self.assertTrue(client.isEncrypted(), name)
- if server.supportedMACs[0] == 'none':
+ if server.supportedMACs[0] == b'none':
self.assertFalse(server.isVerified(), name)
self.assertFalse(client.isVerified(), name)
else:
@@ -2280,7 +2284,7 @@
the various combinations of ciphers.
"""
deferreds = []
- for cipher in transport.SSHTransportBase.supportedCiphers + ['none']:
+ for cipher in transport.SSHTransportBase.supportedCiphers + [b'none']:
def setCipher(proto):
proto.supportedCiphers = [cipher]
return proto
@@ -2293,7 +2297,7 @@
Like test_ciphers, but for the various MACs.
"""
deferreds = []
- for mac in transport.SSHTransportBase.supportedMACs + ['none']:
+ for mac in transport.SSHTransportBase.supportedMACs + [b'none']:
def setMAC(proto):
proto.supportedMACs = [mac]
return proto
Modified: branches/clarity-sshtransport-8237/twisted/conch/test/test_userauth.py
==============================================================================
--- branches/clarity-sshtransport-8237/twisted/conch/test/test_userauth.py (original)
+++ branches/clarity-sshtransport-8237/twisted/conch/test/test_userauth.py Fri Mar 18 03:19:02 2016
@@ -73,14 +73,14 @@
"""
Return 'foo' as the password.
"""
- return defer.succeed('foo')
+ return defer.succeed(b'foo')
def getGenericAnswers(self, name, information, answers):
"""
Return 'foo' as the answer to two questions.
"""
- return defer.succeed(('foo', 'foo'))
+ return defer.succeed((b'foo', b'foo'))