diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_agent.py | 50 | ||||
| -rw-r--r-- | tests/test_kex.py | 3 | ||||
| -rw-r--r-- | tests/test_pkey.py | 26 | ||||
| -rw-r--r-- | tests/test_transport.py | 286 |
4 files changed, 354 insertions, 11 deletions
diff --git a/tests/test_agent.py b/tests/test_agent.py new file mode 100644 index 00000000..c3973bbb --- /dev/null +++ b/tests/test_agent.py @@ -0,0 +1,50 @@ +import unittest + +from paramiko.message import Message +from paramiko.agent import ( + SSH2_AGENT_SIGN_RESPONSE, + cSSH2_AGENTC_SIGN_REQUEST, + SSH_AGENT_RSA_SHA2_256, + SSH_AGENT_RSA_SHA2_512, + AgentKey, +) +from paramiko.py3compat import b + + +class ChaosAgent: + def _send_message(self, msg): + self._sent_message = msg + sig = Message() + sig.add_string(b("lol")) + sig.rewind() + return SSH2_AGENT_SIGN_RESPONSE, sig + + +class AgentTests(unittest.TestCase): + def _sign_with_agent(self, kwargs, expectation): + agent = ChaosAgent() + key = AgentKey(agent, b("secret!!!")) + result = key.sign_ssh_data(b("token"), **kwargs) + assert result == b("lol") + msg = agent._sent_message + msg.rewind() + assert msg.get_byte() == cSSH2_AGENTC_SIGN_REQUEST + assert msg.get_string() == b("secret!!!") + assert msg.get_string() == b("token") + assert msg.get_int() == expectation + + def test_agent_signing_defaults_to_0_for_flags_field(self): + # No algorithm kwarg at all + self._sign_with_agent(kwargs=dict(), expectation=0) + + def test_agent_signing_is_2_for_SHA256(self): + self._sign_with_agent( + kwargs=dict(algorithm="rsa-sha2-256"), + expectation=SSH_AGENT_RSA_SHA2_256, + ) + + def test_agent_signing_is_2_for_SHA512(self): + self._sign_with_agent( + kwargs=dict(algorithm="rsa-sha2-512"), + expectation=SSH_AGENT_RSA_SHA2_512, + ) diff --git a/tests/test_kex.py b/tests/test_kex.py index c251611a..b73989c2 100644 --- a/tests/test_kex.py +++ b/tests/test_kex.py @@ -76,7 +76,7 @@ class FakeKey(object): def asbytes(self): return b"fake-key" - def sign_ssh_data(self, H): + def sign_ssh_data(self, H, algorithm): return b"fake-sig" @@ -93,6 +93,7 @@ class FakeTransport(object): remote_version = "SSH-2.0-lame" local_kex_init = "local-kex-init" remote_kex_init = "remote-kex-init" + host_key_type = "fake-key" def _send_message(self, m): self._message = m diff --git a/tests/test_pkey.py b/tests/test_pkey.py index 94b2492b..0cc20133 100644 --- a/tests/test_pkey.py +++ b/tests/test_pkey.py @@ -63,6 +63,8 @@ FINGER_ECDSA_256 = "256 25:19:eb:55:e6:a1:47:ff:4f:38:d2:75:6f:a5:d5:60" FINGER_ECDSA_384 = "384 c1:8d:a0:59:09:47:41:8e:a8:a6:07:01:29:23:b4:65" FINGER_ECDSA_521 = "521 44:58:22:52:12:33:16:0e:ce:0e:be:2c:7c:7e:cc:1e" SIGNED_RSA = "20:d7:8a:31:21:cb:f7:92:12:f2:a4:89:37:f5:78:af:e6:16:b6:25:b9:97:3d:a2:cd:5f:ca:20:21:73:4c:ad:34:73:8f:20:77:28:e2:94:15:08:d8:91:40:7a:85:83:bf:18:37:95:dc:54:1a:9b:88:29:6c:73:ca:38:b4:04:f1:56:b9:f2:42:9d:52:1b:29:29:b4:4f:fd:c9:2d:af:47:d2:40:76:30:f3:63:45:0c:d9:1d:43:86:0f:1c:70:e2:93:12:34:f3:ac:c5:0a:2f:14:50:66:59:f1:88:ee:c1:4a:e9:d1:9c:4e:46:f0:0e:47:6f:38:74:f1:44:a8" # noqa +SIGNED_RSA_256 = "cc:6:60:e0:0:2c:ac:9e:26:bc:d5:68:64:3f:9f:a7:e5:aa:41:eb:88:4a:25:5:9c:93:84:66:ef:ef:60:f4:34:fb:f4:c8:3d:55:33:6a:77:bd:b2:ee:83:f:71:27:41:7e:f5:7:5:0:a9:4c:7:80:6f:be:76:67:cb:58:35:b9:2b:f3:c2:d3:3c:ee:e1:3f:59:e0:fa:e4:5c:92:ed:ae:74:de:d:d6:27:16:8f:84:a3:86:68:c:94:90:7d:6e:cc:81:12:d8:b6:ad:aa:31:a8:13:3d:63:81:3e:bb:5:b6:38:4d:2:d:1b:5b:70:de:83:cc:3a:cb:31" # noqa +SIGNED_RSA_512 = "87:46:8b:75:92:33:78:a0:22:35:32:39:23:c6:ab:e1:6:92:ad:bc:7f:6e:ab:19:32:e4:78:b2:2c:8f:1d:c:65:da:fc:a5:7:ca:b6:55:55:31:83:b1:a0:af:d1:95:c5:2e:af:56:ba:f5:41:64:f:39:9d:af:82:43:22:8f:90:52:9d:89:e7:45:97:df:f3:f2:bc:7b:3a:db:89:e:34:fd:18:62:25:1b:ef:77:aa:c6:6c:99:36:3a:84:d6:9c:2a:34:8c:7f:f4:bb:c9:a5:9a:6c:11:f2:cf:da:51:5e:1e:7f:90:27:34:de:b2:f3:15:4f:db:47:32:6b:a7" # noqa FINGER_RSA_2K_OPENSSH = "2048 68:d1:72:01:bf:c0:0c:66:97:78:df:ce:75:74:46:d6" FINGER_DSS_1K_OPENSSH = "1024 cf:1d:eb:d7:61:d3:12:94:c6:c0:c6:54:35:35:b0:82" FINGER_EC_384_OPENSSH = "384 72:14:df:c1:9a:c3:e6:0e:11:29:d6:32:18:7b:ea:9b" @@ -238,21 +240,29 @@ class KeyTest(unittest.TestCase): self.assertTrue(not pub.can_sign()) self.assertEqual(key, pub) - def test_sign_rsa(self): - # verify that the rsa private key can sign and verify + def _sign_and_verify_rsa(self, algorithm, saved_sig): key = RSAKey.from_private_key_file(_support("test_rsa.key")) - msg = key.sign_ssh_data(b"ice weasels") - self.assertTrue(type(msg) is Message) + msg = key.sign_ssh_data(b"ice weasels", algorithm) + assert isinstance(msg, Message) msg.rewind() - self.assertEqual("ssh-rsa", msg.get_text()) - sig = bytes().join( - [byte_chr(int(x, 16)) for x in SIGNED_RSA.split(":")] + assert msg.get_text() == algorithm + expected = bytes().join( + [byte_chr(int(x, 16)) for x in saved_sig.split(":")] ) - self.assertEqual(sig, msg.get_binary()) + assert msg.get_binary() == expected msg.rewind() pub = RSAKey(data=key.asbytes()) self.assertTrue(pub.verify_ssh_sig(b"ice weasels", msg)) + def test_sign_and_verify_ssh_rsa(self): + self._sign_and_verify_rsa("ssh-rsa", SIGNED_RSA) + + def test_sign_and_verify_rsa_sha2_512(self): + self._sign_and_verify_rsa("rsa-sha2-512", SIGNED_RSA_512) + + def test_sign_and_verify_rsa_sha2_256(self): + self._sign_and_verify_rsa("rsa-sha2-256", SIGNED_RSA_256) + def test_sign_dss(self): # verify that the dss private key can sign and verify key = DSSKey.from_private_key_file(_support("test_dss.key")) diff --git a/tests/test_transport.py b/tests/test_transport.py index e2174896..77ffd6c1 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -23,6 +23,7 @@ Some unit tests for the ssh2 protocol in Transport. from __future__ import with_statement from binascii import hexlify +from contextlib import contextmanager import select import socket import time @@ -38,6 +39,8 @@ from paramiko import ( Packetizer, RSAKey, SSHException, + AuthenticationException, + IncompatiblePeer, SecurityOptions, ServerInterface, Transport, @@ -80,6 +83,9 @@ class NullServer(ServerInterface): paranoid_did_public_key = False paranoid_key = DSSKey.from_private_key_file(_support("test_dss.key")) + def __init__(self, allowed_keys=None): + self.allowed_keys = allowed_keys if allowed_keys is not None else [] + def get_allowed_auths(self, username): if username == "slowdive": return "publickey,password" @@ -90,6 +96,11 @@ class NullServer(ServerInterface): return AUTH_SUCCESSFUL return AUTH_FAILED + def check_auth_publickey(self, username, key): + if key in self.allowed_keys: + return AUTH_SUCCESSFUL + return AUTH_FAILED + def check_channel_request(self, kind, chanid): if kind == "bogus": return OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED @@ -154,6 +165,7 @@ class TransportTest(unittest.TestCase): self.socks.close() self.sockc.close() + # TODO: unify with newer contextmanager def setup_test_server( self, client_options=None, server_options=None, connect_kwargs=None ): @@ -245,7 +257,7 @@ class TransportTest(unittest.TestCase): self.assertEqual(True, self.tc.is_authenticated()) self.assertEqual(True, self.ts.is_authenticated()) - def testa_long_banner(self): + def test_long_banner(self): """ verify that a long banner doesn't mess up the handshake. """ @@ -339,7 +351,7 @@ class TransportTest(unittest.TestCase): self.assertEqual("This is on stderr.\n", f.readline()) self.assertEqual("", f.readline()) - def testa_channel_can_be_used_as_context_manager(self): + def test_channel_can_be_used_as_context_manager(self): """ verify that exec_command() does something reasonable. """ @@ -1169,3 +1181,273 @@ class AlgorithmDisablingTests(unittest.TestCase): assert "ssh-dss" not in server_keys assert "diffie-hellman-group14-sha256" not in kexen assert "zlib" not in compressions + + +@contextmanager +def server( + hostkey=None, + init=None, + server_init=None, + client_init=None, + connect=None, + pubkeys=None, + catch_error=False, +): + """ + SSH server contextmanager for testing. + + :param hostkey: + Host key to use for the server; if None, loads + ``test_rsa.key``. + :param init: + Default `Transport` constructor kwargs to use for both sides. + :param server_init: + Extends and/or overrides ``init`` for server transport only. + :param client_init: + Extends and/or overrides ``init`` for client transport only. + :param connect: + Kwargs to use for ``connect()`` on the client. + :param pubkeys: + List of public keys for auth. + :param catch_error: + Whether to capture connection errors & yield from contextmanager. + Necessary for connection_time exception testing. + """ + if init is None: + init = {} + if server_init is None: + server_init = {} + if client_init is None: + client_init = {} + if connect is None: + connect = dict(username="slowdive", password="pygmalion") + socks = LoopSocket() + sockc = LoopSocket() + sockc.link(socks) + tc = Transport(sockc, **dict(init, **client_init)) + ts = Transport(socks, **dict(init, **server_init)) + + if hostkey is None: + hostkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + ts.add_server_key(hostkey) + event = threading.Event() + server = NullServer(allowed_keys=pubkeys) + assert not event.is_set() + assert not ts.is_active() + assert tc.get_username() is None + assert ts.get_username() is None + assert not tc.is_authenticated() + assert not ts.is_authenticated() + + err = None + # Trap errors and yield instead of raising right away; otherwise callers + # cannot usefully deal with problems at connect time which stem from errors + # in the server side. + try: + ts.start_server(event, server) + tc.connect(**connect) + + event.wait(1.0) + assert event.is_set() + assert ts.is_active() + assert tc.is_active() + + except Exception as e: + if not catch_error: + raise + err = e + + yield (tc, ts, err) if catch_error else (tc, ts) + + tc.close() + ts.close() + socks.close() + sockc.close() + + +_disable_sha2 = dict( + disabled_algorithms=dict(keys=["rsa-sha2-256", "rsa-sha2-512"]) +) +_disable_sha1 = dict(disabled_algorithms=dict(keys=["ssh-rsa"])) +_disable_sha2_pubkey = dict( + disabled_algorithms=dict(pubkeys=["rsa-sha2-256", "rsa-sha2-512"]) +) +_disable_sha1_pubkey = dict(disabled_algorithms=dict(pubkeys=["ssh-rsa"])) + + +class TestSHA2SignatureKeyExchange(unittest.TestCase): + # NOTE: these all rely on the default server() hostkey being RSA + # NOTE: these rely on both sides being properly implemented re: agreed-upon + # hostkey during kex being what's actually used. Truly proving that eg + # SHA512 was used, is quite difficult w/o super gross hacks. However, there + # are new tests in test_pkey.py which use known signature blobs to prove + # the SHA2 family was in fact used! + + def test_base_case_ssh_rsa_still_used_as_fallback(self): + # Prove that ssh-rsa is used if either, or both, participants have SHA2 + # algorithms disabled + for which in ("init", "client_init", "server_init"): + with server(**{which: _disable_sha2}) as (tc, _): + assert tc.host_key_type == "ssh-rsa" + + def test_kex_with_sha2_512(self): + # It's the default! + with server() as (tc, _): + assert tc.host_key_type == "rsa-sha2-512" + + def test_kex_with_sha2_256(self): + # No 512 -> you get 256 + with server( + init=dict(disabled_algorithms=dict(keys=["rsa-sha2-512"])) + ) as (tc, _): + assert tc.host_key_type == "rsa-sha2-256" + + def _incompatible_peers(self, client_init, server_init): + with server( + client_init=client_init, server_init=server_init, catch_error=True + ) as (tc, ts, err): + # If neither side blew up then that's bad! + assert err is not None + # If client side blew up first, it'll be straightforward + if isinstance(err, IncompatiblePeer): + pass + # If server side blew up first, client sees EOF & we need to check + # the server transport for its saved error (otherwise it can only + # appear in log output) + elif isinstance(err, EOFError): + assert ts.saved_exception is not None + assert isinstance(ts.saved_exception, IncompatiblePeer) + # If it was something else, welp + else: + raise err + + def test_client_sha2_disabled_server_sha1_disabled_no_match(self): + self._incompatible_peers( + client_init=_disable_sha2, server_init=_disable_sha1 + ) + + def test_client_sha1_disabled_server_sha2_disabled_no_match(self): + self._incompatible_peers( + client_init=_disable_sha1, server_init=_disable_sha2 + ) + + def test_explicit_client_hostkey_not_limited(self): + # Be very explicit about the hostkey on BOTH ends, + # and ensure it still ends up choosing sha2-512. + # (This is a regression test vs previous implementation which overwrote + # the entire preferred-hostkeys structure when given an explicit key as + # a client.) + hostkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + with server(hostkey=hostkey, connect=dict(hostkey=hostkey)) as (tc, _): + assert tc.host_key_type == "rsa-sha2-512" + + +class TestExtInfo(unittest.TestCase): + def test_ext_info_handshake(self): + with server() as (tc, _): + kex = tc._get_latest_kex_init() + assert kex["kex_algo_list"][-1] == "ext-info-c" + assert tc.server_extensions == { + "server-sig-algs": b"ssh-ed25519,ecdsa-sha2-nistp256,ecdsa-sha2-nistp384,ecdsa-sha2-nistp521,rsa-sha2-512,rsa-sha2-256,ssh-rsa,ssh-dss" # noqa + } + + def test_client_uses_server_sig_algs_for_pubkey_auth(self): + privkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + with server( + pubkeys=[privkey], + connect=dict(pkey=privkey), + server_init=dict( + disabled_algorithms=dict(pubkeys=["rsa-sha2-512"]) + ), + ) as (tc, _): + assert tc.is_authenticated() + # Client settled on 256 despite itself not having 512 disabled + assert tc._agreed_pubkey_algorithm == "rsa-sha2-256" + + +# TODO: these could move into test_auth.py but that badly needs refactoring +# with this module anyways... +class TestSHA2SignaturePubkeys(unittest.TestCase): + def test_pubkey_auth_honors_disabled_algorithms(self): + privkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + with server( + pubkeys=[privkey], + connect=dict(pkey=privkey), + init=dict( + disabled_algorithms=dict( + pubkeys=["ssh-rsa", "rsa-sha2-256", "rsa-sha2-512"] + ) + ), + catch_error=True, + ) as (_, _, err): + assert isinstance(err, SSHException) + assert "no RSA pubkey algorithms" in str(err) + + def test_client_sha2_disabled_server_sha1_disabled_no_match(self): + privkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + with server( + pubkeys=[privkey], + connect=dict(pkey=privkey), + client_init=_disable_sha2_pubkey, + server_init=_disable_sha1_pubkey, + catch_error=True, + ) as (tc, ts, err): + assert isinstance(err, AuthenticationException) + + def test_client_sha1_disabled_server_sha2_disabled_no_match(self): + privkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + with server( + pubkeys=[privkey], + connect=dict(pkey=privkey), + client_init=_disable_sha1_pubkey, + server_init=_disable_sha2_pubkey, + catch_error=True, + ) as (tc, ts, err): + assert isinstance(err, AuthenticationException) + + def test_ssh_rsa_still_used_when_sha2_disabled(self): + privkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + # NOTE: this works because key obj comparison uses public bytes + # TODO: would be nice for PKey to grow a legit "give me another obj of + # same class but just the public bits" using asbytes() + with server( + pubkeys=[privkey], connect=dict(pkey=privkey), init=_disable_sha2 + ) as (tc, _): + assert tc.is_authenticated() + + def test_sha2_512(self): + privkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + with server( + pubkeys=[privkey], + connect=dict(pkey=privkey), + init=dict( + disabled_algorithms=dict(pubkeys=["ssh-rsa", "rsa-sha2-256"]) + ), + ) as (tc, ts): + assert tc.is_authenticated() + assert tc._agreed_pubkey_algorithm == "rsa-sha2-512" + + def test_sha2_256(self): + privkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + with server( + pubkeys=[privkey], + connect=dict(pkey=privkey), + init=dict( + disabled_algorithms=dict(pubkeys=["ssh-rsa", "rsa-sha2-512"]) + ), + ) as (tc, ts): + assert tc.is_authenticated() + assert tc._agreed_pubkey_algorithm == "rsa-sha2-256" + + def test_sha2_256_when_client_only_enables_256(self): + privkey = RSAKey.from_private_key_file(_support("test_rsa.key")) + with server( + pubkeys=[privkey], + connect=dict(pkey=privkey), + # Client-side only; server still accepts all 3. + client_init=dict( + disabled_algorithms=dict(pubkeys=["ssh-rsa", "rsa-sha2-512"]) + ), + ) as (tc, ts): + assert tc.is_authenticated() + assert tc._agreed_pubkey_algorithm == "rsa-sha2-256" |