dorian
/
paramiko
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Convert and detect types properly, use helper constants, use StringIO and range

This commit is contained in:
Scott Maxwell 2013-10-30 17:14:52 -07:00
parent 2d738fa08b
commit 0b7d0cf0a2
25 changed files with 154 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
paramiko/ber.py
View File

@ -49,13 +49,13 @@ class BER(object):
def decode_next(self):
if self.idx >= len(self.content):
return None
ident = ord(self.content[self.idx])
ident = byte_ord(self.content[self.idx])
self.idx += 1
if (ident & 31) == 31:
# identifier > 30
ident = 0
while self.idx < len(self.content):
t = ord(self.content[self.idx])
t = byte_ord(self.content[self.idx])
self.idx += 1
ident = (ident << 7) | (t & 0x7f)
if not (t & 0x80):
@ -63,7 +63,7 @@ class BER(object):
if self.idx >= len(self.content):
return None
# now fetch length
size = ord(self.content[self.idx])
size = byte_ord(self.content[self.idx])
self.idx += 1
if size & 0x80:
# more complimicated...
@ -102,12 +102,12 @@ class BER(object):
def encode_tlv(self, ident, val):
# no need to support ident > 31 here
self.content += chr(ident)
self.content += byte_chr(ident)
if len(val) > 0x7f:
lenstr = util.deflate_long(len(val))
self.content += chr(0x80 + len(lenstr)) + lenstr
self.content += byte_chr(0x80 + len(lenstr)) + lenstr
else:
self.content += chr(len(val))
self.content += byte_chr(len(val))
self.content += val
def encode(self, x):

2
paramiko/channel.py
View File

@ -969,7 +969,7 @@ class Channel (object):
self.transport._send_user_message(m)
def _feed(self, m):
if type(m) is str:
if isinstance(m, bytes_type):
# passed from _feed_extended
s = m
else:

2
paramiko/client.py
View File

@ -335,7 +335,7 @@ class SSHClient (object):
if key_filename is None:
key_filenames = []
elif isinstance(key_filename, (str, unicode)):
elif isinstance(key_filename, string_types):
key_filenames = [ key_filename ]
else:
key_filenames = key_filename

4
paramiko/kex_gex.py
View File

@ -96,15 +96,15 @@ class KexGex (object):
# generate an "x" (1 < x < (p-1)/2).
q = (self.p - 1) // 2
qnorm = util.deflate_long(q, 0)
qhbyte = ord(qnorm[0])
qhbyte = byte_ord(qnorm[0])
byte_count = len(qnorm)
qmask = 0xff
while not (qhbyte & 0x80):
qhbyte <<= 1
qmask >>= 1
while True:
x_bytes = chr(ord(x_bytes[0]) & qmask) + x_bytes[1:]
x_bytes = self.transport.rng.read(byte_count)
x_bytes = byte_mask(x_bytes[0], qmask) + x_bytes[1:]
x = util.inflate_long(x_bytes, 1)
if (x > 1) and (x < q):
break

11
paramiko/kex_group1.py
View File

@ -30,9 +30,10 @@ from paramiko.ssh_exception import SSHException
_MSG_KEXDH_INIT, _MSG_KEXDH_REPLY = range(30, 32)
c_MSG_KEXDH_INIT, c_MSG_KEXDH_REPLY = [byte_chr(c) for c in range(30, 32)]
# draft-ietf-secsh-transport-09.txt, page 17
P = 0xFFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFFL
P = 0xFFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF
G = 2
@ -42,9 +43,9 @@ class KexGroup1(object):
def __init__(self, transport):
self.transport = transport
self.x = 0L
self.e = 0L
self.f = 0L
self.x = long_zero
self.e = long_zero
self.f = long_zero
def start_kex(self):
self._generate_x()
@ -80,7 +81,7 @@ class KexGroup1(object):
# larger than q (but this is a tiny tiny subset of potential x).
while 1:
x_bytes = self.transport.rng.read(128)
x_bytes = chr(ord(x_bytes[0]) & 0x7f) + x_bytes[1:]
x_bytes = byte_mask(x_bytes[0], 0x7f) + x_bytes[1:]
if (x_bytes[:8] != '\x7F\xFF\xFF\xFF\xFF\xFF\xFF\xFF') and \
(x_bytes[:8] != '\x00\x00\x00\x00\x00\x00\x00\x00'):
break

31
paramiko/packet.py
View File

