diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_auth.py | 1 | ||||
| -rw-r--r-- | tests/test_client.py | 150 | ||||
| -rw-r--r-- | tests/test_kex.py | 2 | ||||
| -rw-r--r-- | tests/test_sftp.py | 9 | ||||
| -rw-r--r-- | tests/test_util.py | 11 | ||||
| -rw-r--r-- | tests/util.py | 2 |
6 files changed, 123 insertions, 52 deletions
diff --git a/tests/test_auth.py b/tests/test_auth.py index 6358a053..fe1a32a1 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -255,6 +255,7 @@ class AuthTest(unittest.TestCase): etype, evalue, etb = sys.exc_info() self.assertTrue(issubclass(etype, AuthenticationException)) + @slow def test_9_auth_non_responsive(self): """ verify that authentication times out if server takes to long to diff --git a/tests/test_client.py b/tests/test_client.py index fec1485e..80b28adf 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -20,7 +20,7 @@ Some unit tests for SSHClient. """ -from __future__ import with_statement +from __future__ import with_statement, print_function import gc import os @@ -33,6 +33,8 @@ import warnings import weakref from tempfile import mkstemp +from pytest_relaxed import raises + import paramiko from paramiko.pkey import PublicBlob from paramiko.common import PY2 @@ -41,6 +43,10 @@ from paramiko.ssh_exception import SSHException, AuthenticationException from .util import _support, slow +requires_gss_auth = unittest.skipUnless( + paramiko.GSS_AUTH_AVAILABLE, "GSS auth not available" +) + 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", @@ -107,7 +113,7 @@ class NullServer(paramiko.ServerInterface): return True -class SSHClientTest(unittest.TestCase): +class ClientTest(unittest.TestCase): def setUp(self): self.sockl = socket.socket() self.sockl.bind(("localhost", 0)) @@ -120,16 +126,42 @@ class SSHClientTest(unittest.TestCase): look_for_keys=False, ) self.event = threading.Event() + self.kill_event = threading.Event() def tearDown(self): - for attr in "tc ts socks sockl".split(): - if hasattr(self, attr): - getattr(self, attr).close() - - def _run(self, allowed_keys=None, delay=0, public_blob=None): + # Shut down client Transport + if hasattr(self, "tc"): + self.tc.close() + # Shut down shared socket + if hasattr(self, "sockl"): + # Signal to server thread that it should shut down early; it checks + # this immediately after accept(). (In scenarios where connection + # actually succeeded during the test, this becomes a no-op.) + self.kill_event.set() + # Forcibly connect to server sock in case the server thread is + # hanging out in its accept() (e.g. if the client side of the test + # fails before it even gets to connecting); there's no other good + # way to force an accept() to exit. + put_a_sock_in_it = socket.socket() + put_a_sock_in_it.connect((self.addr, self.port)) + put_a_sock_in_it.close() + # Then close "our" end of the socket (which _should_ cause the + # accept() to bail out, but does not, for some reason. I blame + # threading.) + self.sockl.close() + + def _run( + self, allowed_keys=None, delay=0, public_blob=None, kill_event=None + ): if allowed_keys is None: allowed_keys = FINGERPRINTS.keys() self.socks, addr = self.sockl.accept() + # If the kill event was set at this point, it indicates an early + # shutdown, so bail out now and don't even try setting up a Transport + # (which will just verbosely die.) + if kill_event and kill_event.is_set(): + self.socks.close() + return self.ts = paramiko.Transport(self.socks) keypath = _support("test_rsa.key") host_key = paramiko.RSAKey.from_private_key_file(keypath) @@ -149,7 +181,7 @@ class SSHClientTest(unittest.TestCase): The exception is ``allowed_keys`` which is stripped and handed to the ``NullServer`` used for testing. """ - run_kwargs = {} + run_kwargs = {"kill_event": self.kill_event} for key in ("allowed_keys", "public_blob"): run_kwargs[key] = kwargs.pop(key, None) # Server setup @@ -194,6 +226,8 @@ class SSHClientTest(unittest.TestCase): stdout.close() stderr.close() + +class SSHClientTest(ClientTest): def test_1_client(self): """ verify that the SSHClient stuff works too. @@ -242,7 +276,7 @@ class SSHClientTest(unittest.TestCase): try: self._test_connection( key_filename=[ - _support("test_{0}.key".format(x)) for x in attempt + _support("test_{}.key".format(x)) for x in attempt ], allowed_keys=[types_[x] for x in accept], ) @@ -271,7 +305,7 @@ class SSHClientTest(unittest.TestCase): # server-side behavior is 100% identical.) # NOTE: only bothered whipping up one cert per overall class/family. for type_ in ("rsa", "dss", "ecdsa_256", "ed25519"): - cert_name = "test_{0}.key-cert.pub".format(type_) + cert_name = "test_{}.key-cert.pub".format(type_) cert_path = _support(os.path.join("cert_support", cert_name)) self._test_connection( key_filename=cert_path, @@ -286,12 +320,12 @@ class SSHClientTest(unittest.TestCase): # that a specific cert was found, along with regular authorization # succeeding proving that the overall flow works. for type_ in ("rsa", "dss", "ecdsa_256", "ed25519"): - key_name = "test_{0}.key".format(type_) + key_name = "test_{}.key".format(type_) key_path = _support(os.path.join("cert_support", key_name)) self._test_connection( key_filename=key_path, public_blob=PublicBlob.from_file( - "{0}-cert.pub".format(key_path) + "{}-cert.pub".format(key_path) ), ) @@ -447,6 +481,7 @@ class SSHClientTest(unittest.TestCase): ) self._test_connection(**kwargs) + @slow def test_9_auth_timeout(self): """ verify that the SSHClient has a configurable auth timeout @@ -459,21 +494,19 @@ class SSHClientTest(unittest.TestCase): auth_timeout=0.5, ) + @requires_gss_auth def test_10_auth_trickledown_gsskex(self): """ Failed gssapi-keyex auth doesn't prevent subsequent key auth from succeeding """ - if not paramiko.GSS_AUTH_AVAILABLE: - return # for python 2.6 lacks skipTest kwargs = dict(gss_kex=True, key_filename=[_support("test_rsa.key")]) self._test_connection(**kwargs) + @requires_gss_auth def test_11_auth_trickledown_gssauth(self): """ Failed gssapi-with-mic auth doesn't prevent subsequent key auth from succeeding """ - if not paramiko.GSS_AUTH_AVAILABLE: - return # for python 2.6 lacks skipTest kwargs = dict(gss_auth=True, key_filename=[_support("test_rsa.key")]) self._test_connection(**kwargs) @@ -493,14 +526,13 @@ class SSHClientTest(unittest.TestCase): **self.connect_kwargs ) + @requires_gss_auth def test_13_reject_policy_gsskex(self): """ verify that SSHClient's RejectPolicy works, even if gssapi-keyex was enabled but not used. """ # Test for a bug present in paramiko versions released before 2017-08-01 - if not paramiko.GSS_AUTH_AVAILABLE: - return # for python 2.6 lacks skipTest threading.Thread(target=self._run).start() self.tc = paramiko.SSHClient() @@ -560,10 +592,7 @@ class SSHClientTest(unittest.TestCase): def test_host_key_negotiation_4(self): self._client_host_key_good(paramiko.RSAKey, "test_rsa.key") - def test_update_environment(self): - """ - Verify that environment variables can be set by the client. - """ + def _setup_for_env(self): threading.Thread(target=self._run).start() self.tc = paramiko.SSHClient() @@ -577,6 +606,11 @@ class SSHClientTest(unittest.TestCase): self.assertTrue(self.event.isSet()) self.assertTrue(self.ts.is_active()) + def test_update_environment(self): + """ + Verify that environment variables can be set by the client. + """ + self._setup_for_env() target_env = {b"A": b"B", b"C": b"d"} self.tc.exec_command("yes", environment=target_env) @@ -584,22 +618,20 @@ class SSHClientTest(unittest.TestCase): self.assertEqual(target_env, getattr(schan, "env", {})) schan.close() - # Cannot use assertRaises in context manager mode as it is not supported - # in Python 2.6. - try: + @unittest.skip("Clients normally fail silently, thus so do we, for now") + def test_env_update_failures(self): + self._setup_for_env() + with self.assertRaises(SSHException) as manager: # Verify that a rejection by the server can be detected self.tc.exec_command("yes", environment={b"INVALID_ENV": b""}) - except SSHException as e: - self.assertTrue( - "INVALID_ENV" in str(e), - "Expected variable name in error message", - ) - self.assertTrue( - isinstance(e.args[1], SSHException), - "Expected original SSHException in exception", - ) - else: - self.assertFalse(False, "SSHException was not thrown.") + self.assertTrue( + "INVALID_ENV" in str(manager.exception), + "Expected variable name in error message", + ) + self.assertTrue( + isinstance(manager.exception.args[1], SSHException), + "Expected original SSHException in exception", + ) def test_missing_key_policy_accepts_classes_or_instances(self): """ @@ -616,3 +648,49 @@ class SSHClientTest(unittest.TestCase): # Hand in just the class (new behavior) client.set_missing_host_key_policy(paramiko.AutoAddPolicy) assert isinstance(client._policy, paramiko.AutoAddPolicy) + + +class PasswordPassphraseTests(ClientTest): + # TODO: most of these could reasonably be set up to use mocks/assertions + # (e.g. "gave passphrase -> expect PKey was given it as the passphrase") + # instead of suffering a real connection cycle. + # TODO: in that case, move the below to be part of an integration suite? + + def test_password_kwarg_works_for_password_auth(self): + # Straightforward / duplicate of earlier basic password test. + self._test_connection(password="pygmalion") + + # TODO: more granular exception pending #387; should be signaling "no auth + # methods available" because no key and no password + @raises(SSHException) + def test_passphrase_kwarg_not_used_for_password_auth(self): + # Using the "right" password in the "wrong" field shouldn't work. + self._test_connection(passphrase="pygmalion") + + def test_passphrase_kwarg_used_for_key_passphrase(self): + # Straightforward again, with new passphrase kwarg. + self._test_connection( + key_filename=_support("test_rsa_password.key"), + passphrase="television", + ) + + def test_password_kwarg_used_for_passphrase_when_no_passphrase_kwarg_given( + self + ): # noqa + # Backwards compatibility: passphrase in the password field. + self._test_connection( + key_filename=_support("test_rsa_password.key"), + password="television", + ) + + @raises(AuthenticationException) # TODO: more granular + def test_password_kwarg_not_used_for_passphrase_when_passphrase_kwarg_given( + self + ): # noqa + # Sanity: if we're given both fields, the password field is NOT used as + # a passphrase. + self._test_connection( + key_filename=_support("test_rsa_password.key"), + password="television", + passphrase="wat? lol no", + ) diff --git a/tests/test_kex.py b/tests/test_kex.py index 65eb9a17..62512beb 100644 --- a/tests/test_kex.py +++ b/tests/test_kex.py @@ -33,8 +33,6 @@ from paramiko.kex_gex import KexGex, KexGexSHA256 from paramiko import Message from paramiko.common import byte_chr from paramiko.kex_ecdh_nist import KexNistp256 -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives.asymmetric import ec def dummy_urandom(n): diff --git a/tests/test_sftp.py b/tests/test_sftp.py index 87c57340..fdb7c9bc 100644 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -194,20 +194,15 @@ class TestSFTP(object): sftp.rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b") with sftp.open(sftp.FOLDER + "/a", "w") as f: f.write("two") - try: + with pytest.raises(IOError): # actual message seems generic 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: data = u(f.read()) err = "Contents of renamed file not the same as original file" - assert data == "two", err + assert "two" == data, err finally: try: diff --git a/tests/test_util.py b/tests/test_util.py index 12256b39..705baa14 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -419,8 +419,7 @@ IdentityFile something_%l_using_fqdn f = StringIO(test_config_file) config = paramiko.util.parse_ssh_config(f) self.assertEqual( - config.get_hostnames(), - set(["*", "*.example.com", "spoo.example.com"]), + config.get_hostnames(), {"*", "*.example.com", "spoo.example.com"} ) def test_quoted_host_names(self): @@ -512,12 +511,12 @@ Host param3 parara self.assertRaises(Exception, conf._get_hosts, host) def test_safe_string(self): - vanilla = b("vanilla") - has_bytes = b("has \7\3 bytes") + vanilla = b"vanilla" + has_bytes = b"has \7\3 bytes" safe_vanilla = safe_string(vanilla) safe_has_bytes = safe_string(has_bytes) - expected_bytes = b("has %07%03 bytes") - err = "{0!r} != {1!r}" + expected_bytes = b"has %07%03 bytes" + err = "{!r} != {!r}" msg = err.format(safe_vanilla, vanilla) assert safe_vanilla == vanilla, msg msg = err.format(safe_has_bytes, expected_bytes) diff --git a/tests/util.py b/tests/util.py index c1ba4b2c..4ca02374 100644 --- a/tests/util.py +++ b/tests/util.py @@ -20,7 +20,7 @@ def needs_builtin(name): """ Skip decorated test if builtin name does not exist. """ - reason = "Test requires a builtin '{0}'".format(name) + reason = "Test requires a builtin '{}'".format(name) return pytest.mark.skipif(not hasattr(builtins, name), reason=reason) |