diff options
Diffstat (limited to 'tests')
-rw-r--r-- | tests/stub_sftp.py | 1 | ||||
-rw-r--r-- | tests/test_auth.py | 6 | ||||
-rw-r--r-- | tests/test_client.py | 14 | ||||
-rw-r--r-- | tests/test_kex.py | 49 | ||||
-rw-r--r-- | tests/test_pkey.py | 10 | ||||
-rw-r--r-- | tests/test_sftp.py | 25 |
6 files changed, 61 insertions, 44 deletions
diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py index 13e7f3b4..100076d6 100644 --- a/tests/stub_sftp.py +++ b/tests/stub_sftp.py @@ -168,7 +168,6 @@ class StubSFTPServer(SFTPServerInterface): return SFTPServer.convert_errno(e.errno) return SFTP_OK - def mkdir(self, path, attr): path = self._realpath(path) try: diff --git a/tests/test_auth.py b/tests/test_auth.py index 14d465ca..6358a053 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -85,7 +85,7 @@ class NullServer(ServerInterface): return AUTH_SUCCESSFUL if username == "bad-server": raise Exception("Ack!") - if username == 'unresponsive-server': + if username == "unresponsive-server": sleep(5) return AUTH_SUCCESSFUL return AUTH_FAILED @@ -264,8 +264,8 @@ class AuthTest(unittest.TestCase): self.start_server() self.tc.connect() try: - remain = self.tc.auth_password('unresponsive-server', 'hello') + remain = self.tc.auth_password("unresponsive-server", "hello") except: etype, evalue, etb = sys.exc_info() self.assertTrue(issubclass(etype, AuthenticationException)) - self.assertTrue('Authentication timeout' in str(evalue)) + self.assertTrue("Authentication timeout" in str(evalue)) diff --git a/tests/test_client.py b/tests/test_client.py index 53ba53e4..87f7bcb2 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -41,10 +41,10 @@ from .util import _support, slow FINGERPRINTS = { - 'ssh-dss': b'\x44\x78\xf0\xb9\xa2\x3c\xc5\x18\x20\x09\xff\x75\x5b\xc1\xd2\x6c', - 'ssh-rsa': b'\x60\x73\x38\x44\xcb\x51\x86\x65\x7f\xde\xda\xa2\x2b\x5a\x57\xd5', - 'ecdsa-sha2-nistp256': b'\x25\x19\xeb\x55\xe6\xa1\x47\xff\x4f\x38\xd2\x75\x6f\xa5\xd5\x60', - 'ssh-ed25519': b'\xb3\xd5"\xaa\xf9u^\xe8\xcd\x0e\xea\x02\xb9)\xa2\x80', + "ssh-dss": b"\x44\x78\xf0\xb9\xa2\x3c\xc5\x18\x20\x09\xff\x75\x5b\xc1\xd2\x6c", + "ssh-rsa": b"\x60\x73\x38\x44\xcb\x51\x86\x65\x7f\xde\xda\xa2\x2b\x5a\x57\xd5", + "ecdsa-sha2-nistp256": b"\x25\x19\xeb\x55\xe6\xa1\x47\xff\x4f\x38\xd2\x75\x6f\xa5\xd5\x60", + "ssh-ed25519": b'\xb3\xd5"\xaa\xf9u^\xe8\xcd\x0e\xea\x02\xb9)\xa2\x80', } @@ -62,7 +62,7 @@ class NullServer(paramiko.ServerInterface): def check_auth_password(self, username, password): if (username == "slowdive") and (password == "pygmalion"): return paramiko.AUTH_SUCCESSFUL - if (username == 'slowdive') and (password == 'unresponsive-server'): + if (username == "slowdive") and (password == "unresponsive-server"): time.sleep(5) return paramiko.AUTH_SUCCESSFUL return paramiko.AUTH_FAILED @@ -208,7 +208,7 @@ class SSHClientTest(unittest.TestCase): self._test_connection(key_filename=_support("test_ecdsa_256.key")) def test_client_ed25519(self): - self._test_connection(key_filename=_support('test_ed25519.key')) + self._test_connection(key_filename=_support("test_ed25519.key")) def test_3_multiple_key_files(self): """ @@ -408,7 +408,7 @@ class SSHClientTest(unittest.TestCase): self.assertRaises( AuthenticationException, self._test_connection, - password='unresponsive-server', + password="unresponsive-server", auth_timeout=0.5, ) diff --git a/tests/test_kex.py b/tests/test_kex.py index b9b59db1..65eb9a17 100644 --- a/tests/test_kex.py +++ b/tests/test_kex.py @@ -40,15 +40,24 @@ from cryptography.hazmat.primitives.asymmetric import ec def dummy_urandom(n): return byte_chr(0xcc) * n + def dummy_generate_key_pair(obj): private_key_value = 94761803665136558137557783047955027733968423115106677159790289642479432803037 public_key_numbers = "042bdab212fa8ba1b7c843301682a4db424d307246c7e1e6083c41d9ca7b098bf30b3d63e2ec6278488c135360456cc054b3444ecc45998c08894cbc1370f5f989" - public_key_numbers_obj = ec.EllipticCurvePublicNumbers.from_encoded_point(ec.SECP256R1(), unhexlify(public_key_numbers)) - obj.P = ec.EllipticCurvePrivateNumbers(private_value=private_key_value, public_numbers=public_key_numbers_obj).private_key(default_backend()) + public_key_numbers_obj = ec.EllipticCurvePublicNumbers.from_encoded_point( + ec.SECP256R1(), unhexlify(public_key_numbers) + ) + obj.P = ec.EllipticCurvePrivateNumbers( + private_value=private_key_value, public_numbers=public_key_numbers_obj + ).private_key(default_backend()) if obj.transport.server_mode: - obj.Q_S = ec.EllipticCurvePublicNumbers.from_encoded_point(ec.SECP256R1(), unhexlify(public_key_numbers)).public_key(default_backend()) + obj.Q_S = ec.EllipticCurvePublicNumbers.from_encoded_point( + ec.SECP256R1(), unhexlify(public_key_numbers) + ).public_key(default_backend()) return - obj.Q_C = ec.EllipticCurvePublicNumbers.from_encoded_point(ec.SECP256R1(), unhexlify(public_key_numbers)).public_key(default_backend()) + obj.Q_C = ec.EllipticCurvePublicNumbers.from_encoded_point( + ec.SECP256R1(), unhexlify(public_key_numbers) + ).public_key(default_backend()) class FakeKey(object): @@ -446,20 +455,24 @@ class KexTest(unittest.TestCase): transport.server_mode = False kex = KexNistp256(transport) kex.start_kex() - self.assertEqual((paramiko.kex_ecdh_nist._MSG_KEXECDH_REPLY,), transport._expect) + self.assertEqual( + (paramiko.kex_ecdh_nist._MSG_KEXECDH_REPLY,), transport._expect + ) - #fake reply + # fake reply msg = Message() - msg.add_string('fake-host-key') - Q_S = unhexlify("043ae159594ba062efa121480e9ef136203fa9ec6b6e1f8723a321c16e62b945f573f3b822258cbcd094b9fa1c125cbfe5f043280893e66863cc0cb4dccbe70210") + msg.add_string("fake-host-key") + Q_S = unhexlify( + "043ae159594ba062efa121480e9ef136203fa9ec6b6e1f8723a321c16e62b945f573f3b822258cbcd094b9fa1c125cbfe5f043280893e66863cc0cb4dccbe70210" + ) msg.add_string(Q_S) - msg.add_string('fake-sig') + msg.add_string("fake-sig") msg.rewind() kex.parse_next(paramiko.kex_ecdh_nist._MSG_KEXECDH_REPLY, msg) - H = b'BAF7CE243A836037EB5D2221420F35C02B9AB6C957FE3BDE3369307B9612570A' + H = b"BAF7CE243A836037EB5D2221420F35C02B9AB6C957FE3BDE3369307B9612570A" self.assertEqual(K, kex.transport._K) self.assertEqual(H, hexlify(transport._H).upper()) - self.assertEqual((b'fake-host-key', b'fake-sig'), transport._verify) + self.assertEqual((b"fake-host-key", b"fake-sig"), transport._verify) self.assertTrue(transport._activated) def test_12_kex_nistp256_server(self): @@ -468,12 +481,16 @@ class KexTest(unittest.TestCase): transport.server_mode = True kex = KexNistp256(transport) kex.start_kex() - self.assertEqual((paramiko.kex_ecdh_nist._MSG_KEXECDH_INIT,), transport._expect) + self.assertEqual( + (paramiko.kex_ecdh_nist._MSG_KEXECDH_INIT,), transport._expect + ) - #fake init - msg=Message() - Q_C = unhexlify("043ae159594ba062efa121480e9ef136203fa9ec6b6e1f8723a321c16e62b945f573f3b822258cbcd094b9fa1c125cbfe5f043280893e66863cc0cb4dccbe70210") - H = b'2EF4957AFD530DD3F05DBEABF68D724FACC060974DA9704F2AEE4C3DE861E7CA' + # fake init + msg = Message() + Q_C = unhexlify( + "043ae159594ba062efa121480e9ef136203fa9ec6b6e1f8723a321c16e62b945f573f3b822258cbcd094b9fa1c125cbfe5f043280893e66863cc0cb4dccbe70210" + ) + H = b"2EF4957AFD530DD3F05DBEABF68D724FACC060974DA9704F2AEE4C3DE861E7CA" msg.add_string(Q_C) msg.rewind() kex.parse_next(paramiko.kex_ecdh_nist._MSG_KEXECDH_INIT, msg) diff --git a/tests/test_pkey.py b/tests/test_pkey.py index a70671b5..3a1279b6 100644 --- a/tests/test_pkey.py +++ b/tests/test_pkey.py @@ -462,15 +462,15 @@ class KeyTest(unittest.TestCase): self.assertEqual(str(key), comparable) def test_ed25519(self): - key1 = Ed25519Key.from_private_key_file(_support('test_ed25519.key')) + key1 = Ed25519Key.from_private_key_file(_support("test_ed25519.key")) key2 = Ed25519Key.from_private_key_file( - _support('test_ed25519_password.key'), b'abc123' + _support("test_ed25519_password.key"), b"abc123" ) self.assertNotEqual(key1.asbytes(), key2.asbytes()) def test_ed25519_compare(self): # verify that the private & public keys compare equal - key = Ed25519Key.from_private_key_file(_support('test_ed25519.key')) + key = Ed25519Key.from_private_key_file(_support("test_ed25519.key")) self.assertEqual(key, key) pub = Ed25519Key(data=key.asbytes()) self.assertTrue(key.can_sign()) @@ -480,10 +480,10 @@ class KeyTest(unittest.TestCase): def test_ed25519_nonbytes_password(self): # https://github.com/paramiko/paramiko/issues/1039 key = Ed25519Key.from_private_key_file( - _support('test_ed25519_password.key'), + _support("test_ed25519_password.key"), # NOTE: not a bytes. Amusingly, the test above for same key DOES # explicitly cast to bytes...code smell! - 'abc123', + "abc123", ) # No exception -> it's good. Meh. diff --git a/tests/test_sftp.py b/tests/test_sftp.py index a86fca5d..87c57340 100644 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -189,35 +189,36 @@ class TestSFTP(object): """Test posix-rename@openssh.com protocol extension.""" try: # first check that the normal rename works as specified - with sftp.open(sftp.FOLDER + '/a', 'w') as f: - f.write('one') - sftp.rename(sftp.FOLDER + '/a', sftp.FOLDER + '/b') - with sftp.open(sftp.FOLDER + '/a', 'w') as f: - f.write('two') + with sftp.open(sftp.FOLDER + "/a", "w") as f: + f.write("one") + sftp.rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b") + with sftp.open(sftp.FOLDER + "/a", "w") as f: + f.write("two") try: - sftp.rename(sftp.FOLDER + '/a', sftp.FOLDER + '/b') - self.assertTrue(False, 'no exception when rename-ing onto existing file') + sftp.rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b") + self.assertTrue( + False, "no exception when rename-ing onto existing file" + ) except (OSError, IOError): pass # now check with the posix_rename - sftp.posix_rename(sftp.FOLDER + '/a', sftp.FOLDER + '/b') - with sftp.open(sftp.FOLDER + '/b', 'r') as f: + sftp.posix_rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b") + with sftp.open(sftp.FOLDER + "/b", "r") as f: data = u(f.read()) err = "Contents of renamed file not the same as original file" assert data == "two", err finally: try: - sftp.remove(sftp.FOLDER + '/a') + sftp.remove(sftp.FOLDER + "/a") except: pass try: - sftp.remove(sftp.FOLDER + '/b') + sftp.remove(sftp.FOLDER + "/b") except: pass - def test_6_folder(self, sftp): """ create a temporary folder, verify that we can create a file in it, then |