Repository: ncclient/ncclient

Issue:
ncclient doesn't support ssh certificate authentication
ncclient directly tries to [load SSH private keys depending on their type](https://github.com/ncclient/ncclient/blob/652ef67b266df0ee02bbbbc4cc6eb09ca01f4878/ncclient/transport/ssh.py#L408-L419), circumventing Paramiko's [SSH certificate detection](https://github.com/paramiko/paramiko/blob/23f92003898b060df0e2b8b1d889455264e63a3e/paramiko/pkey.py#L151-L162) (and [loading](https://github.com/paramiko/paramiko/blob/23f92003898b060df0e2b8b1d889455264e63a3e/paramiko/pkey.py#L196-L198)).

The following PR offers to address this issue: #600

Discussion:

Solution patch:
diff --git a/ncclient/transport/ssh.py b/ncclient/transport/ssh.py
index 483f9d3f..d1796807 100644
--- a/ncclient/transport/ssh.py
+++ b/ncclient/transport/ssh.py
@@ -406,17 +406,16 @@ def _auth(self, username, password, key_filenames, allow_agent,
         saved_exception = None
 
         for key_filename in key_filenames:
-            for cls in (paramiko.RSAKey, paramiko.DSSKey, paramiko.ECDSAKey, paramiko.Ed25519Key):
-                try:
-                    key = cls.from_private_key_file(key_filename, password)
-                    self.logger.debug("Trying key %s from %s",
-                                      hexlify(key.get_fingerprint()),
-                                      key_filename)
-                    self._transport.auth_publickey(username, key)
-                    return
-                except Exception as e:
-                    saved_exception = e
-                    self.logger.debug(e)
+            try:
+                key = paramiko.PKey.from_path(key_filename, password.encode("utf-8"))
+                self.logger.debug("Trying key %s from %s",
+                                  hexlify(key.get_fingerprint()),
+                                  key_filename)
+                self._transport.auth_publickey(username, key)
+                return
+            except Exception as e:
+                saved_exception = e
+                self.logger.debug(e)
 
         if allow_agent:
             # resequence keys from agent using private key names
@@ -455,30 +454,30 @@ def _auth(self, username, password, key_filenames, allow_agent,
             ecdsa_key = os.path.expanduser("~/.ssh/id_ecdsa")
             ed25519_key = os.path.expanduser("~/.ssh/id_ed25519")
             if os.path.isfile(rsa_key):
-                keyfiles.append((paramiko.RSAKey, rsa_key))
+                keyfiles.append(rsa_key)
             if os.path.isfile(dsa_key):
-                keyfiles.append((paramiko.DSSKey, dsa_key))
+                keyfiles.append(dsa_key)
             if os.path.isfile(ecdsa_key):
-                keyfiles.append((paramiko.ECDSAKey, ecdsa_key))
+                keyfiles.append(ecdsa_key)
             if os.path.isfile(ed25519_key):
-                keyfiles.append((paramiko.Ed25519Key, ed25519_key))
+                keyfiles.append(ed25519_key)
             # look in ~/ssh/ for windows users:
             rsa_key = os.path.expanduser("~/ssh/id_rsa")
             dsa_key = os.path.expanduser("~/ssh/id_dsa")
             ecdsa_key = os.path.expanduser("~/ssh/id_ecdsa")
             ed25519_key = os.path.expanduser("~/ssh/id_ed25519")
             if os.path.isfile(rsa_key):
-                keyfiles.append((paramiko.RSAKey, rsa_key))
+                keyfiles.append(rsa_key)
             if os.path.isfile(dsa_key):
-                keyfiles.append((paramiko.DSSKey, dsa_key))
+                keyfiles.append(dsa_key)
             if os.path.isfile(ecdsa_key):
-                keyfiles.append((paramiko.ECDSAKey, ecdsa_key))
+                keyfiles.append(ecdsa_key)
             if os.path.isfile(ed25519_key):
-                keyfiles.append((paramiko.Ed25519Key, ed25519_key))
+                keyfiles.append(ed25519_key)
 
-        for cls, filename in keyfiles:
+        for filename in keyfiles:
             try:
-                key = cls.from_private_key_file(filename, password)
+                key = paramiko.PKey.from_path(filename, password.encode("utf-8"))
                 self.logger.debug("Trying discovered key %s in %s",
                                   hexlify(key.get_fingerprint()), filename)
                 self._transport.auth_publickey(username, key)


Test patch:
diff --git a/test/unit/transport/test_ssh.py b/test/unit/transport/test_ssh.py
index b4b44edf..73a82be5 100644
--- a/test/unit/transport/test_ssh.py
+++ b/test/unit/transport/test_ssh.py
@@ -164,7 +164,7 @@ def test_auth_agent_exception(self, mock_get_key, mock_auth_public_key):
             obj._auth,'user', None, [], True, False)
 
     @patch('paramiko.transport.Transport.auth_publickey')
-    @patch('paramiko.pkey.PKey.from_private_key_file')
+    @patch('paramiko.PKey.from_path')
     def test_auth_keyfiles(self, mock_get_key, mock_auth_public_key):
         key = paramiko.PKey()
         mock_get_key.return_value = key
@@ -177,7 +177,7 @@ def test_auth_keyfiles(self, mock_get_key, mock_auth_public_key):
             key.__repr__())
 
     @patch('paramiko.transport.Transport.auth_publickey')
-    @patch('paramiko.pkey.PKey.from_private_key_file')
+    @patch('paramiko.PKey.from_path')
     def test_auth_keyfiles_exception(self, mock_get_key, mock_auth_public_key):
         key = paramiko.PKey()
         mock_get_key.side_effect = paramiko.ssh_exception.PasswordRequiredException
@@ -189,7 +189,7 @@ def test_auth_keyfiles_exception(self, mock_get_key, mock_auth_public_key):
 
     @patch('os.path.isfile')
     @patch('paramiko.transport.Transport.auth_publickey')
-    @patch('paramiko.pkey.PKey.from_private_key_file')
+    @patch('paramiko.PKey.from_path')
     def test_auth_default_keyfiles(self, mock_get_key, mock_auth_public_key,
                                    mock_is_file):
         key = paramiko.PKey()
@@ -205,7 +205,7 @@ def test_auth_default_keyfiles(self, mock_get_key, mock_auth_public_key,
 
     @patch('os.path.isfile')
     @patch('paramiko.transport.Transport.auth_publickey')
-    @patch('paramiko.pkey.PKey.from_private_key_file')
+    @patch('paramiko.PKey.from_path')
     def test_auth_default_keyfiles_exception(self, mock_get_key,
                                              mock_auth_public_key, mock_is_file):
         key = paramiko.PKey()