Added fake SSH agent to fake_server module. Added SSH agent forwarding support to fake server. Added unittest for client authentication via SSH agent.

Author: pkittenis

fake_server/fake_agent.py

@@ -0,0 +1,28 @@
+"""
+Fake SSH Agent for testing ssh agent forwarding and agent based key
+authentication
+"""
+
+import paramiko.agent
+
+class FakeAgent(paramiko.agent.AgentSSH):
+
+    def __init__(self):
+        self._conn = None
+        self.keys = []
+    
+    def add_key(self, key):
+        """Add key to agent.
+        :param key: Key to add
+        :type key: :mod:`paramiko.pkey.PKey`
+        """
+        self.keys.append(key)
+
+    def _connect(self, conn):
+        pass
+
+    def _close(self):
+        self._keys = []
+
+    def get_keys(self):
+        return tuple(self.keys)

fake_server/fake_server.py

@@ -53,6 +53,10 @@ def check_channel_pty_request(self, channel, term, width, height, pixelwidth,
                                   pixelheight, modes):
         return True
 
+    def check_channel_forward_agent_request(self, channel):
+        logger.debug("Forward agent key request for channel %s" % (channel,))
+        return True
+
     def check_channel_exec_request(self, channel, cmd):
         logger.debug("Got exec request on channel %s for cmd %s" % (channel, cmd,))
         # Remove any 'bash -c' and/or quotes from command

tests/test_ssh_client.py

@@ -22,13 +22,18 @@
 from pssh import SSHClient, ParallelSSHClient, UnknownHostException, AuthenticationException, _setup_logger, logger
 from fake_server.fake_server import start_server, make_socket, logger as server_logger, \
     paramiko_logger
+from fake_server.fake_agent import FakeAgent
+import paramiko
 import os
 from test_pssh_client import USER_KEY
 
 # _setup_logger(server_logger)
 # _setup_logger(logger)
 # _setup_logger(paramiko_logger)
 
+USER_KEY = paramiko.RSAKey.from_private_key_file(
+    os.path.sep.join([os.path.dirname(__file__), 'test_client_private_key']))
+
 class SSHClientTest(unittest.TestCase):
 
     def setUp(self):
@@ -70,5 +75,27 @@ def test_ssh_client_sftp(self):
         del client
         server.join()
 
+    def test_ssh_agent_authentication(self):
+        """Test authentication via SSH agent.
+        Do not provide public key to use when creating SSHClient,
+        instead override the client's agent with our own fake SSH agent,
+        add our to key to agent and try to login to server.
+        Key should be automatically picked up from the overriden agent"""
+        agent = FakeAgent()
+        agent.add_key(USER_KEY)
+        server = start_server({ self.fake_cmd : self.fake_resp },
+                                self.listen_socket)
+        client = SSHClient('127.0.0.1', port=self.listen_port)
+        client.client._agent = agent
+        channel, host, _stdout, _stderr = client.exec_command(self.fake_cmd)
+        output = (line.strip() for line in _stdout)
+        channel.close()
+        output = list(output)
+        expected = [self.fake_resp]
+        self.assertEqual(expected, output,
+                         msg = "Got unexpected command output - %s" % (output,))
+        del client
+        server.join()
+
 if __name__ == '__main__':
     unittest.main()