diff options
| -rw-r--r-- | paramiko/client.py | 25 | ||||
| -rw-r--r-- | tests/test_client.py | 61 |
2 files changed, 73 insertions, 13 deletions
diff --git a/paramiko/client.py b/paramiko/client.py index c832c578..ae7186c3 100644 --- a/paramiko/client.py +++ b/paramiko/client.py @@ -347,22 +347,21 @@ class SSHClient (ClosingContextManager): server_hostkey_name = "[%s]:%d" % (hostname, port) our_server_keys = None - # If GSS-API Key Exchange is performed we are not required to check the - # host key, because the host is authenticated via GSS-API / SSPI as - # well as our client. - if not self._transport.use_gss_kex: - our_server_keys = self._system_host_keys.get(server_hostkey_name) - if our_server_keys is None: - our_server_keys = self._host_keys.get(server_hostkey_name) - if our_server_keys is not None: - keytype = our_server_keys.keys()[0] - sec_opts = t.get_security_options() - other_types = [x for x in sec_opts.key_types if x != keytype] - sec_opts.key_types = [keytype] + other_types + our_server_keys = self._system_host_keys.get(server_hostkey_name) + if our_server_keys is None: + our_server_keys = self._host_keys.get(server_hostkey_name) + if our_server_keys is not None: + keytype = our_server_keys.keys()[0] + sec_opts = t.get_security_options() + other_types = [x for x in sec_opts.key_types if x != keytype] + sec_opts.key_types = [keytype] + other_types t.start_client(timeout=timeout) - if not self._transport.use_gss_kex: + # If GSS-API Key Exchange is performed we are not required to check the + # host key, because the host is authenticated via GSS-API / SSPI as + # well as our client. + if not self._transport.gss_kex_used: server_key = t.get_remote_server_key() if our_server_keys is None: # will raise exception if the key is rejected diff --git a/tests/test_client.py b/tests/test_client.py index 2288cf57..4a7117ac 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -155,6 +155,7 @@ class SSHClientTest (unittest.TestCase): self.assertTrue(self.ts.is_active()) self.assertEqual('slowdive', self.ts.get_username()) self.assertEqual(True, self.ts.is_authenticated()) + self.assertEqual(False, self.tc.get_transport().gss_kex_used) # Command execution functions? stdin, stdout, stderr = self.tc.exec_command('yes') @@ -383,6 +384,66 @@ class SSHClientTest (unittest.TestCase): ) self._test_connection(**kwargs) + def test_9_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=[test_path('test_rsa.key')], + ) + self._test_connection(**kwargs) + + def test_10_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=[test_path('test_rsa.key')], + ) + self._test_connection(**kwargs) + + def test_11_reject_policy(self): + """ + verify that SSHClient's RejectPolicy works. + """ + threading.Thread(target=self._run).start() + + self.tc = paramiko.SSHClient() + self.tc.set_missing_host_key_policy(paramiko.RejectPolicy()) + self.assertEqual(0, len(self.tc.get_host_keys())) + self.assertRaises( + paramiko.SSHException, + self.tc.connect, + password='pygmalion', **self.connect_kwargs + ) + + def test_12_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() + self.tc.set_missing_host_key_policy(paramiko.RejectPolicy()) + self.assertEqual(0, len(self.tc.get_host_keys())) + self.assertRaises( + paramiko.SSHException, + self.tc.connect, + password='pygmalion', + gss_kex=True, + **self.connect_kwargs + ) + def _client_host_key_bad(self, host_key): threading.Thread(target=self._run).start() hostname = '[%s]:%d' % (self.addr, self.port) |