@ -38,6 +38,7 @@ try:
except ImportError:
from Crypto.Hash.HMAC import HMAC
def compute_hmac(key, message, digest_class):
return HMAC(key, message, digest_class).digest()
@ -66,7 +67,7 @@ class Packetizer (object):
self.__dump_packets = False
self.__need_rekey = False
self.__init_count = 0
self.__remainder = ''
self.__remainder = bytes()
# used for noticing when to re-key:
self.__sent_bytes = 0
@ -90,8 +91,8 @@ class Packetizer (object):
self.__mac_key_in = ''
self.__compress_engine_out = None
self.__compress_engine_in = None
self.__sequence_number_out = 0L
self.__sequence_number_in = 0L
self.__sequence_number_out = long_zero
self.__sequence_number_in = long_zero
# lock around outbound writes (packet computation)
self.__write_lock = threading.RLock()
@ -196,7 +197,7 @@ class Packetizer (object):
@raise EOFError: if the socket was closed before all the bytes could
be read
"""
out = ''
out = bytes()
# handle over-reading from reading the banner line
if len(self.__remainder) > 0:
out = self.__remainder[:n]
@ -275,22 +276,22 @@ class Packetizer (object):
line, so it's okay to attempt large reads.
"""
buf = self.__remainder
while not '\n' in buf:
while not newline_byte in buf:
buf += self._read_timeout(timeout)
n = buf.index('\n')
n = buf.index(newline_byte)
self.__remainder = buf[n+1:]
buf = buf[:n]
if (len(buf) > 0) and (buf[-1] == '\r'):
if (len(buf) > 0) and (buf[-1] == cr_byte):
buf = buf[:-1]
return buf
return u(buf)
def send_message(self, data):
"""
Write a block of data using the current cipher, as an SSH block.
"""
# encrypt this sucka
data = str(data)
cmd = ord(data[0])
data = asbytes(data)
cmd = byte_ord(data[0])
if cmd in MSG_NAMES:
cmd_name = MSG_NAMES[cmd]
else:
@ -312,7 +313,7 @@ class Packetizer (object):
if self.__block_engine_out != None:
payload = struct.pack('>I', self.__sequence_number_out) + packet
out += compute_hmac(self.__mac_key_out, payload, self.__mac_engine_out)[:self.__mac_size_out]
self.__sequence_number_out = (self.__sequence_number_out + 1) & 0xffffffffL
self.__sequence_number_out = (self.__sequence_number_out + 1) & xffffffff
self.write_all(out)
self.__sent_bytes += len(out)
@ -361,7 +362,7 @@ class Packetizer (object):
my_mac = compute_hmac(self.__mac_key_in, mac_payload, self.__mac_engine_in)[:self.__mac_size_in]
if my_mac != mac:
raise SSHException('Mismatched MAC')
padding = ord(packet[0])
padding = byte_ord(packet[0])
payload = packet[1:packet_size - padding]
if self.__dump_packets:
@ -372,7 +373,7 @@ class Packetizer (object):
msg = Message(payload[1:])
msg.seqno = self.__sequence_number_in
self.__sequence_number_in = (self.__sequence_number_in + 1) & 0xffffffffL
self.__sequence_number_in = (self.__sequence_number_in + 1) & xffffffff
# check for rekey
raw_packet_size = packet_size + self.__mac_size_in + 4
@ -395,7 +396,7 @@ class Packetizer (object):
self.__received_packets_overflow = 0
self._trigger_rekey()
cmd = ord(payload[0])
cmd = byte_ord(payload[0])
if cmd in MSG_NAMES:
cmd_name = MSG_NAMES[cmd]
else:
@ -493,7 +494,7 @@ class Packetizer (object):
if self.__sdctr_out or self.__block_engine_out is None:
# cute trick i caught openssh doing: if we're not encrypting or SDCTR mode (RFC4344),
# don't waste random bytes for the padding
packet += (chr(0) * padding)
packet += (zero_byte * padding)
else:
packet += rng.read(padding)
return packet

4
paramiko/pipe.py
View File

@ -64,7 +64,7 @@ class PosixPipe (object):
if self._set or self._closed:
return
self._set = True
os.write(self._wfd, '*')
os.write(self._wfd, b('*'))
def set_forever (self):
self._forever = True
@ -110,7 +110,7 @@ class WindowsPipe (object):
if self._set or self._closed:
return
self._set = True
self._wsock.send('*')
self._wsock.send(b('*'))
def set_forever (self):
self._forever = True

4
paramiko/pkey.py
View File

