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

Use Crypto.Random rather than Crypto.Util.RandomPool.

This commit is contained in:
Gary van der Merwe 2010-08-03 00:13:08 +02:00
parent e2add90981
commit 044e702998
21 changed files with 50 additions and 385 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
NEWS
View File

@ -8,6 +8,11 @@ Highlights of what's new in each release:
Releases
========
Unreleased
----------
* Use Crypto.Random rather than Crypto.Util.RandomPool.
(Gary van der Merwe, #271791)
v1.7.6 (Fanny) 1nov09
---------------------
* fixed bugs 411099 (sftp chdir isn't unicode-safe), 363163 & 411910 (more

2
paramiko/__init__.py
View File

@ -66,7 +66,7 @@ __version_info__ = (1, 7, 6)
__license__ = "GNU Lesser General Public License (LGPL)"
from transport import randpool, SecurityOptions, Transport
from transport import SecurityOptions, Transport
from client import SSHClient, MissingHostKeyPolicy, AutoAddPolicy, RejectPolicy, WarningPolicy
from auth_handler import AuthHandler
from channel import Channel, ChannelFile

2
paramiko/agent.py
View File

@ -139,7 +139,7 @@ class AgentKey(PKey):
def get_name(self):
return self.name
def sign_ssh_data(self, randpool, data):
def sign_ssh_data(self, rng, data):
msg = Message()
msg.add_byte(chr(SSH2_AGENTC_SIGN_REQUEST))
msg.add_string(self.blob)

2
paramiko/auth_handler.py
View File

@ -206,7 +206,7 @@ class AuthHandler (object):
m.add_string(self.private_key.get_name())
m.add_string(str(self.private_key))
blob = self._get_session_blob(self.private_key, 'ssh-connection', self.username)
sig = self.private_key.sign_ssh_data(self.transport.randpool, blob)
sig = self.private_key.sign_ssh_data(self.transport.rng, blob)
m.add_string(str(sig))
elif self.auth_method == 'keyboard-interactive':
m.add_string('')

2
paramiko/channel.py
View File

@ -364,7 +364,7 @@ class Channel (object):
if auth_protocol is None:
auth_protocol = 'MIT-MAGIC-COOKIE-1'
if auth_cookie is None:
auth_cookie = binascii.hexlify(self.transport.randpool.get_bytes(16))
auth_cookie = binascii.hexlify(self.transport.rng.read(16))
m = Message()
m.add_byte(chr(MSG_CHANNEL_REQUEST))

4
paramiko/common.py
View File

@ -95,10 +95,10 @@ CONNECTION_FAILED_CODE = {
DISCONNECT_SERVICE_NOT_AVAILABLE, DISCONNECT_AUTH_CANCELLED_BY_USER, \
DISCONNECT_NO_MORE_AUTH_METHODS_AVAILABLE = 7, 13, 14
from rng import StrongLockingRandomPool
from Crypto import Random
# keep a crypto-strong PRNG nearby
randpool = StrongLockingRandomPool()
rng = Random.new()
import sys
if sys.version_info < (2, 3):

7
paramiko/dsskey.py
View File

@ -91,13 +91,13 @@ class DSSKey (PKey):
def can_sign(self):
return self.x is not None
def sign_ssh_data(self, rpool, data):
def sign_ssh_data(self, rng, data):
digest = SHA.new(data).digest()
dss = DSA.construct((long(self.y), long(self.g), long(self.p), long(self.q), long(self.x)))
# generate a suitable k
qsize = len(util.deflate_long(self.q, 0))
while True:
k = util.inflate_long(rpool.get_bytes(qsize), 1)
k = util.inflate_long(rng.read(qsize), 1)
if (k > 2) and (k < self.q):
break
r, s = dss.sign(util.inflate_long(digest, 1), k)
@ -161,8 +161,7 @@ class DSSKey (PKey):
@return: new private key
@rtype: L{DSSKey}
"""
randpool.stir()
dsa = DSA.generate(bits, randpool.get_bytes, progress_func)
dsa = DSA.generate(bits, rng.read, progress_func)
key = DSSKey(vals=(dsa.p, dsa.q, dsa.g, dsa.y))
key.x = dsa.x
return key

2
paramiko/hostkeys.py
View File

@ -303,7 +303,7 @@ class HostKeys (UserDict.DictMixin):
@rtype: str
"""
if salt is None:
salt = randpool.get_bytes(SHA.digest_size)
salt = rng.read(SHA.digest_size)
else:
if salt.startswith('|1|'):
salt = salt.split('|')[2]

5
paramiko/kex_gex.py
View File

@ -101,8 +101,7 @@ class KexGex (object):
qhbyte <<= 1
qmask >>= 1
while True:
self.transport.randpool.stir()
x_bytes = self.transport.randpool.get_bytes(bytes)
x_bytes = self.transport.rng.read(bytes)
x_bytes = chr(ord(x_bytes[0]) & qmask) + x_bytes[1:]
x = util.inflate_long(x_bytes, 1)
if (x > 1) and (x < q):
@ -207,7 +206,7 @@ class KexGex (object):
H = SHA.new(str(hm)).digest()
self.transport._set_K_H(K, H)
# sign it
sig = self.transport.get_server_key().sign_ssh_data(self.transport.randpool, H)
sig = self.transport.get_server_key().sign_ssh_data(self.transport.rng, H)
# send reply
m = Message()
m.add_byte(chr(_MSG_KEXDH_GEX_REPLY))

5
paramiko/kex_group1.py
View File

@ -79,8 +79,7 @@ class KexGroup1(object):
# potential x where the first 63 bits are 1, because some of those will be
# larger than q (but this is a tiny tiny subset of potential x).
while 1:
self.transport.randpool.stir()
x_bytes = self.transport.randpool.get_bytes(128)
x_bytes = self.transport.rng.read(128)
x_bytes = chr(ord(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'):
@ -125,7 +124,7 @@ class KexGroup1(object):
H = SHA.new(str(hm)).digest()
self.transport._set_K_H(K, H)
# sign it
sig = self.transport.get_server_key().sign_ssh_data(self.transport.randpool, H)
sig = self.transport.get_server_key().sign_ssh_data(self.transport.rng, H)
# send reply
m = Message()
m.add_byte(chr(_MSG_KEXDH_REPLY))

7
paramiko/packet.py
View File

@ -311,9 +311,6 @@ class Packetizer (object):
self.__sent_bytes += len(out)
self.__sent_packets += 1
if (self.__sent_packets % 100) == 0:
# stirring the randpool takes 30ms on my ibook!!
randpool.stir()
if ((self.__sent_packets >= self.REKEY_PACKETS) or (self.__sent_bytes >= self.REKEY_BYTES)) \
and not self.__need_rekey:
# only ask once for rekeying
@ -359,7 +356,7 @@ class Packetizer (object):
raise SSHException('Mismatched MAC')
padding = ord(packet[0])
payload = packet[1:packet_size - padding]
randpool.add_event()
if self.__dump_packets:
self._log(DEBUG, 'Got payload (%d bytes, %d padding)' % (packet_size, padding))
@ -476,7 +473,7 @@ class Packetizer (object):
packet = struct.pack('>IB', len(payload) + padding + 1, padding)
packet += payload
if self.__block_engine_out is not None:
packet += randpool.get_bytes(padding)
packet += rng.read(padding)
else:
# cute trick i caught openssh doing: if we're not encrypting,
# don't waste random bytes for the padding

10
paramiko/pkey.py
View File

@ -143,13 +143,13 @@ class PKey (object):
"""
return base64.encodestring(str(self)).replace('\n', '')
def sign_ssh_data(self, randpool, data):
def sign_ssh_data(self, rng, data):
"""
Sign a blob of data with this private key, and return a L{Message}
representing an SSH signature message.
@param randpool: a secure random number generator.
@type randpool: L{Crypto.Util.randpool.RandomPool}
@param rng: a secure random number generator.
@type rng: L{Crypto.Util.rng.RandomPool}
@param data: the data to sign.
@type data: str
@return: an SSH signature message.
@ -360,11 +360,11 @@ class PKey (object):
keysize = self._CIPHER_TABLE[cipher_name]['keysize']
blocksize = self._CIPHER_TABLE[cipher_name]['blocksize']
mode = self._CIPHER_TABLE[cipher_name]['mode']
salt = randpool.get_bytes(8)
salt = rng.read(8)
key = util.generate_key_bytes(MD5, salt, password, keysize)
if len(data) % blocksize != 0:
n = blocksize - len(data) % blocksize
#data += randpool.get_bytes(n)
#data += rng.read(n)
# that would make more sense ^, but it confuses openssh.
data += '\0' * n
data = cipher.new(key, mode, salt).encrypt(data)

12
paramiko/primes.py
View File

@ -26,12 +26,12 @@ from paramiko import util
from paramiko.ssh_exception import SSHException
def _generate_prime(bits, randpool):
def _generate_prime(bits, rng):
"primtive attempt at prime generation"
hbyte_mask = pow(2, bits % 8) - 1
while True:
# loop catches the case where we increment n into a higher bit-range
x = randpool.get_bytes((bits+7) // 8)
x = rng.read((bits+7) // 8)
if hbyte_mask > 0:
x = chr(ord(x[0]) & hbyte_mask) + x[1:]
n = util.inflate_long(x, 1)
@ -43,7 +43,7 @@ def _generate_prime(bits, randpool):
break
return n
def _roll_random(rpool, n):
def _roll_random(rng, n):
"returns a random # from 0 to N-1"
bits = util.bit_length(n-1)
bytes = (bits + 7) // 8
@ -56,7 +56,7 @@ def _roll_random(rpool, n):
# fits, so i can't guarantee that this loop will ever finish, but the odds
# of it looping forever should be infinitesimal.
while True:
x = rpool.get_bytes(bytes)
x = rng.read(bytes)
if hbyte_mask > 0:
x = chr(ord(x[0]) & hbyte_mask) + x[1:]
num = util.inflate_long(x, 1)
@ -75,7 +75,7 @@ class ModulusPack (object):
# pack is a hash of: bits -> [ (generator, modulus) ... ]
self.pack = {}
self.discarded = []
self.randpool = rpool
self.rng = rpool
def _parse_modulus(self, line):
timestamp, mod_type, tests, tries, size, generator, modulus = line.split()
@ -147,5 +147,5 @@ class ModulusPack (object):
if min > good:
good = bitsizes[-1]
# now pick a random modulus of this bitsize
n = _roll_random(self.randpool, len(self.pack[good]))
n = _roll_random(self.rng, len(self.pack[good]))
return self.pack[good][n]

112
paramiko/rng.py
View File

@ -1,112 +0,0 @@
#!/usr/bin/python
# -*- coding: ascii -*-
# Copyright (C) 2008 Dwayne C. Litzenberger <dlitz@dlitz.net>
#
# This file is part of paramiko.
#
# Paramiko is free software; you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2.1 of the License, or (at your option)
# any later version.
#
# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
import sys
import threading
from Crypto.Util.randpool import RandomPool as _RandomPool
try:
import platform
except ImportError:
platform = None # Not available using Python 2.2
def _strxor(a, b):
assert len(a) == len(b)
return "".join(map(lambda x, y: chr(ord(x) ^ ord(y)), a, b))
##
## Find a strong random entropy source, depending on the detected platform.
## WARNING TO DEVELOPERS: This will fail on some systems, but do NOT use
## Crypto.Util.randpool.RandomPool as a fall-back. RandomPool will happily run
## with very little entropy, thus _silently_ defeating any security that
## Paramiko attempts to provide. (This is current as of PyCrypto 2.0.1).
## See http://www.lag.net/pipermail/paramiko/2008-January/000599.html
## and http://www.lag.net/pipermail/paramiko/2008-April/000678.html
##
if ((platform is not None and platform.system().lower() == 'windows') or
sys.platform == 'win32'):
# MS Windows
from paramiko import rng_win32
rng_device = rng_win32.open_rng_device()
else:
# Assume POSIX (any system where /dev/urandom exists)
from paramiko import rng_posix
rng_device = rng_posix.open_rng_device()
class StrongLockingRandomPool(object):
"""Wrapper around RandomPool guaranteeing strong random numbers.
Crypto.Util.randpool.RandomPool will silently operate even if it is seeded
with little or no entropy, and it provides no prediction resistance if its
state is ever compromised throughout its runtime. It is also not thread-safe.
This wrapper augments RandomPool by XORing its output with random bits from
the operating system, and by controlling access to the underlying
RandomPool using an exclusive lock.
"""
def __init__(self, instance=None):
if instance is None:
instance = _RandomPool()
self.randpool = instance
self.randpool_lock = threading.Lock()
self.entropy = rng_device
# Stir 256 bits of entropy from the RNG device into the RandomPool.
self.randpool.stir(self.entropy.read(32))
self.entropy.randomize()
def stir(self, s=''):
self.randpool_lock.acquire()
try:
self.randpool.stir(s)
finally:
self.randpool_lock.release()
self.entropy.randomize()
def randomize(self, N=0):
self.randpool_lock.acquire()
try:
self.randpool.randomize(N)
finally:
self.randpool_lock.release()
self.entropy.randomize()
def add_event(self, s=''):
self.randpool_lock.acquire()
try:
self.randpool.add_event(s)
finally:
self.randpool_lock.release()
def get_bytes(self, N):
self.randpool_lock.acquire()
try:
randpool_data = self.randpool.get_bytes(N)
finally:
self.randpool_lock.release()
entropy_data = self.entropy.read(N)
result = _strxor(randpool_data, entropy_data)
assert len(randpool_data) == N and len(entropy_data) == N and len(result) == N
return result
# vim:set ts=4 sw=4 sts=4 expandtab:

97
paramiko/rng_posix.py
View File

@ -1,97 +0,0 @@
#!/usr/bin/python
# -*- coding: ascii -*-
# Copyright (C) 2008 Dwayne C. Litzenberger <dlitz@dlitz.net>
# Copyright (C) 2008 Open Systems Canada Limited
#
# This file is part of paramiko.
#
# Paramiko is free software; you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2.1 of the License, or (at your option)
# any later version.
#
# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
import os
import stat
class error(Exception):
pass
class _RNG(object):
def __init__(self, file):
self.file = file
def read(self, bytes):
return self.file.read(bytes)
def close(self):
return self.file.close()
def randomize(self):
return
def open_rng_device(device_path=None):
"""Open /dev/urandom and perform some sanity checks."""
f = None
g = None
if device_path is None:
device_path = "/dev/urandom"
try:
# Try to open /dev/urandom now so that paramiko will be able to access
# it even if os.chroot() is invoked later.
try:
f = open(device_path, "rb", 0)
except EnvironmentError:
raise error("Unable to open /dev/urandom")
# Open a second file descriptor for sanity checking later.
try:
g = open(device_path, "rb", 0)
except EnvironmentError:
raise error("Unable to open /dev/urandom")
# Check that /dev/urandom is a character special device, not a regular file.
st = os.fstat(f.fileno()) # f
if stat.S_ISREG(st.st_mode) or not stat.S_ISCHR(st.st_mode):
raise error("/dev/urandom is not a character special device")
st = os.fstat(g.fileno()) # g
if stat.S_ISREG(st.st_mode) or not stat.S_ISCHR(st.st_mode):
raise error("/dev/urandom is not a character special device")
# Check that /dev/urandom always returns the number of bytes requested
x = f.read(20)
y = g.read(20)
if len(x) != 20 or len(y) != 20:
raise error("Error reading from /dev/urandom: input truncated")
# Check that different reads return different data
if x == y:
raise error("/dev/urandom is broken; returning identical data: %r == %r" % (x, y))
# Close the duplicate file object
g.close()
# Return the first file object
return _RNG(f)
except error:
if f is not None:
f.close()
if g is not None:
g.close()
raise
# vim:set ts=4 sw=4 sts=4 expandtab:

121
paramiko/rng_win32.py
View File

@ -1,121 +0,0 @@
#!/usr/bin/python
# -*- coding: ascii -*-
# Copyright (C) 2008 Dwayne C. Litzenberger <dlitz@dlitz.net>
# Copyright (C) 2008 Open Systems Canada Limited
#
# This file is part of paramiko.
#
# Paramiko is free software; you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2.1 of the License, or (at your option)
# any later version.
#
# Paramiko is distrubuted in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
class error(Exception):
pass
# Try to import the "winrandom" module
try:
from Crypto.Util import winrandom as _winrandom
except ImportError:
_winrandom = None
# Try to import the "urandom" module
try:
from os import urandom as _urandom
except ImportError:
_urandom = None
class _RNG(object):
def __init__(self, readfunc):
self.read = readfunc
def randomize(self):
# According to "Cryptanalysis of the Random Number Generator of the
# Windows Operating System", by Leo Dorrendorf and Zvi Gutterman
# and Benny Pinkas <http://eprint.iacr.org/2007/419>,
# CryptGenRandom only updates its internal state using kernel-provided
# random data every 128KiB of output.
self.read(128*1024) # discard 128 KiB of output
def _open_winrandom():
if _winrandom is None:
raise error("Crypto.Util.winrandom module not found")
# Check that we can open the winrandom module
try:
r0 = _winrandom.new()
r1 = _winrandom.new()
except Exception, exc:
raise error("winrandom.new() failed: %s" % str(exc), exc)
# Check that we can read from the winrandom module
try:
x = r0.get_bytes(20)
y = r1.get_bytes(20)
except Exception, exc:
raise error("winrandom get_bytes failed: %s" % str(exc), exc)
# Check that the requested number of bytes are returned
if len(x) != 20 or len(y) != 20:
raise error("Error reading from winrandom: input truncated")
# Check that different reads return different data
if x == y:
raise error("winrandom broken: returning identical data")
return _RNG(r0.get_bytes)
def _open_urandom():
if _urandom is None:
raise error("os.urandom function not found")
# Check that we can read from os.urandom()
try:
x = _urandom(20)
y = _urandom(20)
except Exception, exc:
raise error("os.urandom failed: %s" % str(exc), exc)
# Check that the requested number of bytes are returned
if len(x) != 20 or len(y) != 20:
raise error("os.urandom failed: input truncated")
# Check that different reads return different data
if x == y:
raise error("os.urandom failed: returning identical data")
return _RNG(_urandom)
def open_rng_device():
# Try using the Crypto.Util.winrandom module
try:
return _open_winrandom()
except error:
pass
# Several versions of PyCrypto do not contain the winrandom module, but
# Python >= 2.4 has os.urandom, so try to use that.
try:
return _open_urandom()
except error:
pass
# SECURITY NOTE: DO NOT USE Crypto.Util.randpool.RandomPool HERE!
# If we got to this point, RandomPool will silently run with very little
# entropy. (This is current as of PyCrypto 2.0.1).
# See http://www.lag.net/pipermail/paramiko/2008-January/000599.html
# and http://www.lag.net/pipermail/paramiko/2008-April/000678.html
raise error("Unable to find a strong random entropy source. You cannot run this software securely under the current configuration.")
# vim:set ts=4 sw=4 sts=4 expandtab:

3
paramiko/rsakey.py
View File

@ -137,8 +137,7 @@ class RSAKey (PKey):
@return: new private key
@rtype: L{RSAKey}
"""
randpool.stir()
rsa = RSA.generate(bits, randpool.get_bytes, progress_func)
rsa = RSA.generate(bits, rng.read, progress_func)
key = RSAKey(vals=(rsa.e, rsa.n))
key.d = rsa.d
key.p = rsa.p

12
paramiko/transport.py
View File

@ -297,7 +297,7 @@ class Transport (threading.Thread):
# okay, normal socket-ish flow here...
threading.Thread.__init__(self)
self.setDaemon(True)
self.randpool = randpool
self.rng = rng
self.sock = sock
# Python < 2.3 doesn't have the settimeout method - RogerB
try:
@ -585,7 +585,7 @@ class Transport (threading.Thread):
@note: This has no effect when used in client mode.
"""
Transport._modulus_pack = ModulusPack(randpool)
Transport._modulus_pack = ModulusPack(rng)
# places to look for the openssh "moduli" file
file_list = [ '/etc/ssh/moduli', '/usr/local/etc/moduli' ]
if filename is not None:
@ -837,10 +837,9 @@ class Transport (threading.Thread):
"""
m = Message()
m.add_byte(chr(MSG_IGNORE))
randpool.stir()
if bytes is None:
bytes = (ord(randpool.get_bytes(1)) % 32) + 10
m.add_bytes(randpool.get_bytes(bytes))
bytes = (ord(rng.read(1)) % 32) + 10
m.add_bytes(rng.read(bytes))
self._send_user_message(m)
def renegotiate_keys(self):
@ -1674,10 +1673,9 @@ class Transport (threading.Thread):
else:
available_server_keys = self._preferred_keys
randpool.stir()
m = Message()
m.add_byte(chr(MSG_KEXINIT))
m.add_bytes(randpool.get_bytes(16))
m.add_bytes(rng.read(16))
m.add_list(self._preferred_kex)
m.add_list(available_server_keys)
m.add_list(self._preferred_ciphers)

10
tests/test_kex.py
View File

@ -28,17 +28,15 @@ from paramiko.kex_gex import KexGex
from paramiko import Message
class FakeRandpool (object):
def stir(self):
pass
def get_bytes(self, n):
class FakeRng (object):
def read(self, n):
return chr(0xcc) * n
class FakeKey (object):
def __str__(self):
return 'fake-key'
def sign_ssh_data(self, randpool, H):
def sign_ssh_data(self, rng, H):
return 'fake-sig'
@ -50,7 +48,7 @@ class FakeModulusPack (object):
class FakeTransport (object):
randpool = FakeRandpool()
rng = FakeRng()
local_version = 'SSH-2.0-paramiko_1.0'
remote_version = 'SSH-2.0-lame'
local_kex_init = 'local-kex-init'

11
tests/test_pkey.py
View File

@ -23,7 +23,8 @@ Some unit tests for public/private key objects.
from binascii import hexlify, unhexlify
import StringIO
import unittest
from paramiko import RSAKey, DSSKey, Message, util, randpool
from paramiko import RSAKey, DSSKey, Message, util
from paramiko.common import rng
# from openssh's ssh-keygen
PUB_RSA = 'ssh-rsa AAAAB3NzaC1yc2EAAAABIwAAAIEA049W6geFpmsljTwfvI1UmKWWJPNFI74+vNKTk4dmzkQY2yAMs6FhlvhlI8ysU4oj71ZsRYMecHbBbxdN79+JRFVYTKaLqjwGENeTd+yv4q+V2PvZv3fLnzApI3l7EJCqhWwJUHJ1jAkZzqDx0tyOL4uoZpww3nmE0kb3y21tH4c='
@ -151,7 +152,7 @@ class KeyTest (unittest.TestCase):
def test_8_sign_rsa(self):
# verify that the rsa private key can sign and verify
key = RSAKey.from_private_key_file('tests/test_rsa.key')
msg = key.sign_ssh_data(randpool, 'ice weasels')
msg = key.sign_ssh_data(rng, 'ice weasels')
self.assert_(type(msg) is Message)
msg.rewind()
self.assertEquals('ssh-rsa', msg.get_string())
@ -164,7 +165,7 @@ class KeyTest (unittest.TestCase):
def test_9_sign_dss(self):
# verify that the dss private key can sign and verify
key = DSSKey.from_private_key_file('tests/test_dss.key')
msg = key.sign_ssh_data(randpool, 'ice weasels')
msg = key.sign_ssh_data(rng, 'ice weasels')
self.assert_(type(msg) is Message)
msg.rewind()
self.assertEquals('ssh-dss', msg.get_string())
@ -178,12 +179,12 @@ class KeyTest (unittest.TestCase):
def test_A_generate_rsa(self):
key = RSAKey.generate(1024)
msg = key.sign_ssh_data(randpool, 'jerri blank')
msg = key.sign_ssh_data(rng, 'jerri blank')
msg.rewind()
self.assert_(key.verify_ssh_sig('jerri blank', msg))
def test_B_generate_dss(self):
key = DSSKey.generate(1024)
msg = key.sign_ssh_data(randpool, 'jerri blank')
msg = key.sign_ssh_data(rng, 'jerri blank')
msg.rewind()
self.assert_(key.verify_ssh_sig('jerri blank', msg))

4
tests/test_util.py
View File

@ -147,8 +147,8 @@ class UtilTest (unittest.TestCase):
os.unlink('hostfile.temp')
def test_6_random(self):
from paramiko.common import randpool
from paramiko.common import rng
# just verify that we can pull out 32 bytes and not get an exception.
x = randpool.get_bytes(32)
x = rng.read(32)
self.assertEquals(len(x), 32)