r47004 - porting
hawkowl-TA+aISz0psMTMxyoc4vAAJOcrHinNvQL0E9HWUfgJXw@public.gmane.org Wed, 16 Mar 2016 01:47:59 -0600 (MDT)
| Newsgroups | gmane.comp.python.twisted.commits |
|---|---|
| Message-ID | <[email protected]> |
Author: hawkowl
Date: Wed Mar 16 01:47:54 2016
New Revision: 47004
Modified:
branches/conch-transport-py3-8232/twisted/conch/ssh/_kex.py
branches/conch-transport-py3-8232/twisted/conch/ssh/address.py
branches/conch-transport-py3-8232/twisted/conch/ssh/factory.py
branches/conch-transport-py3-8232/twisted/conch/ssh/transport.py
branches/conch-transport-py3-8232/twisted/conch/test/test_transport.py
Log:
porting
Modified: branches/conch-transport-py3-8232/twisted/conch/ssh/_kex.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/conch/ssh/_kex.py (original)
+++ branches/conch-transport-py3-8232/twisted/conch/ssh/_kex.py Wed Mar 16 01:47:54 2016
@@ -6,11 +6,14 @@
SSH key exchange handling.
"""
+from __future__ import absolute_import, division
+
from hashlib import sha1, sha256
from zope.interface import Attribute, implementer, Interface
from twisted.conch import error
+from twisted.python.compat import long
class _IKexAlgorithm(Interface):
@@ -95,8 +98,8 @@
'44236841971802161585193689478337958649255415021805654859805036464405'
'48199239100050792877003355816639229553136239076508735759914822574862'
'57500742530207744771258955095793777842444242661733472762929938766870'
- '9205606050270810842907692932019128194467627007L')
- generator = 2L
+ '9205606050270810842907692932019128194467627007')
+ generator = long(2)
@@ -119,17 +122,17 @@
'00977202194168647225871031411336429319536193471636533209717077448227'
'98858856536920864529663607725026895550592836275112117409697299806841'
'05543595848665832916421362182310789909994486524682624169720359118525'
- '07045361090559L')
- generator = 2L
+ '07045361090559')
+ generator = long(2)
_kexAlgorithms = {
- "diffie-hellman-group-exchange-sha256": _DHGroupExchangeSHA256(),
- "diffie-hellman-group-exchange-sha1": _DHGroupExchangeSHA1(),
- "diffie-hellman-group1-sha1": _DHGroup1SHA1(),
- "diffie-hellman-group14-sha1": _DHGroup14SHA1(),
- }
+ b"diffie-hellman-group-exchange-sha256": _DHGroupExchangeSHA256(),
+ b"diffie-hellman-group-exchange-sha1": _DHGroupExchangeSHA1(),
+ b"diffie-hellman-group1-sha1": _DHGroup1SHA1(),
+ b"diffie-hellman-group14-sha1": _DHGroup14SHA1(),
+}
Modified: branches/conch-transport-py3-8232/twisted/conch/ssh/address.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/conch/ssh/address.py (original)
+++ branches/conch-transport-py3-8232/twisted/conch/ssh/address.py Wed Mar 16 01:47:54 2016
@@ -18,7 +18,7 @@
@implementer(IAddress)
-class SSHTransportAddress(object, util.FancyEqMixin):
+class SSHTransportAddress(util.FancyEqMixin, object):
"""
Object representing an SSH Transport endpoint.
@@ -43,4 +43,3 @@
def __hash__(self):
return hash(('SSH', self.address))
-
Modified: branches/conch-transport-py3-8232/twisted/conch/ssh/factory.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/conch/ssh/factory.py (original)
+++ branches/conch-transport-py3-8232/twisted/conch/ssh/factory.py Wed Mar 16 01:47:54 2016
@@ -8,12 +8,16 @@
Maintainer: Paul Swartz
"""
+from __future__ import absolute_import, division
+
+from functools import cmp_to_key
+
from twisted.internet import protocol
from twisted.python import log
+from twisted.python.compat import cmp
from twisted.conch import error
-from twisted.conch.ssh import _kex
-import transport, userauth, connection
+from twisted.conch.ssh import _kex, transport, userauth, connection
import random
@@ -103,8 +107,8 @@
@type bits: C{int}
@rtype: C{tuple}
"""
- primesKeys = self.primes.keys()
- primesKeys.sort(lambda x, y: cmp(abs(x - bits), abs(y - bits)))
+ primesKeys = list(self.primes.keys())
+ primesKeys.sort(key=cmp_to_key(lambda x, y: cmp(abs(x - bits), abs(y - bits))))
realBits = primesKeys[0]
return random.choice(self.primes[realBits])
Modified: branches/conch-transport-py3-8232/twisted/conch/ssh/transport.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/conch/ssh/transport.py (original)
+++ branches/conch-transport-py3-8232/twisted/conch/ssh/transport.py Wed Mar 16 01:47:54 2016
@@ -10,11 +10,15 @@
Maintainer: Paul Swartz
"""
+from __future__ import absolute_import, division
+
import struct
import zlib
-from hashlib import md5, sha1, sha256, sha512
import string
import hmac
+import binascii
+
+from hashlib import md5, sha1, sha256, sha512
from cryptography.exceptions import UnsupportedAlgorithm
from cryptography.hazmat.backends import default_backend
@@ -22,8 +26,9 @@
from twisted.internet import protocol, defer
from twisted.python import log, randbytes
+from twisted.python.compat import items, _bytesChr as chr, networkString, iterbytes, nativeString
-
+from twisted.conch.error import ConchError
from twisted.conch.ssh import address, keys, _kex
from twisted.conch.ssh.common import (
NS, getNS, MP, getMP, _MPpow, ffs, int_from_bytes
@@ -48,7 +53,7 @@
"""
if bits % 8:
raise ValueError("bits (%d) must be a multiple of 8" % (bits,))
- return int_from_bytes(random(bits / 8), 'big')
+ return int_from_bytes(random(bits // 8), 'big')
@@ -122,26 +127,26 @@
"""
cipherMap = {
- '3des-cbc': (algorithms.TripleDES, 24, modes.CBC),
- 'blowfish-cbc': (algorithms.Blowfish, 16, modes.CBC),
- 'aes256-cbc': (algorithms.AES, 32, modes.CBC),
- 'aes192-cbc': (algorithms.AES, 24, modes.CBC),
- 'aes128-cbc': (algorithms.AES, 16, modes.CBC),
- 'cast128-cbc': (algorithms.CAST5, 16, modes.CBC),
- 'aes128-ctr': (algorithms.AES, 16, modes.CTR),
- 'aes192-ctr': (algorithms.AES, 24, modes.CTR),
- 'aes256-ctr': (algorithms.AES, 32, modes.CTR),
- '3des-ctr': (algorithms.TripleDES, 24, modes.CTR),
- 'blowfish-ctr': (algorithms.Blowfish, 16, modes.CTR),
- 'cast128-ctr': (algorithms.CAST5, 16, modes.CTR),
- 'none': (None, 0, modes.CBC),
+ b'3des-cbc': (algorithms.TripleDES, 24, modes.CBC),
+ b'blowfish-cbc': (algorithms.Blowfish, 16, modes.CBC),
+ b'aes256-cbc': (algorithms.AES, 32, modes.CBC),
+ b'aes192-cbc': (algorithms.AES, 24, modes.CBC),
+ b'aes128-cbc': (algorithms.AES, 16, modes.CBC),
+ b'cast128-cbc': (algorithms.CAST5, 16, modes.CBC),
+ b'aes128-ctr': (algorithms.AES, 16, modes.CTR),
+ b'aes192-ctr': (algorithms.AES, 24, modes.CTR),
+ b'aes256-ctr': (algorithms.AES, 32, modes.CTR),
+ b'3des-ctr': (algorithms.TripleDES, 24, modes.CTR),
+ b'blowfish-ctr': (algorithms.Blowfish, 16, modes.CTR),
+ b'cast128-ctr': (algorithms.CAST5, 16, modes.CTR),
+ b'none': (None, 0, modes.CBC),
}
macMap = {
- 'hmac-sha2-512': sha512,
- 'hmac-sha2-256': sha256,
- 'hmac-sha1': sha1,
- 'hmac-md5': md5,
- 'none': None
+ b'hmac-sha2-512': sha512,
+ b'hmac-sha2-256': sha256,
+ b'hmac-sha1': sha1,
+ b'hmac-md5': md5,
+ b'none': None
}
@@ -153,8 +158,8 @@
self.encBlockSize = 0
self.decBlockSize = 0
self.verifyDigestSize = 0
- self.outMAC = (None, '', '', 0)
- self.inMAC = (None, '', '', 0)
+ self.outMAC = (None, b'', b'', 0)
+ self.inMAC = (None, b'', b'', 0)
def setKeys(self, outIV, outKey, inIV, inKey, outInteg, inInteg):
@@ -218,7 +223,7 @@
"""
mod = self.macMap[mac]
if not mod:
- return (None, '', '', 0)
+ return (None, b'', b'', 0)
# With stdlib we can only get attributes fron an instantiated object.
hashObject = mod()
@@ -228,9 +233,9 @@
# Truncation here appears to contravene RFC 2104, section 2. However,
# implementing the hashing behavior prescribed by the RFC breaks
# interoperability with OpenSSH (at least version 5.5p1).
- key = key[:digestSize] + ('\x00' * (blockSize - digestSize))
- i = string.translate(key, hmac.trans_36)
- o = string.translate(key, hmac.trans_5C)
+ key = key[:digestSize] + (b'\x00' * (blockSize - digestSize))
+ i = key.translate(hmac.trans_36)
+ o = key.translate(hmac.trans_5C)
result = _MACParams((mod, i, o, digestSize))
result.key = key
return result
@@ -277,7 +282,7 @@
@return: The serialized MAC.
"""
if not self.outMAC[0]:
- return ''
+ return b''
data = struct.pack('>L', seqid) + data
return hmac.HMAC(self.outMAC.key, data, self.outMAC[0]).digest()
@@ -299,7 +304,7 @@
@return: C{True} if the MAC is valid.
"""
if not self.inMAC[0]:
- return mac == ''
+ return mac == b''
data = struct.pack('>L', seqid) + data
outer = hmac.HMAC(self.inMAC.key, data, self.inMAC[0]).digest()
return mac == outer
@@ -314,9 +319,9 @@
@rtype: L{list} of L{str}
"""
supportedCiphers = []
- cs = ['aes256-ctr', 'aes256-cbc', 'aes192-ctr', 'aes192-cbc',
- 'aes128-ctr', 'aes128-cbc', 'cast128-ctr', 'cast128-cbc',
- 'blowfish-ctr', 'blowfish-cbc', '3des-ctr', '3des-cbc']
+ cs = [b'aes256-ctr', b'aes256-cbc', b'aes192-ctr', b'aes192-cbc',
+ b'aes128-ctr', b'aes128-cbc', b'cast128-ctr', b'cast128-cbc',
+ b'blowfish-ctr', b'blowfish-cbc', b'3des-ctr', b'3des-cbc']
for cipher in cs:
algorithmClass, keySize, modeClass = SSHCiphers.cipherMap[cipher]
try:
@@ -447,10 +452,10 @@
to send them while a key exchange is in progress. When the key
exchange completes, another attempt is made to send these messages.
"""
- protocolVersion = '2.0'
- version = 'Twisted'
- comment = ''
- ourVersionString = ('SSH-' + protocolVersion + '-' + version + ' '
+ protocolVersion = b'2.0'
+ version = b'Twisted'
+ comment = b''
+ ourVersionString = (b'SSH-' + protocolVersion + b'-' + version + b' '
+ comment).strip()
# C{none} is supported as cipher and hmac. For security they are disabled
@@ -459,21 +464,21 @@
# List ordered by preference.
supportedCiphers = _getSupportedCiphers()
supportedMACs = [
- 'hmac-sha2-512',
- 'hmac-sha2-256',
- 'hmac-sha1',
- 'hmac-md5',
+ b'hmac-sha2-512',
+ b'hmac-sha2-256',
+ b'hmac-sha1',
+ b'hmac-md5',
# `none`,
- ]
+ ]
supportedKeyExchanges = _kex.getSupportedKeyExchanges()
- supportedPublicKeys = ['ssh-rsa', 'ssh-dss']
- supportedCompressions = ['none', 'zlib']
+ supportedPublicKeys = [b'ssh-rsa', b'ssh-dss']
+ supportedCompressions = [b'none', b'zlib']
supportedLanguages = ()
- supportedVersions = ('1.99', '2.0')
+ supportedVersions = (b'1.99', b'2.0')
isClient = False
gotVersion = False
- buf = ''
+ buf = b''
outgoingPacketSequence = 0
incomingPacketSequence = 0
outgoingCompression = None
@@ -518,9 +523,9 @@
Called when the connection is made to the other side. We sent our
version and the MSG_KEXINIT packet.
"""
- self.transport.write('%s\r\n' % (self.ourVersionString,))
- self.currentEncryptions = SSHCiphers('none', 'none', 'none', 'none')
- self.currentEncryptions.setKeys('', '', '', '', '', '')
+ self.transport.write(self.ourVersionString + b'\r\n')
+ self.currentEncryptions = SSHCiphers(b'none', b'none', b'none', b'none')
+ self.currentEncryptions.setKeys(b'', b'', b'', b'', b'', b'')
self.sendKexInit()
@@ -541,17 +546,17 @@
self.ourKexInitPayload = (chr(MSG_KEXINIT) +
randbytes.secureRandom(16) +
- NS(','.join(self.supportedKeyExchanges)) +
- NS(','.join(self.supportedPublicKeys)) +
- NS(','.join(self.supportedCiphers)) +
- NS(','.join(self.supportedCiphers)) +
- NS(','.join(self.supportedMACs)) +
- NS(','.join(self.supportedMACs)) +
- NS(','.join(self.supportedCompressions)) +
- NS(','.join(self.supportedCompressions)) +
- NS(','.join(self.supportedLanguages)) +
- NS(','.join(self.supportedLanguages)) +
- '\000' + '\000\000\000\000')
+ NS(b','.join(self.supportedKeyExchanges)) +
+ NS(b','.join(self.supportedPublicKeys)) +
+ NS(b','.join(self.supportedCiphers)) +
+ NS(b','.join(self.supportedCiphers)) +
+ NS(b','.join(self.supportedMACs)) +
+ NS(b','.join(self.supportedMACs)) +
+ NS(b','.join(self.supportedCompressions)) +
+ NS(b','.join(self.supportedCompressions)) +
+ NS(b','.join(self.supportedLanguages)) +
+ NS(b','.join(self.supportedLanguages)) +
+ b'\000' + b'\000\000\000\000')
self.sendPacket(MSG_KEXINIT, self.ourKexInitPayload[1:])
self._keyExchangeState = self._KEY_EXCHANGE_REQUESTED
self._blockedByKeyExchange = []
@@ -639,7 +644,7 @@
packetLen, paddingLen = struct.unpack('!LB', first[:5])
if packetLen > 1048576: # 1024 ** 2
self.sendDisconnect(DISCONNECT_PROTOCOL_ERROR,
- 'bad packet length %s' % (packetLen,))
+ networkString('bad packet length %s' % (packetLen,)))
return
if len(self.buf) < packetLen + 4 + ms:
# Not enough data for a packet
@@ -648,20 +653,21 @@
if (packetLen + 4) % bs != 0:
self.sendDisconnect(
DISCONNECT_PROTOCOL_ERROR,
- 'bad packet mod (%i%%%i == %i)' % (packetLen + 4, bs,
- (packetLen + 4) % bs))
+ networkString(
+ 'bad packet mod (%i%%%i == %i)' % (packetLen + 4, bs,
+ (packetLen + 4) % bs)))
return
encData, self.buf = self.buf[:4 + packetLen], self.buf[4 + packetLen:]
packet = first + self.currentEncryptions.decrypt(encData[bs:])
if len(packet) != 4 + packetLen:
self.sendDisconnect(DISCONNECT_PROTOCOL_ERROR,
- 'bad decryption')
+ b'bad decryption')
return
if ms:
macData, self.buf = self.buf[:ms], self.buf[ms:]
if not self.currentEncryptions.verify(self.incomingPacketSequence,
packet, macData):
- self.sendDisconnect(DISCONNECT_MAC_ERROR, 'bad MAC')
+ self.sendDisconnect(DISCONNECT_MAC_ERROR, b'bad MAC')
return
payload = packet[5:-paddingLen]
if self.incomingCompression:
@@ -671,7 +677,7 @@
# Tolerate any errors in decompression
log.err()
self.sendDisconnect(DISCONNECT_COMPRESSION_ERROR,
- 'compression error')
+ b'compression error')
return
self.incomingPacketSequence += 1
return payload
@@ -687,7 +693,7 @@
@type remoteVersion: C{str}
"""
self.sendDisconnect(DISCONNECT_PROTOCOL_VERSION_NOT_SUPPORTED,
- 'bad version ' + remoteVersion)
+ b'bad version ' + remoteVersion)
def dataReceived(self, data):
@@ -701,22 +707,22 @@
"""
self.buf = self.buf + data
if not self.gotVersion:
- if self.buf.find('\n', self.buf.find('SSH-')) == -1:
+ if self.buf.find(b'\n', self.buf.find(b'SSH-')) == -1:
return
- lines = self.buf.split('\n')
+ lines = self.buf.split(b'\n')
for p in lines:
- if p.startswith('SSH-'):
+ if p.startswith(b'SSH-'):
self.gotVersion = True
self.otherVersionString = p.strip()
- remoteVersion = p.split('-')[1]
+ remoteVersion = p.split(b'-')[1]
if remoteVersion not in self.supportedVersions:
self._unsupportedVersionReceived(remoteVersion)
return
i = lines.index(p)
- self.buf = '\n'.join(lines[i + 1:])
+ self.buf = b'\n'.join(lines[i + 1:])
packet = self.getPacket()
while packet:
- messageNum = ord(packet[0])
+ messageNum = ord(packet[0:1])
self.dispatchMessage(messageNum, packet[1:])
packet = self.getPacket()
@@ -785,12 +791,13 @@
def kexAlg(self, value):
"""
Set the key exchange algorithm name.
-
- @raises ConchError: if the key exchange algorithm is not found.
"""
- # Check for supportedness.
- _kex.getKex(value)
- self._kexAlg = value
+ try:
+ # Check for supportedness.
+ _kex.getKex(value)
+ self._kexAlg = value
+ except ConchError:
+ self._kexAlg = None
# Client-initiated rekeying looks like this:
#
@@ -835,42 +842,50 @@
algorithms, and unhandled data, or C{None} if something went wrong.
"""
self.otherKexInitPayload = chr(MSG_KEXINIT) + packet
+
# This is useless to us:
# cookie = packet[: 16]
k = getNS(packet[16:], 10)
strings, rest = k[:-1], k[-1]
(kexAlgs, keyAlgs, encCS, encSC, macCS, macSC, compCS, compSC, langCS,
- langSC) = [s.split(',') for s in strings]
+ langSC) = [s.split(b',') for s in strings]
# These are the server directions
outs = [encSC, macSC, compSC]
ins = [encCS, macSC, compCS]
+
if self.isClient:
outs, ins = ins, outs # Switch directions
+
server = (self.supportedKeyExchanges, self.supportedPublicKeys,
- self.supportedCiphers, self.supportedCiphers,
- self.supportedMACs, self.supportedMACs,
- self.supportedCompressions, self.supportedCompressions)
+ self.supportedCiphers, self.supportedCiphers,
+ self.supportedMACs, self.supportedMACs,
+ self.supportedCompressions, self.supportedCompressions)
+
client = (kexAlgs, keyAlgs, outs[0], ins[0], outs[1], ins[1],
outs[2], ins[2])
+
if self.isClient:
server, client = client, server
+
self.kexAlg = ffs(client[0], server[0])
self.keyAlg = ffs(client[1], server[1])
+
self.nextEncryptions = SSHCiphers(
ffs(client[2], server[2]),
ffs(client[3], server[3]),
ffs(client[4], server[4]),
ffs(client[5], server[5]))
+
self.outgoingCompressionType = ffs(client[6], server[6])
self.incomingCompressionType = ffs(client[7], server[7])
if None in (self.kexAlg, self.keyAlg, self.outgoingCompressionType,
self.incomingCompressionType):
self.sendDisconnect(DISCONNECT_KEY_EXCHANGE_FAILED,
- "couldn't match all kex parts")
+ b"couldn't match all kex parts")
return
if None in self.nextEncryptions.__dict__.values():
self.sendDisconnect(DISCONNECT_KEY_EXCHANGE_FAILED,
- "couldn't match all kex parts")
+ b"couldn't match all kex parts")
return
log.msg('kex alg, key alg: %s %s' % (self.kexAlg, self.keyAlg))
log.msg('outgoing: %s %s %s' % (self.nextEncryptions.outCipType,
@@ -963,7 +978,7 @@
self.service.serviceStarted()
- def sendDebug(self, message, alwaysDisplay=False, language=''):
+ def sendDebug(self, message, alwaysDisplay=False, language=b''):
"""
Send a debug message to the other side.
@@ -1011,7 +1026,7 @@
@type desc: C{str}
"""
self.sendPacket(
- MSG_DISCONNECT, struct.pack('>L', reason) + NS(desc) + NS(''))
+ MSG_DISCONNECT, struct.pack('>L', reason) + NS(desc) + NS(b''))
log.msg('Disconnecting with error, code %s\nreason: %s' % (reason,
desc))
self.transport.loseConnection()
@@ -1054,12 +1069,12 @@
"""
if not self.sessionID:
self.sessionID = exchangeHash
- initIVCS = self._getKey('A', sharedSecret, exchangeHash)
- initIVSC = self._getKey('B', sharedSecret, exchangeHash)
- encKeyCS = self._getKey('C', sharedSecret, exchangeHash)
- encKeySC = self._getKey('D', sharedSecret, exchangeHash)
- integKeyCS = self._getKey('E', sharedSecret, exchangeHash)
- integKeySC = self._getKey('F', sharedSecret, exchangeHash)
+ initIVCS = self._getKey(b'A', sharedSecret, exchangeHash)
+ initIVSC = self._getKey(b'B', sharedSecret, exchangeHash)
+ encKeyCS = self._getKey(b'C', sharedSecret, exchangeHash)
+ encKeySC = self._getKey(b'D', sharedSecret, exchangeHash)
+ integKeyCS = self._getKey(b'E', sharedSecret, exchangeHash)
+ integKeySC = self._getKey(b'F', sharedSecret, exchangeHash)
outs = [initIVSC, encKeySC, integKeySC]
ins = [initIVCS, encKeyCS, integKeyCS]
if self.isClient: # Reverse for the client
@@ -1067,7 +1082,7 @@
outs, ins = ins, outs
self.nextEncryptions.setKeys(outs[0], outs[1], ins[0], ins[1],
outs[2], ins[2])
- self.sendPacket(MSG_NEWKEYS, '')
+ self.sendPacket(MSG_NEWKEYS, b'')
def _newKeys(self):
@@ -1079,9 +1094,9 @@
"""
log.msg('NEW KEYS')
self.currentEncryptions = self.nextEncryptions
- if self.outgoingCompressionType == 'zlib':
+ if self.outgoingCompressionType == b'zlib':
self.outgoingCompression = zlib.compressobj(6)
- if self.incomingCompressionType == 'zlib':
+ if self.incomingCompressionType == b'zlib':
self.incomingCompression = zlib.decompressobj()
self._keyExchangeState = self._KEY_EXCHANGE_NONE
@@ -1102,9 +1117,9 @@
@return: C{True} if it is encrypted.
"""
if direction == "out":
- return self.currentEncryptions.outCipType != 'none'
+ return self.currentEncryptions.outCipType != b'none'
elif direction == "in":
- return self.currentEncryptions.inCipType != 'none'
+ return self.currentEncryptions.inCipType != b'none'
elif direction == "both":
return self.isEncrypted("in") and self.isEncrypted("out")
else:
@@ -1122,9 +1137,9 @@
@return: C{True} if it is verified.
"""
if direction == "out":
- return self.currentEncryptions.outMACType != 'none'
+ return self.currentEncryptions.outMACType != b'none'
elif direction == "in":
- return self.currentEncryptions.inMACType != 'none'
+ return self.currentEncryptions.inMACType != b'none'
elif direction == "both":
return self.isVerified("in") and self.isVerified("out")
else:
@@ -1137,7 +1152,7 @@
DISCONNECT_CONNECTION_LOST message.
"""
self.sendDisconnect(DISCONNECT_CONNECTION_LOST,
- "user closed connection")
+ b"user closed connection")
# Client methods
@@ -1220,7 +1235,7 @@
return
else:
kexAlgs, keyAlgs, rest = retval
- if ord(rest[0]): # Flag first_kex_packet_follows?
+ if ord(rest[0:1]): # Flag first_kex_packet_follows?
if (kexAlgs[0] != self.supportedKeyExchanges[0] or
keyAlgs[0] != self.supportedPublicKeys[0]):
self.ignoreNextPacket = True # Guess was wrong
@@ -1380,9 +1395,9 @@
@type packet: L{bytes}
@param packet: The message data.
"""
- if packet != '':
+ if packet != b'':
self.sendDisconnect(DISCONNECT_PROTOCOL_ERROR,
- "NEWKEYS takes no data")
+ b"NEWKEYS takes no data")
return
self._newKeys()
@@ -1400,10 +1415,10 @@
@param packet: The message data.
"""
service, rest = getNS(packet)
- cls = self.factory.getService(self, service)
+ cls = self.factory.getService(self, nativeString(service))
if not cls:
self.sendDisconnect(DISCONNECT_SERVICE_NOT_AVAILABLE,
- "don't have service %s" % service)
+ b"don't have service " + service)
return
else:
self.sendPacket(MSG_SERVICE_ACCEPT, NS(service))
@@ -1522,13 +1537,13 @@
pubKey, packet = getNS(packet)
f, packet = getMP(packet)
signature, packet = getNS(packet)
- fingerprint = ':'.join([ch.encode('hex') for ch in
- md5(pubKey).digest()])
+ fingerprint = b':'.join([binascii.hexlify(ch) for ch in
+ iterbytes(md5(pubKey).digest())])
d = self.verifyHostKey(pubKey, fingerprint)
d.addCallback(self._continueKEXDH_REPLY, pubKey, f, signature)
d.addErrback(
lambda unused: self.sendDisconnect(
- DISCONNECT_HOST_KEY_NOT_VERIFIABLE, 'bad host key'))
+ DISCONNECT_HOST_KEY_NOT_VERIFIABLE, b'bad host key'))
return d
@@ -1585,7 +1600,7 @@
exchangeHash = h.digest()
if not serverKey.verify(signature, exchangeHash):
self.sendDisconnect(DISCONNECT_KEY_EXCHANGE_FAILED,
- 'bad signature')
+ b'bad signature')
return
self._keySetup(sharedSecret, exchangeHash)
@@ -1607,13 +1622,13 @@
pubKey, packet = getNS(packet)
f, packet = getMP(packet)
signature, packet = getNS(packet)
- fingerprint = ':'.join(map(lambda c: '%02x' % (ord(c),),
- md5(pubKey).digest()))
+ fingerprint = networkString(':'.join(map(lambda c: '%02x' % (ord(c),),
+ iterbytes(md5(pubKey).digest()))))
d = self.verifyHostKey(pubKey, fingerprint)
d.addCallback(self._continueGEX_REPLY, pubKey, f, signature)
d.addErrback(
lambda unused: self.sendDisconnect(
- DISCONNECT_HOST_KEY_NOT_VERIFIABLE, 'bad host key'))
+ DISCONNECT_HOST_KEY_NOT_VERIFIABLE, b'bad host key'))
return d
@@ -1653,7 +1668,7 @@
exchangeHash = h.digest()
if not serverKey.verify(signature, exchangeHash):
self.sendDisconnect(DISCONNECT_KEY_EXCHANGE_FAILED,
- 'bad signature')
+ b'bad signature')
return
self._keySetup(sharedSecret, exchangeHash)
@@ -1664,7 +1679,7 @@
"""
SSHTransportBase._keySetup(self, sharedSecret, exchangeHash)
if self._gotNewKeys:
- self.ssh_NEWKEYS('')
+ self.ssh_NEWKEYS(b'')
def ssh_NEWKEYS(self, packet):
@@ -1676,9 +1691,9 @@
@type packet: L{bytes}
@param packet: The message data.
"""
- if packet != '':
+ if packet != b'':
self.sendDisconnect(DISCONNECT_PROTOCOL_ERROR,
- "NEWKEYS takes no data")
+ b"NEWKEYS takes no data")
return
if not self.nextEncryptions.encBlockSize:
self._gotNewKeys = 1
@@ -1697,14 +1712,14 @@
@type packet: L{bytes}
@param packet: The message data.
"""
- if packet == '':
+ if packet == b'':
log.msg('got SERVICE_ACCEPT without payload')
else:
name = getNS(packet)[0]
if name != self.instance.name:
self.sendDisconnect(
DISCONNECT_PROTOCOL_ERROR,
- "received accept for service we did not request")
+ b"received accept for service we did not request")
self.setService(self.instance)
@@ -1715,7 +1730,7 @@
@type instance: subclass of L{twisted.conch.ssh.service.SSHService}
@param instance: The service to run.
"""
- self.sendPacket(MSG_SERVICE_REQUEST, NS(instance.name))
+ self.sendPacket(MSG_SERVICE_REQUEST, NS(networkString(instance.name)))
self.instance = instance
# Client methods
@@ -1802,7 +1817,7 @@
DH_GENERATOR, DH_PRIME = _kex.getDHGeneratorAndPrime(
- 'diffie-hellman-group1-sha1')
+ b'diffie-hellman-group1-sha1')
MSG_DISCONNECT = 1
@@ -1842,7 +1857,7 @@
messages = {}
-for name, value in globals().items():
+for name, value in items(globals()):
# Avoid legacy messages which overlap with never ones
if name.startswith('MSG_') and not name.startswith('MSG_KEXDH_'):
messages[value] = name
Modified: branches/conch-transport-py3-8232/twisted/conch/test/test_transport.py
==============================================================================
--- branches/conch-transport-py3-8232/twisted/conch/test/test_transport.py (original)
+++ branches/conch-transport-py3-8232/twisted/conch/test/test_transport.py Wed Mar 16 01:47:54 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
@@ -42,6 +45,7 @@
from twisted.protocols import loopback
from twisted.python import randbytes
from twisted.python.randbytes import insecureRandom
+from twisted.python.compat import _bytesChr as chr, iterbytes
from twisted.conch.ssh import address, service, common, _kex
from twisted.test import proto_helpers
@@ -125,17 +129,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 +207,7 @@
def flush(self, kind):
- return '\x66'
+ return b'\x66'
@@ -258,8 +262,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 +271,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 +290,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 +349,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 +364,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 +394,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 +404,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 +425,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 +444,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 +462,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 +479,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 +487,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 +508,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 +527,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 +535,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 +555,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 +571,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 +593,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 +610,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 +641,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 +724,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 +735,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 +745,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 +758,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 +767,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 +778,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 +790,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 +799,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 +814,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 +832,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 +862,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 +899,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 +919,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'))
@@ -935,7 +939,7 @@
self.transport.loseConnection = stubLoseConnection
self.proto.loseConnection()
self.assertEqual(self.packets[0][0], transport.MSG_DISCONNECT)
- self.assertEqual(self.packets[0][1][3],
+ self.assertEqual(self.packets[0][1][3:4],
chr(transport.DISCONNECT_CONNECTION_LOST))
@@ -950,16 +954,16 @@
def stubLoseConnection():
disconnected[0] = True
self.transport.loseConnection = stubLoseConnection
- for c in version + '\r\n':
+ for c in iterbytes(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],
+ self.packets[0][1][3:4],
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,10 +972,10 @@
"""
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.dataReceived(c) for c in data]
+""" + proto.ourVersionString + b"\r\n")
+ [proto.dataReceived(c) for c in iterbytes(data)]
self.assertTrue(proto.gotVersion)
self.assertEqual(proto.otherVersionString, proto.ourVersionString)
@@ -983,9 +987,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 +998,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 +1010,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):
@@ -1023,22 +1027,22 @@
self.assertEqual(self.proto.getPacket(), None)
self.assertEqual(len(self.packets), 1)
self.assertEqual(self.packets[0][0], transport.MSG_DISCONNECT)
- self.assertEqual(self.packets[0][1][3], chr(error))
+ self.assertEqual(self.packets[0][1][3:4], 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()
@@ -1052,21 +1056,21 @@
def checkUnimplemented(seqnum=seqnum):
self.assertEqual(self.packets[0][0],
transport.MSG_UNIMPLEMENTED)
- self.assertEqual(self.packets[0][1][3], chr(seqnum))
+ self.assertEqual(self.packets[0][1][3:4], chr(seqnum))
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 +1086,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 +1109,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)
@@ -1145,7 +1149,7 @@
if kind is None:
kind = transport.DISCONNECT_PROTOCOL_ERROR
self.assertEqual(self.packets[-1][0], transport.MSG_DISCONNECT)
- self.assertEqual(self.packets[-1][1][3], chr(kind))
+ self.assertEqual(self.packets[-1][1][3:4], chr(kind))
def connectModifiedProtocol(self, protoModification,
@@ -1262,35 +1266,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 +1305,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 +1318,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 +1340,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 +1353,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 +1378,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 +1414,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 +1422,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 iterbytes(b'ABCDEF')]
self.assertEqual(
self.proto.nextEncryptions.keys,
(newKeys[1], newKeys[3], newKeys[0], newKeys[2], newKeys[5],
@@ -1446,20 +1451,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 +1473,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')
@@ -1487,7 +1492,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 +1509,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 +1538,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 +1559,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 +1577,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 +1591,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 +1609,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 +1647,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 +1672,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 +1730,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 +1743,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,18 +1751,7 @@
KEXINIT messages requesting diffie-hellman-group1-sha1 result in
KEXDH_INIT responses.
"""
- self.assertKexInitResponseForDH('diffie-hellman-group1-sha1')
-
-
- def test_KEXINIT_badKexAlg(self):
- """
- Test that the client raises a ConchError if it receives a
- 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.assertRaises(ConchError, self.proto.dataReceived, data)
+ self.assertKexInitResponseForDH(b'diffie-hellman-group1-sha1')
def test_KEXDH_REPLY(self):
@@ -1773,7 +1767,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 +1779,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 +1790,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 iterbytes(b'ABCDEF')]
self.assertEqual(self.proto.nextEncryptions.keys,
(newKeys[0], newKeys[2], newKeys[1], newKeys[3],
newKeys[4], newKeys[5]))
@@ -1821,25 +1815,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 +1842,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 +1852,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 +1860,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 +1868,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 +1878,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 +1889,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 +1910,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 +1920,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 +1942,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 +1958,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 +1969,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,9 +2037,9 @@
params = self.ciphers._getMAC(hmacName, secret)
- key = secret[:digestSize] + '\x00' * blockPadSize
- innerPad = b''.join(chr(ord(b) ^ 0x36) for b in key)
- outerPad = b''.join(chr(ord(b) ^ 0x5c) for b in key)
+ key = secret[:digestSize] + b'\x00' * blockPadSize
+ innerPad = b''.join(chr(ord(b) ^ 0x36) for b in iterbytes(key))
+ outerPad = b''.join(chr(ord(b) ^ 0x5c) for b in iterbytes(key))
self.assertEqual(
(hashProcessor, innerPad, outerPad, digestSize), params)
self.assertEqual(key, params.key)
@@ -2117,22 +2111,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 +2136,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 +2160,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 +2175,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 +2200,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 +2249,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 +2274,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 +2287,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