@ -352,9 +352,9 @@ class PKey (object):
@raise IOError: if there was an error writing the file.
"""
f = open(filename, 'w', 0600)
f = open(filename, 'w', o600)
# grrr... the mode doesn't always take hold
os.chmod(filename, 0600)
os.chmod(filename, o600)
self._write_private_key(tag, f, data, password)
f.close()

4
paramiko/primes.py
View File

@ -34,7 +34,7 @@ def _generate_prime(bits, rng):
# loop catches the case where we increment n into a higher bit-range
x = rng.read((bits+7) // 8)
if hbyte_mask > 0:
x = chr(ord(x[0]) & hbyte_mask) + x[1:]
x = byte_mask(x[0], hbyte_mask) + x[1:]
n = util.inflate_long(x, 1)
n |= 1
n |= (1 << (bits - 1))
@ -59,7 +59,7 @@ def _roll_random(rng, n):
while True:
x = rng.read(byte_count)
if hbyte_mask > 0:
x = chr(ord(x[0]) & hbyte_mask) + x[1:]
x = byte_mask(x[0], hbyte_mask) + x[1:]
num = util.inflate_long(x, 1)
if num < n:
break

2
paramiko/server.py
View File

@ -48,7 +48,7 @@ class InteractiveQuery (object):
self.instructions = instructions
self.prompts = []
for x in prompts:
if (type(x) is str) or (type(x) is unicode):
if isinstance(x, string_types):
self.add_prompt(x)
else:
self.add_prompt(x[0], x[1])

4
paramiko/sftp.py
View File

@ -166,8 +166,8 @@ class BaseSFTP (object):
def _send_packet(self, t, packet):
#self._log(DEBUG2, 'write: %s (len=%d)' % (CMD_NAMES.get(t, '0x%02x' % t), len(packet)))
out = struct.pack('>I', len(packet) + 1) + chr(t) + packet
packet = asbytes(packet)
out = struct.pack('>I', len(packet) + 1) + byte_chr(t) + packet
if self.ultra_debug:
self._log(DEBUG, util.format_binary(out, 'OUT: '))
self._write_all(out)
@ -183,7 +183,7 @@ class BaseSFTP (object):
if self.ultra_debug:
self._log(DEBUG, util.format_binary(data, 'IN: '));
if size > 0:
t = ord(data[0])
t = byte_ord(data[0])
#self._log(DEBUG2, 'read: %s (len=%d)' % (CMD_NAMES.get(t), '0x%02x' % t, len(data)-1))
return t, data[1:]
return 0, ''

8
paramiko/sftp_attr.py
View File

@ -44,7 +44,7 @@ class SFTPAttributes (object):
FLAG_UIDGID = 2
FLAG_PERMISSIONS = 4
FLAG_AMTIME = 8
FLAG_EXTENDED = 0x80000000L
FLAG_EXTENDED = x80000000
def __init__(self):
"""
@ -194,13 +194,13 @@ class SFTPAttributes (object):
ks = 's'
else:
ks = '?'
ks += self._rwx((self.st_mode & 0700) >> 6, self.st_mode & stat.S_ISUID)
ks += self._rwx((self.st_mode & 070) >> 3, self.st_mode & stat.S_ISGID)
ks += self._rwx((self.st_mode & o700) >> 6, self.st_mode & stat.S_ISUID)
ks += self._rwx((self.st_mode & o70) >> 3, self.st_mode & stat.S_ISGID)
ks += self._rwx(self.st_mode & 7, self.st_mode & stat.S_ISVTX, True)
else:
ks = '?---------'
# compute display date
if (self.st_mtime is None) or (self.st_mtime == 0xffffffffL):
if (self.st_mtime is None) or (self.st_mtime == xffffffff):
# shouldn't really happen
datestr = '(unknown date)'
else:

4
paramiko/sftp_client.py
View File

@ -285,7 +285,7 @@ class SFTPClient (BaseSFTP):
self._log(DEBUG, 'rename(%r, %r)' % (oldpath, newpath))
self._request(CMD_RENAME, oldpath, newpath)
def mkdir(self, path, mode=0777):
def mkdir(self, path, mode=o777):
"""
Create a folder (directory) named C{path} with numeric mode C{mode}.
The default mode is 0777 (octal). On some systems, mode is ignored.
@ -698,7 +698,7 @@ class SFTPClient (BaseSFTP):
msg.add_int(item)
elif isinstance(item, long):
msg.add_int64(item)
elif isinstance(item, str):
elif isinstance(item, string_types):
msg.add_string(item)
elif isinstance(item, SFTPAttributes):
item._pack(msg)

4
paramiko/sftp_server.py
View File

@ -183,7 +183,7 @@ class SFTPServer (BaseSFTP, SubsystemHandler):
msg.add_int(item)
elif type(item) is long:
msg.add_int64(item)
elif type(item) is str:
elif isinstance(item, string_types):
msg.add_string(item)
elif type(item) is SFTPAttributes:
item._pack(msg)
@ -420,7 +420,7 @@ class SFTPServer (BaseSFTP, SubsystemHandler):
elif t == CMD_READLINK:
path = msg.get_text()
resp = self.server.readlink(path)
if type(resp) is str:
if isinstance(resp, string_types):
self._response(request_number, CMD_NAME, 1, resp, '', SFTPAttributes())
else:
self._send_status(request_number, resp)

8
paramiko/transport.py
View File

@ -376,7 +376,7 @@ class Transport (threading.Thread):
@rtype: str
"""
out = '<paramiko.Transport at %s' % hex(long(id(self)) & 0xffffffffL)
out = '<paramiko.Transport at %s' % hex(long(id(self)) & xffffffff)
if not self.active:
out += ' (unconnected)'
else:
@ -1553,12 +1553,12 @@ class Transport (threading.Thread):
# active=True occurs before the thread is launched, to avoid a race
_active_threads.append(self)
if self.server_mode:
self._log(DEBUG, 'starting thread (server mode): %s' % hex(long(id(self)) & 0xffffffffL))
self._log(DEBUG, 'starting thread (server mode): %s' % hex(long(id(self)) & xffffffff))
else:
self._log(DEBUG, 'starting thread (client mode): %s' % hex(long(id(self)) & 0xffffffffL))
self._log(DEBUG, 'starting thread (client mode): %s' % hex(long(id(self)) & xffffffff))
try:
try:
self.packetizer.write_all(self.local_version + '\r\n')
self.packetizer.write_all(b(self.local_version + '\r\n'))
self._check_banner()
self._send_kex_init()
self._expect_packet(MSG_KEXINIT)

