Add support for an explicit path for agent_forwarding and ForwardAgent

This commit adds support for setting an exlpicit path to forward SSH
agent requests to, rather than having them always be forward to the
samt path as client requests. This support is also available through
the ForwardAgent config option, and that also provides support for
SSH config percent expansion.

Author: ronf

asyncssh/config.py

@@ -219,6 +219,22 @@ def _set_bool(self, option: str, args: List[str]) -> None:
         if option not in self._options:
             self._options[option] = value
 
+    def _set_bool_or_str(self, option: str, args: List[str]) -> None:
+        """Set a boolean or string config option"""
+
+        value_str = args.pop(0)
+        value_lower = value_str.lower()
+
+        if value_lower in ('yes', 'true'):
+            value: Union[bool, str] = True
+        elif value_lower in ('no', 'false'):
+            value = False
+        else:
+            value = value_str
+
+        if option not in self._options:
+            self._options[option] = value
+
     def _set_int(self, option: str, args: List[str]) -> None:
         """Set an integer config option"""
 
@@ -468,7 +484,7 @@ class SSHClientConfig(SSHConfig):
 
     _conditionals = {'host', 'match'}
     _no_split = {'proxycommand', 'remotecommand'}
-    _percent_expand = {'CertificateFile', 'IdentityAgent',
+    _percent_expand = {'CertificateFile', 'ForwardAgent', 'IdentityAgent',
                        'IdentityFile', 'ProxyCommand', 'RemoteCommand'}
 
     def __init__(self, last_config: 'SSHConfig', reload: bool,
@@ -587,7 +603,7 @@ def _set_tokens(self) -> None:
         ('Compression',                     SSHConfig._set_bool),
         ('ConnectTimeout',                  SSHConfig._set_int),
         ('EnableSSHKeySign',                SSHConfig._set_bool),
-        ('ForwardAgent',                    SSHConfig._set_bool),
+        ('ForwardAgent',                    SSHConfig._set_bool_or_str),
         ('ForwardX11Trusted',               SSHConfig._set_bool),
         ('GlobalKnownHostsFile',            SSHConfig._set_string_list),
         ('GSSAPIAuthentication',            SSHConfig._set_bool),

asyncssh/connection.py

@@ -7640,8 +7640,11 @@ class SSHClientConnectionOptions(SSHConnectionOptions):
            made available for use. This is the default.
        :param agent_forwarding: (optional)
            Whether or not to allow forwarding of ssh-agent requests from
-           processes running on the server. By default, ssh-agent forwarding
-           requests from the server are not allowed.
+           processes running on the server. This argument can also be set
+           to the path of a UNIX domain socket in cases where forwarded
+           agent requests should be sent to a different path than client
+           agent requests. By default, forwarding ssh-agent requests from
+           the server is not allowed.
        :param pkcs11_provider: (optional)
            The path of a shared library which should be used as a PKCS#11
            provider for accessing keys on PIV security tokens. By default,
@@ -7874,7 +7877,7 @@ class SSHClientConnectionOptions(SSHConnectionOptions):
        :type agent_path: `str`
        :type agent_identities:
            *see* :ref:`SpecifyingPublicKeys` and :ref:`SpecifyingCertificates`
-       :type agent_forwarding: `bool`
+       :type agent_forwarding: `bool` or `str`
        :type pkcs11_provider: `str` or `None`
        :type pkcs11_pin: `str`
        :type client_version: `str`
@@ -8016,7 +8019,7 @@ def prepare(self, # type: ignore
                 disable_trivial_auth: bool = False,
                 agent_path: DefTuple[Optional[str]] = (),
                 agent_identities: DefTuple[Optional[IdentityListArg]] = (),
-                agent_forwarding: DefTuple[bool] = (),
+                agent_forwarding: DefTuple[Union[bool, str]] = (),
                 pkcs11_provider: DefTuple[Optional[str]] = (),
                 pkcs11_pin: Optional[str] = None,
                 command: DefTuple[Optional[str]] = (),
@@ -8242,9 +8245,17 @@ def prepare(self, # type: ignore
             self.pkcs11_pin = None
 
         if agent_forwarding == ():
-            agent_forwarding = cast(bool, config.get('ForwardAgent', False))
+            agent_forwarding = cast(Union[bool, str],
+                                    config.get('ForwardAgent', False))
 
-        self.agent_forward_path = agent_path if agent_forwarding else None
+        agent_forwarding: Union[bool, str]
+
+        if not agent_forwarding:
+            self.agent_forward_path = None
+        elif agent_forwarding is True:
+            self.agent_forward_path = agent_path
+        else:
+            self.agent_forward_path = agent_forwarding
 
         self.command = cast(Optional[str], command if command != () else
             config.get('RemoteCommand'))

tests/test_channel.py

@@ -862,14 +862,14 @@ async def test_unneeded_resume_reading(self):
             chan.close()
 
     @asynctest
-    async def test_agent_forwarding(self):
-        """Test SSH agent forwarding"""
+    async def test_agent_forwarding_explicit(self):
+        """Test SSH agent forwarding with explicit path"""
 
         if not self.agent_available(): # pragma: no cover
             self.skipTest('ssh-agent not available')
 
         async with self.connect(username='ckey',
-                                agent_forwarding=True) as conn:
+                                agent_forwarding='agent') as conn:
             chan, session = await _create_session(conn, 'agent')
 
             await chan.wait_closed()

tests/test_config.py

@@ -335,6 +335,18 @@ def test_set_remote_command(self):
         config = self._parse_config('    RemoteCommand     foo  bar  baz')
         self.assertEqual(config.get('RemoteCommand'), 'foo  bar  baz')
 
+    def test_set_forward_agent(self):
+        """Test agent forwarding path config option"""
+
+        for value, result in (('yes', True), ('true', True),
+                              ('no', False), ('false', False),
+                              ('agent', 'agent'), ('%d/agent', './agent')):
+            config = self._parse_config(f'ForwardAgent {value}')
+            self.assertEqual(config.get('ForwardAgent'), result)
+
+        config = self._parse_config('ForwardAgent yes\nForwardAgent no')
+        self.assertEqual(config.get('ForwardAgent'), True)
+
     def test_set_request_tty(self):
         """Test pseudo-terminal request config option"""