Merge pull request #158 from stacywsmith/support_ecdsa_auth_keys

Add support for ECDSA private keys.

ncclient/transport/ssh.py

@@ -426,7 +426,7 @@ 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):
+            for cls in (paramiko.RSAKey, paramiko.DSSKey, paramiko.ECDSAKey):
                 try:
                     key = cls.from_private_key_file(key_filename, password)
                     logger.debug("Trying key %s from %s" %
@@ -452,17 +452,23 @@ def _auth(self, username, password, key_filenames, allow_agent,
         if look_for_keys:
             rsa_key = os.path.expanduser("~/.ssh/id_rsa")
             dsa_key = os.path.expanduser("~/.ssh/id_dsa")
+            ecdsa_key = os.path.expanduser("~/.ssh/id_ecdsa")
             if os.path.isfile(rsa_key):
                 keyfiles.append((paramiko.RSAKey, rsa_key))
             if os.path.isfile(dsa_key):
                 keyfiles.append((paramiko.DSSKey, dsa_key))
+            if os.path.isfile(ecdsa_key):
+                keyfiles.append((paramiko.ECDSAKey, ecdsa_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")
             if os.path.isfile(rsa_key):
                 keyfiles.append((paramiko.RSAKey, rsa_key))
             if os.path.isfile(dsa_key):
                 keyfiles.append((paramiko.DSSKey, dsa_key))
+            if os.path.isfile(ecdsa_key):
+                keyfiles.append((paramiko.ECDSAKey, ecdsa_key))
 
         for cls, filename in keyfiles:
             try:

requirements.txt

@@ -1,4 +1,4 @@
 setuptools>0.6
-paramiko>=1.7.7.1
+paramiko>=1.15.0
 lxml>=3.3.0
 six

test/unit/transport/test_ssh.py

@@ -114,6 +114,18 @@ def test_auth_agent(self, mock_get_key, mock_auth_public_key):
             (mock_auth_public_key.call_args_list[0][0][1]).__repr__(),
             key.__repr__())
 
+    @patch('paramiko.transport.Transport.auth_publickey')
+    @patch('paramiko.agent.AgentSSH.get_keys')
+    def test_auth_agent_exception(self, mock_get_key, mock_auth_public_key):
+        key = paramiko.PKey()
+        mock_get_key.return_value = [key]
+        mock_auth_public_key.side_effect = paramiko.ssh_exception.AuthenticationException
+        device_handler = JunosDeviceHandler({'name': 'junos'})
+        obj = SSHSession(device_handler)
+        obj._transport = paramiko.Transport(None)
+        self.assertRaises(AuthenticationError,
+            obj._auth,'user', None, [], True, False)
+
     @patch('paramiko.transport.Transport.auth_publickey')
     @patch('paramiko.pkey.PKey.from_private_key_file')
     def test_auth_keyfiles(self, mock_get_key, mock_auth_public_key):
@@ -127,6 +139,47 @@ def test_auth_keyfiles(self, mock_get_key, mock_auth_public_key):
             (mock_auth_public_key.call_args_list[0][0][1]).__repr__(),
             key.__repr__())
 
+    @patch('paramiko.transport.Transport.auth_publickey')
+    @patch('paramiko.pkey.PKey.from_private_key_file')
+    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
+        device_handler = JunosDeviceHandler({'name': 'junos'})
+        obj = SSHSession(device_handler)
+        obj._transport = paramiko.Transport(None)
+        self.assertRaises(AuthenticationError,
+            obj._auth,'user', None, ["key_file_name"], False, True)
+
+    @patch('os.path.isfile')
+    @patch('paramiko.transport.Transport.auth_publickey')
+    @patch('paramiko.pkey.PKey.from_private_key_file')
+    def test_auth_default_keyfiles(self, mock_get_key, mock_auth_public_key,
+                                   mock_is_file):
+        key = paramiko.PKey()
+        mock_get_key.return_value = key
+        mock_is_file.return_value = True
+        device_handler = JunosDeviceHandler({'name': 'junos'})
+        obj = SSHSession(device_handler)
+        obj._transport = paramiko.Transport(None)
+        obj._auth('user', 'password', [], False, True)
+        self.assertEqual(
+            (mock_auth_public_key.call_args_list[0][0][1]).__repr__(),
+            key.__repr__())
+
+    @patch('os.path.isfile')
+    @patch('paramiko.transport.Transport.auth_publickey')
+    @patch('paramiko.pkey.PKey.from_private_key_file')
+    def test_auth_default_keyfiles_exception(self, mock_get_key,
+                                             mock_auth_public_key, mock_is_file):
+        key = paramiko.PKey()
+        mock_is_file.return_value = True
+        mock_get_key.side_effect = paramiko.ssh_exception.PasswordRequiredException
+        device_handler = JunosDeviceHandler({'name': 'junos'})
+        obj = SSHSession(device_handler)
+        obj._transport = paramiko.Transport(None)
+        self.assertRaises(AuthenticationError,
+			              obj._auth,'user', None, [], False, True)
+
     @patch('paramiko.transport.Transport.auth_password')
     def test_auth_password(self, mock_auth_password):
         device_handler = JunosDeviceHandler({'name': 'junos'})
@@ -147,6 +200,13 @@ def test_auth_exception(self, mock_auth_password):
         self.assertRaises(AuthenticationError,
             obj._auth, 'user', 'password', [], False, True)
 
+    def test_auth_no_methods_exception(self):
+        device_handler = JunosDeviceHandler({'name': 'junos'})
+        obj = SSHSession(device_handler)
+        obj._transport = paramiko.Transport(None)
+        self.assertRaises(AuthenticationError,
+            obj._auth,'user', None, [], False, False)
+
     @patch('paramiko.transport.Transport.close')
     def test_close(self, mock_close):
         device_handler = JunosDeviceHandler({'name': 'junos'})