61
paramiko/util.py
View File

@ -48,60 +48,53 @@ if sys.version_info < (2,3):
def inflate_long(s, always_positive=False):
"turns a normalized byte string into a long-int (adapted from Crypto.Util.number)"
out = 0L
out = long_zero
negative = 0
if not always_positive and (len(s) > 0) and (ord(s[0]) >= 0x80):
if not always_positive and (len(s) > 0) and (byte_ord(s[0]) >= 0x80):
negative = 1
if len(s) % 4:
filler = '\x00'
filler = zero_byte
if negative:
filler = '\xff'
filler = max_byte
s = filler * (4 - len(s) % 4) + s
for i in range(0, len(s), 4):
out = (out << 32) + struct.unpack('>I', s[i:i+4])[0]
if negative:
out -= (1L << (8 * len(s)))
out -= (long_one << (8 * len(s)))
return out
deflate_zero = 0 if PY3 else zero_byte
deflate_ff = 0xff if PY3 else max_byte
def deflate_long(n, add_sign_padding=True):
"turns a long-int into a normalized byte string (adapted from Crypto.Util.number)"
# after much testing, this algorithm was deemed to be the fastest
s = ''
s = bytes()
n = long(n)
while (n != 0) and (n != -1):
s = struct.pack('>I', n & 0xffffffffL) + s
s = struct.pack('>I', n & xffffffff) + s
n = n >> 32
# strip off leading zeros, FFs
for i in enumerate(s):
if (n == 0) and (i[1] != '\000'):
if (n == 0) and (i[1] != deflate_zero):
break
if (n == -1) and (i[1] != '\xff'):
if (n == -1) and (i[1] != deflate_ff):
break
else:
# degenerate case, n was either 0 or -1
i = (0,)
if n == 0:
s = '\000'
s = zero_byte
else:
s = '\xff'
s = max_byte
s = s[i[0]:]
if add_sign_padding:
if (n == 0) and (ord(s[0]) >= 0x80):
s = '\x00' + s
if (n == -1) and (ord(s[0]) < 0x80):
s = '\xff' + s
if (n == 0) and (byte_ord(s[0]) >= 0x80):
s = zero_byte + s
if (n == -1) and (byte_ord(s[0]) < 0x80):
s = max_byte + s
return s
def format_binary_weird(data):
out = ''
for i in enumerate(data):
out += '%02X' % ord(i[1])
if i[0] % 2:
out += ' '
if i[0] % 16 == 15:
out += '\n'
return out
def format_binary(data, prefix=''):
x = 0
out = []
@ -113,8 +106,8 @@ def format_binary(data, prefix=''):
return [prefix + x for x in out]
def format_binary_line(data):
left = ' '.join(['%02X' % ord(c) for c in data])
right = ''.join([('.%c..' % c)[(ord(c)+63)//95] for c in data])
left = ' '.join(['%02X' % byte_ord(c) for c in data])
right = ''.join([('.%c..' % c)[(byte_ord(c)+63)//95] for c in data])
return '%-50s %s' % (left, right)
def hexify(s):
@ -126,17 +119,17 @@ def unhexify(s):
def safe_string(s):
out = ''
for c in s:
if (ord(c) >= 32) and (ord(c) <= 127):
if (byte_ord(c) >= 32) and (byte_ord(c) <= 127):
out += c
else:
out += '%%%02X' % ord(c)
out += '%%%02X' % byte_ord(c)
return out
# ''.join([['%%%02X' % ord(c), c][(ord(c) >= 32) and (ord(c) <= 127)] for c in s])
def bit_length(n):
norm = deflate_long(n, 0)
hbyte = ord(norm[0])
hbyte = byte_ord(norm[0])
if hbyte == 0:
return 1
bitlen = len(norm) * 8
@ -298,15 +291,15 @@ class Counter (object):
"""Increament the counter and return the new value"""
i = self.blocksize - 1
while i > -1:
c = self.value[i] = chr((ord(self.value[i]) + 1) % 256)
if c != '\x00':
c = self.value[i] = byte_chr((byte_ord(self.value[i]) + 1) % 256)
if c != zero_byte:
return self.value.tostring()
i -= 1
# counter reset
x = deflate_long(self.overflow, add_sign_padding=False)
self.value = array.array('c', '\x00' * (self.blocksize - len(x)) + x)
self.value = array.array('c', zero_byte * (self.blocksize - len(x)) + x)
return self.value.tostring()
def new(cls, nbits, initial_value=1L, overflow=0L):
def new(cls, nbits, initial_value=long_one, overflow=long_zero):
return cls(nbits, initial_value=initial_value, overflow=overflow)
new = classmethod(new)

4
tests/loop.py
View File

@ -32,7 +32,7 @@ class LoopSocket (object):
"""
def __init__(self):
self.__in_buffer = ''
self.__in_buffer = bytes()
self.__lock = threading.Lock()
self.__cv = threading.Condition(self.__lock)
self.__timeout = None
@ -42,7 +42,7 @@ class LoopSocket (object):
self.__unlink()
try:
self.__lock.acquire()
self.__in_buffer = ''
self.__in_buffer = bytes()
finally:
self.__lock.release()

2
tests/stub_sftp.py
View File

@ -104,7 +104,7 @@ class StubSFTPServer (SFTPServerInterface):
else:
# os.open() defaults to 0777 which is
# an odd default mode for files
fd = os.open(path, flags, 0666)
fd = os.open(path, flags, o666)
except OSError:
e = sys.exc_info()[1]
return SFTPServer.convert_errno(e.errno)

9
tests/test_auth.py
View File

@ -32,6 +32,11 @@ from paramiko import OPEN_SUCCEEDED, OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
from tests.loop import LoopSocket
from tests.util import test_path
try:
_pwd = u'\u2022'
except Exception:
_pwd = '\u2022'
class NullServer (ServerInterface):
paranoid_did_password = False
@ -65,7 +70,7 @@ class NullServer (ServerInterface):
if self.paranoid_did_public_key:
return AUTH_SUCCESSFUL
return AUTH_PARTIALLY_SUCCESSFUL
if (username == 'utf8') and (password == u'\u2022'):
if (username == 'utf8') and (password == _pwd):
return AUTH_SUCCESSFUL
if (username == 'non-utf8') and (password == '\xff'):
return AUTH_SUCCESSFUL
@ -203,7 +208,7 @@ class AuthTest (unittest.TestCase):
"""
self.start_server()
self.tc.connect(hostkey=self.public_host_key)
remain = self.tc.auth_password('utf8', u'\u2022')
remain = self.tc.auth_password('utf8', _pwd)
self.assertEquals([], remain)
self.verify_finished()

6
tests/test_kex.py
View File

@ -31,7 +31,7 @@ from paramiko.common import *
class FakeRng (object):
def read(self, n):
return chr(0xcc) * n
return byte_chr(0xcc) * n
class FakeKey (object):
@ -44,7 +44,7 @@ class FakeKey (object):
class FakeModulusPack (object):
P = 0xFFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFFL
P = 0xFFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7EDEE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF
G = 2
def get_modulus(self, min, ask, max):
return self.G, self.P
@ -78,7 +78,7 @@ class FakeTransport (object):
class KexTest (unittest.TestCase):
K = 14730343317708716439807310032871972459448364195094179797249681733965528989482751523943515690110179031004049109375612685505881911274101441415545039654102474376472240501616988799699744135291070488314748284283496055223852115360852283821334858541043710301057312858051901453919067023103730011648890038847384890504L
K = 14730343317708716439807310032871972459448364195094179797249681733965528989482751523943515690110179031004049109375612685505881911274101441415545039654102474376472240501616988799699744135291070488314748284283496055223852115360852283821334858541043710301057312858051901453919067023103730011648890038847384890504
def setUp(self):
pass

6
tests/test_pkey.py
View File

@ -99,7 +99,7 @@ class KeyTest (unittest.TestCase):
self.assertEquals(PUB_RSA.split()[1], key.get_base64())
self.assertEquals(1024, key.get_bits())
s = StringIO.StringIO()
s = StringIO()
key.write_private_key(s)
self.assertEquals(RSA_PRIVATE_OUT, s.getvalue())
s.seek(0)
@ -124,7 +124,7 @@ class KeyTest (unittest.TestCase):
self.assertEquals(PUB_DSS.split()[1], key.get_base64())
self.assertEquals(1024, key.get_bits())
s = StringIO.StringIO()
s = StringIO()
key.write_private_key(s)
self.assertEquals(DSS_PRIVATE_OUT, s.getvalue())
s.seek(0)
@ -207,7 +207,7 @@ class KeyTest (unittest.TestCase):
self.assertEquals(PUB_ECDSA.split()[1], key.get_base64())
self.assertEquals(256, key.get_bits())
s = StringIO.StringIO()
s = StringIO()
key.write_private_key(s)
self.assertEquals(ECDSA_PRIVATE_OUT, s.getvalue())
s.seek(0)

45
tests/test_sftp.py
View File

@ -71,7 +71,10 @@ FOLDER = os.environ.get('TEST_FOLDER', 'temp-testing000')
sftp = None
tc = None
g_big_file_test = True
try:
unicode_folder = u'\u00fcnic\u00f8de'
except SyntaxError:
unicode_folder = '\u00fcnic\u00f8de'
def get_sftp():
global sftp
@ -141,7 +144,7 @@ class SFTPTest (unittest.TestCase):
def setUp(self):
global FOLDER
for i in xrange(1000):
for i in range(1000):
FOLDER = FOLDER[:-3] + '%03d' % i
try:
sftp.mkdir(FOLDER)
@ -236,7 +239,7 @@ class SFTPTest (unittest.TestCase):
pass
f = sftp.open(FOLDER + '/second.txt', 'r')
f.seek(-6, f.SEEK_END)
self.assertEqual(f.read(4), 'tent')
self.assertEqual(u(f.read(4)), 'tent')
f.close()
finally:
try:
@ -301,16 +304,16 @@ class SFTPTest (unittest.TestCase):
f.close()
stat = sftp.stat(FOLDER + '/special')
sftp.chmod(FOLDER + '/special', (stat.st_mode & ~0777) | 0600)
sftp.chmod(FOLDER + '/special', (stat.st_mode & ~o777) | o600)
stat = sftp.stat(FOLDER + '/special')
expected_mode = 0600
expected_mode = o600
if sys.platform == 'win32':
# chmod not really functional on windows
expected_mode = 0666
expected_mode = o666
if sys.platform == 'cygwin':
# even worse.
expected_mode = 0644
self.assertEqual(stat.st_mode & 0777, expected_mode)
expected_mode = o644
self.assertEqual(stat.st_mode & o777, expected_mode)
self.assertEqual(stat.st_size, 1024)
mtime = stat.st_mtime - 3600
@ -341,17 +344,17 @@ class SFTPTest (unittest.TestCase):
f = sftp.open(FOLDER + '/special', 'r+')
stat = f.stat()
f.chmod((stat.st_mode & ~0777) | 0600)
f.chmod((stat.st_mode & ~o777) | o600)
stat = f.stat()
expected_mode = 0600
expected_mode = o600
if sys.platform == 'win32':
# chmod not really functional on windows
expected_mode = 0666
expected_mode = o666
if sys.platform == 'cygwin':
# even worse.
expected_mode = 0644
self.assertEqual(stat.st_mode & 0777, expected_mode)
expected_mode = o644
self.assertEqual(stat.st_mode & o777, expected_mode)
self.assertEqual(stat.st_size, 1024)
mtime = stat.st_mtime - 3600
@ -643,7 +646,7 @@ class SFTPTest (unittest.TestCase):
f.close()
try:
sftp.rename(FOLDER + '/something', FOLDER + u'/\u00fcnic\u00f8de')
sftp.rename(FOLDER + '/something', FOLDER + '/' + unicode_folder)
sftp.open(FOLDER + '/\xc3\xbcnic\xc3\xb8\x64\x65', 'r')
except Exception:
e = sys.exc_info()[1]
@ -651,16 +654,16 @@ class SFTPTest (unittest.TestCase):
sftp.unlink(FOLDER + '/\xc3\xbcnic\xc3\xb8\x64\x65')
def test_L_utf8_chdir(self):
sftp.mkdir(FOLDER + u'\u00fcnic\u00f8de')
sftp.mkdir(FOLDER + '/' + unicode_folder)
try:
sftp.chdir(FOLDER + u'\u00fcnic\u00f8de')
sftp.chdir(FOLDER + '/' + unicode_folder)
f = sftp.open('something', 'w')
f.write('okay')
f.close()
sftp.unlink('something')
finally:
sftp.chdir(None)
sftp.rmdir(FOLDER + u'\u00fcnic\u00f8de')
sftp.rmdir(FOLDER + '/' + unicode_folder)
def test_M_bad_readv(self):
"""
@ -733,10 +736,16 @@ class SFTPTest (unittest.TestCase):
Send an empty file and confirm it is sent.
"""
target = FOLDER + '/empty file.txt'
stream = StringIO.StringIO()
stream = StringIO()
try:
attrs = sftp.putfo(stream, target)
# the returned attributes should not be null
self.assertNotEqual(attrs, None)
finally:
sftp.remove(target)
if __name__ == '__main__':
SFTPTest.init_loopback()
from unittest import main
main()

32
tests/test_sftp_big.py
View File

@ -46,7 +46,7 @@ class BigSFTPTest (unittest.TestCase):
def setUp(self):
global FOLDER
sftp = get_sftp()
for i in xrange(1000):
for i in range(1000):
FOLDER = FOLDER[:-3] + '%03d' % i
try:
sftp.mkdir(FOLDER)
@ -69,7 +69,7 @@ class BigSFTPTest (unittest.TestCase):
f = sftp.open('%s/file%d.txt' % (FOLDER, i), 'w', 1)
f.write('this is file #%d.\n' % i)
f.close()
sftp.chmod('%s/file%d.txt' % (FOLDER, i), 0660)
sftp.chmod('%s/file%d.txt' % (FOLDER, i), o660)
# now make sure every file is there, by creating a list of filenmes
# and reading them in random order.
@ -124,7 +124,7 @@ class BigSFTPTest (unittest.TestCase):
write a 1MB file, with no linefeeds, using pipelining.
"""
sftp = get_sftp()
kblob = ''.join([struct.pack('>H', n) for n in xrange(512)])
kblob = ''.join([struct.pack('>H', n) for n in range(512)])
start = time.time()
try:
f = sftp.open('%s/hongry.txt' % FOLDER, 'w')
@ -165,7 +165,7 @@ class BigSFTPTest (unittest.TestCase):
def test_4_prefetch_seek(self):
sftp = get_sftp()
kblob = ''.join([struct.pack('>H', n) for n in xrange(512)])
kblob = ''.join([struct.pack('>H', n) for n in range(512)])
try:
f = sftp.open('%s/hongry.txt' % FOLDER, 'w')
f.set_pipelined(True)
@ -181,13 +181,13 @@ class BigSFTPTest (unittest.TestCase):
start = time.time()
k2blob = kblob + kblob
chunk = 793
for i in xrange(10):
for i in range(10):
f = sftp.open('%s/hongry.txt' % FOLDER, 'r')
f.prefetch()
base_offset = (512 * 1024) + 17 * random.randint(1000, 2000)
offsets = [base_offset + j * chunk for j in xrange(100)]
offsets = [base_offset + j * chunk for j in range(100)]
# randomly seek around and read them out
for j in xrange(100):
for j in range(100):
offset = offsets[random.randint(0, len(offsets) - 1)]
offsets.remove(offset)
f.seek(offset)
@ -203,7 +203,7 @@ class BigSFTPTest (unittest.TestCase):
def test_5_readv_seek(self):
sftp = get_sftp()
kblob = ''.join([struct.pack('>H', n) for n in xrange(512)])
kblob = ''.join([struct.pack('>H', n) for n in range(512)])
try:
f = sftp.open('%s/hongry.txt' % FOLDER, 'w')
f.set_pipelined(True)
@ -219,21 +219,21 @@ class BigSFTPTest (unittest.TestCase):
start = time.time()
k2blob = kblob + kblob
chunk = 793
for i in xrange(10):
for i in range(10):
f = sftp.open('%s/hongry.txt' % FOLDER, 'r')
base_offset = (512 * 1024) + 17 * random.randint(1000, 2000)
# make a bunch of offsets and put them in random order
offsets = [base_offset + j * chunk for j in xrange(100)]
offsets = [base_offset + j * chunk for j in range(100)]
readv_list = []
for j in xrange(100):
for j in range(100):
o = offsets[random.randint(0, len(offsets) - 1)]
offsets.remove(o)
readv_list.append((o, chunk))
ret = f.readv(readv_list)
for i in xrange(len(readv_list)):
for i in range(len(readv_list)):
offset = readv_list[i][0]
n_offset = offset % 1024
self.assertEqual(ret.next(), k2blob[n_offset:n_offset + chunk])
self.assertEqual(next(ret), k2blob[n_offset:n_offset + chunk])
f.close()
end = time.time()
sys.stderr.write('%ds ' % round(end - start))
@ -279,7 +279,7 @@ class BigSFTPTest (unittest.TestCase):
verify that prefetch and readv don't conflict with each other.
"""
sftp = get_sftp()
kblob = ''.join([struct.pack('>H', n) for n in xrange(512)])
kblob = ''.join([struct.pack('>H', n) for n in range(512)])
try:
f = sftp.open('%s/hongry.txt' % FOLDER, 'w')
f.set_pipelined(True)
@ -318,7 +318,7 @@ class BigSFTPTest (unittest.TestCase):
returned as a single blob.
"""
sftp = get_sftp()
kblob = ''.join([struct.pack('>H', n) for n in xrange(512)])
kblob = ''.join([struct.pack('>H', n) for n in range(512)])
try:
f = sftp.open('%s/hongry.txt' % FOLDER, 'w')
f.set_pipelined(True)
@ -367,7 +367,7 @@ class BigSFTPTest (unittest.TestCase):
k32blob = (32 * 1024 * 'x')
try:
f = sftp.open('%s/hongry.txt' % FOLDER, 'w', 128 * 1024)
for i in xrange(32):
for i in range(32):
f.write(k32blob)
f.close()

6
tests/test_transport.py
View File

@ -158,7 +158,7 @@ class TransportTest(ParamikoTest):
pass
def test_2_compute_key(self):
self.tc.K = 123281095979686581523377256114209720774539068973101330872763622971399429481072519713536292772709507296759612401802191955568143056534122385270077606457721553469730659233569339356140085284052436697480759510519672848743794433460113118986816826624865291116513647975790797391795651716378444844877749505443714557929L
self.tc.K = 123281095979686581523377256114209720774539068973101330872763622971399429481072519713536292772709507296759612401802191955568143056534122385270077606457721553469730659233569339356140085284052436697480759510519672848743794433460113118986816826624865291116513647975790797391795651716378444844877749505443714557929
self.tc.H = unhexlify('0C8307CDE6856FF30BA93684EB0F04C2520E9ED3')
self.tc.session_id = self.tc.H
key = self.tc._compute_key('C', 32)
@ -406,7 +406,7 @@ class TransportTest(ParamikoTest):
chan.close()
# allow a few seconds for the rekeying to complete
for i in xrange(50):
for i in range(50):
if self.tc.H != self.tc.session_id:
break
time.sleep(0.1)
@ -659,7 +659,7 @@ class TransportTest(ParamikoTest):
def run(self):
try:
for i in xrange(1, 1+self.iterations):
for i in range(1, 1+self.iterations):
if self.done_event.isSet():
break
self.watchdog_event.set()

20
tests/test_util.py
View File

@ -101,7 +101,7 @@ class UtilTest(ParamikoTest):
def test_2_parse_config(self):
global test_config_file
f = cStringIO.StringIO(test_config_file)
f = StringIO(test_config_file)
config = paramiko.util.parse_ssh_config(f)
self.assertEquals(config._config,
[{'host': ['*'], 'config': {}}, {'host': ['*'], 'config': {'identityfile': ['~/.ssh/id_rsa'], 'user': 'robey'}},
@ -111,7 +111,7 @@ class UtilTest(ParamikoTest):
def test_3_host_config(self):
global test_config_file
f = cStringIO.StringIO(test_config_file)
f = StringIO(test_config_file)
config = paramiko.util.parse_ssh_config(f)
for host, values in {
@ -138,7 +138,7 @@ class UtilTest(ParamikoTest):
def test_4_generate_key_bytes(self):
x = paramiko.util.generate_key_bytes(SHA, 'ABCDEFGH', 'This is my secret passphrase.', 64)
hex = ''.join(['%02x' % ord(c) for c in x])
hex = ''.join(['%02x' % byte_ord(c) for c in x])
self.assertEquals(hex, '9110e2f6793b69363e58173e9436b13a5a4b339005741d5c680e505f57d871347b4239f14fb5c46e857d5e100424873ba849ac699cea98d729e57b3e84378e8b')
def test_5_host_keys(self):
@ -172,7 +172,7 @@ Host *.example.com
Host *
Port 3333
"""
f = cStringIO.StringIO(test_config_file)
f = StringIO(test_config_file)
config = paramiko.util.parse_ssh_config(f)
host = 'www13.example.com'
self.assertEquals(
@ -216,7 +216,7 @@ Host space-delimited
Host equals-delimited
ProxyCommand=foo bar=biz baz
"""
f = cStringIO.StringIO(conf)
f = StringIO(conf)
config = paramiko.util.parse_ssh_config(f)
for host in ('space-delimited', 'equals-delimited'):
self.assertEquals(
@ -228,7 +228,7 @@ Host equals-delimited
"""
ProxyCommand should perform interpolation on the value
"""
config = paramiko.util.parse_ssh_config(cStringIO.StringIO("""
config = paramiko.util.parse_ssh_config(StringIO("""
Host specific
Port 37
ProxyCommand host %h port %p lol
@ -264,7 +264,7 @@ Host www13.*
Host *
Port 3333
"""
f = cStringIO.StringIO(test_config_file)
f = StringIO(test_config_file)
config = paramiko.util.parse_ssh_config(f)
host = 'www13.example.com'
self.assertEquals(
@ -293,7 +293,7 @@ ProxyCommand foo=bar:%h-%p
'foo=bar:proxy-without-equal-divisor-22'}
}.items():
f = cStringIO.StringIO(test_config_file)
f = StringIO(test_config_file)
config = paramiko.util.parse_ssh_config(f)
self.assertEquals(
paramiko.util.lookup_ssh_host_config(host, config),
@ -323,7 +323,7 @@ IdentityFile id_dsa22
'identityfile': ['id_dsa0', 'id_dsa1', 'id_dsa22']}
}.items():
f = cStringIO.StringIO(test_config_file)
f = StringIO(test_config_file)
config = paramiko.util.parse_ssh_config(f)
self.assertEquals(
paramiko.util.lookup_ssh_host_config(host, config),
@ -338,5 +338,5 @@ IdentityFile id_dsa22
AddressFamily inet
IdentityFile something_%l_using_fqdn
"""
config = paramiko.util.parse_ssh_config(cStringIO.StringIO(test_config))
config = paramiko.util.parse_ssh_config(StringIO(test_config))
assert config.lookup('meh') # will die during lookup() if bug regresses