Allow to disable ssh agent usage

Author: penguinolog

doc/source/SSHClient.rst

@@ -10,7 +10,7 @@ API: SSHClient and SSHAuth.
 
     SSHClient helper.
 
-    .. py:method:: __init__(host, port=22, username=None, password=None, *, auth=None, verbose=True, ssh_config=None, ssh_auth_map=None, sock=None, keepalive=1)
+    .. py:method:: __init__(host, port=22, username=None, password=None, *, auth=None, verbose=True, ssh_config=None, ssh_auth_map=None, sock=None, keepalive=1, allow_ssh_agent=True)
 
         :param host: remote hostname
         :type host: ``str``
@@ -32,6 +32,8 @@ API: SSHClient and SSHAuth.
         :type sock: paramiko.ProxyCommand | paramiko.Channel | socket.socket | None
         :param keepalive: keepalive period
         :type keepalive: int | bool
+        :param allow_ssh_agent: use SSH Agent if available
+        :type allow_ssh_agent: bool
 
         .. note:: auth has priority over username/password/private_keys
         .. note::
@@ -49,6 +51,7 @@ API: SSHClient and SSHAuth.
         .. versionchanged:: 7.0.0 private_keys is removed
         .. versionchanged:: 7.0.0 keepalive_mode is removed
         .. versionchanged:: 7.4.0 return of keepalive_mode to prevent mix with keepalive period. Default is `False`
+        .. versionchanged:: 8.0.0 expose SSH Agent usage override
 
     .. py:attribute:: log_mask_re
 
@@ -102,6 +105,11 @@ API: SSHClient and SSHAuth.
         ``int | bool``
         Keepalive period for connection object.
 
+    .. py:attribute:: use_ssh_agent
+
+        ``bool``
+        Use SSH Agent if available.
+
     .. py:method:: close()
 
         Close connection

exec_helpers/_ssh_base.py

@@ -454,6 +454,8 @@ class SSHClientBase(api.ExecHelper):
     :type sock: paramiko.ProxyCommand | paramiko.Channel | socket.socket | None
     :param keepalive: keepalive period
     :type keepalive: int | bool
+    :param allow_ssh_agent: use SSH Agent if available
+    :type allow_ssh_agent: bool
 
     .. note:: auth has priority over username/password/private_keys
     .. note::
@@ -471,6 +473,7 @@ class SSHClientBase(api.ExecHelper):
     .. versionchanged:: 7.0.0 private_keys is removed
     .. versionchanged:: 7.0.0 keepalive_mode is removed
     .. versionchanged:: 7.4.0 return of keepalive_mode to prevent mix with keepalive period. Default is `False`
+    .. versionchanged:: 8.0.0 expose SSH Agent usage override
     """
 
     __slots__ = (
@@ -486,6 +489,7 @@ class SSHClientBase(api.ExecHelper):
         "__ssh_config",
         "__sock",
         "__conn_chain",
+        "__allow_agent",
     )
 
     def __hash__(self) -> int:
@@ -509,8 +513,19 @@ def __init__(
         ssh_auth_map: dict[str, ssh_auth.SSHAuth] | ssh_auth.SSHAuthMapping | None = None,
         sock: paramiko.ProxyCommand | paramiko.Channel | socket.socket | None = None,
         keepalive: KeepAlivePeriodT = 1,
+        allow_ssh_agent: bool = True,
     ) -> None:
         """Main SSH Client helper."""
+        self.__sudo_mode = False
+        self.__keepalive_period: int = int(keepalive)
+        self.__keepalive_mode = False
+        self.__verbose: bool = verbose
+        self.__sock = sock
+
+        self.__ssh: paramiko.SSHClient
+        self.__sftp: paramiko.SFTPClient | None = None
+        self.__allow_agent = allow_ssh_agent
+
         # Init ssh config. It's main source for connection parameters
         if isinstance(ssh_config, _ssh_helpers.HostsSSHConfigs):
             self.__ssh_config: _ssh_helpers.HostsSSHConfigs = ssh_config
@@ -533,35 +548,25 @@ def __init__(
         if self.hostname not in self.__auth_mapping and host in self.__auth_mapping:
             self.__auth_mapping[self.hostname] = self.__auth_mapping[host]
 
-        self.__sudo_mode = False
-        self.__keepalive_period: int = int(keepalive)
-        self.__keepalive_mode = False
-        self.__verbose: bool = verbose
-        self.__sock = sock
-
-        self.__ssh: paramiko.SSHClient
-        self.__sftp: paramiko.SFTPClient | None = None
-
         # Rebuild SSHAuth object if required.
         # Priority: auth > credentials > auth mapping
-        if auth is not None:
-            self.__auth_mapping[self.hostname] = real_auth = copy.copy(auth)
-        elif self.hostname not in self.__auth_mapping or any((username, password)):
-            self.__auth_mapping[self.hostname] = real_auth = ssh_auth.SSHAuth(
-                username=username if username is not None else config.user,
-                password=password,
-                key_filename=config.identityfile,
-            )
-        else:
-            real_auth = self.__auth_mapping[self.hostname]
+        real_auth = self.__handle_explicit_auth(
+            username=username,
+            config_username=config.user,
+            password=password,
+            auth=auth,
+            key_filename=config.identityfile,
+        )
 
         # Init super with host and real port and username
         mod_name = "exec_helpers" if self.__module__.startswith("exec_helpers") else self.__module__
         log_username: str = real_auth.username if real_auth.username is not None else getpass.getuser()
 
         super().__init__(
-            logger=logging.getLogger(f"{mod_name}.{self.__class__.__name__}").getChild(
-                f"({log_username}@{host}:{self.port})"
+            logger=logging.getLogger(
+                f"{mod_name}.{self.__class__.__name__}",
+            ).getChild(
+                f"({log_username}@{host}:{self.port})",
             )
         )
 
@@ -577,6 +582,26 @@ def __init__(
 
         self.__connect()
 
+    def __handle_explicit_auth(
+        self,
+        *,
+        username: str | None,
+        config_username: str | None,
+        password: str | None,
+        auth: ssh_auth.SSHAuth | None,
+        key_filename: Iterable[str] | None,
+    ) -> ssh_auth.SSHAuth:
+        if auth is not None:
+            self.__auth_mapping[self.hostname] = auth
+        elif self.hostname not in self.__auth_mapping or any((username, password)):
+            self.__auth_mapping[self.hostname] = ssh_auth.SSHAuth(
+                username=username if username is not None else config_username,
+                password=password,
+                key_filename=key_filename,
+            )
+
+        return self.__auth_mapping[self.hostname]
+
     def __rebuild_ssh_config(self) -> None:
         """Rebuild main ssh config from available information."""
         self.__ssh_config[self.hostname] = self.__ssh_config[self.hostname].overridden_by(
@@ -598,7 +623,11 @@ def __build_connection_chain(self) -> list[tuple[_ssh_helpers.SSHConfig, ssh_aut
 
         config = self.ssh_config[self.hostname]
         default_auth = ssh_auth.SSHAuth(username=config.user, key_filename=config.identityfile)
-        auth = self.__auth_mapping.get_with_alt_hostname(config.hostname, self.hostname, default=default_auth)
+        auth = self.__auth_mapping.get_with_alt_hostname(
+            config.hostname,
+            self.hostname,
+            default=default_auth,
+        )
         conn_chain.append((config, auth))
 
         while config.proxyjump is not None:
@@ -621,6 +650,15 @@ def auth(self) -> ssh_auth.SSHAuth:
         """
         return self.__auth_mapping[self.hostname]
 
+    @property
+    def allow_ssh_agent(self) -> bool:
+        """Use SSH Agent if available.
+
+        :return: SSH Agent usage allowed
+        :rtype: bool
+        """
+        return self.__allow_agent
+
     @property
     def hostname(self) -> str:
         """Connected remote host name.
@@ -714,16 +752,15 @@ def __connect(self) -> None:
         """Main method for connection open."""
         with self.lock:
             if self.__sock is not None:
-                sock = self.__sock
-
                 self.__ssh = paramiko.SSHClient()
                 self.__ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                 self.auth.connect(
                     client=self.__ssh,
                     hostname=self.hostname,
                     port=self.port,
                     log=self.__verbose,
-                    sock=sock,
+                    sock=self.__sock,
+                    allow_ssh_agent=self.allow_ssh_agent,
                 )
             else:
                 self.__ssh = self.__get_client()
@@ -745,15 +782,14 @@ def __get_client(self) -> paramiko.SSHClient:
         last_ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 
         config, auth = self.__conn_chain[0]
-        if config.proxycommand:
-            auth.connect(
-                last_ssh_client,
-                hostname=config.hostname,
-                port=config.port or 22,
-                sock=paramiko.ProxyCommand(config.proxycommand),
-            )
-        else:
-            auth.connect(last_ssh_client, hostname=config.hostname, port=config.port or 22)
+
+        auth.connect(
+            last_ssh_client,
+            hostname=config.hostname,
+            port=config.port or 22,
+            sock=paramiko.ProxyCommand(config.proxycommand) if config.proxycommand else None,
+            allow_ssh_agent=self.allow_ssh_agent,
+        )
 
         for config, auth in self.__conn_chain[1:]:  # start has another logic, so do it out of cycle
             ssh = paramiko.SSHClient()
@@ -768,7 +804,13 @@ def __get_client(self) -> paramiko.SSHClient:
                     dest_addr=(config.hostname, config.port or 22),
                     src_addr=(config.proxyjump, 0),
                 )
-                auth.connect(ssh, hostname=config.hostname, port=config.port or 22, sock=sock)
+                auth.connect(
+                    ssh,
+                    hostname=config.hostname,
+                    port=config.port or 22,
+                    sock=sock,
+                    allow_ssh_agent=self.allow_ssh_agent,
+                )
                 last_ssh_client = ssh
                 continue
 
@@ -1421,33 +1463,6 @@ def check_stderr(
             **kwargs,
         )
 
-    def _get_proxy_channel(
-        self,
-        port: int | None,
-        ssh_config: _ssh_helpers.SSHConfig,
-    ) -> paramiko.Channel:
-        """Get ssh proxy channel.
-
-        :param port: target port
-        :type port: int | None
-        :param ssh_config: pre-parsed ssh config
-        :type ssh_config: SSHConfig
-        :return: ssh channel for usage as socket for new connection over it
-        :rtype: paramiko.Channel
-
-        .. versionadded:: 6.0.0
-        """
-        if port is not None:
-            dest_port: int = port
-        else:
-            dest_port = ssh_config.port if ssh_config.port is not None else 22
-
-        return self._ssh_transport.open_channel(
-            kind="direct-tcpip",
-            dest_addr=(ssh_config.hostname, dest_port),
-            src_addr=(self.hostname, 0),
-        )
-
     def proxy_to(
         self,
         host: str,
@@ -1498,13 +1513,25 @@ def proxy_to(
         else:
             parsed_ssh_config = _ssh_helpers.parse_ssh_config(ssh_config, host)
 
-        hostname = parsed_ssh_config[host].hostname
+        host_config = parsed_ssh_config[host]
+
+        if port is not None:
+            dest_port: int = port
+        elif host_config.port is not None:
+            dest_port = host_config.port
+        else:
+            dest_port = 22
+
+        sock: paramiko.Channel = self._ssh_transport.open_channel(
+            kind="direct-tcpip",
+            dest_addr=(host_config.hostname, dest_port),
+            src_addr=(self.hostname, 0),
+        )
 
-        sock: paramiko.Channel = self._get_proxy_channel(port=port, ssh_config=parsed_ssh_config[hostname])
         cls: type[Self] = self.__class__
         return cls(
             host=host,
-            port=port,
+            port=dest_port,
             username=username,
             password=password,
             auth=auth,

exec_helpers/ssh_auth.py

@@ -163,6 +163,7 @@ def connect(
         log: bool = True,
         *,
         sock: paramiko.ProxyCommand | paramiko.Channel | socket.socket | None = None,
+        allow_ssh_agent: bool = True,
     ) -> None:
         """Connect SSH client object using credentials.
 
@@ -176,6 +177,8 @@ def connect(
         :type log: bool
         :param sock: socket for connection. Useful for ssh proxies support
         :type sock: paramiko.ProxyCommand | paramiko.Channel | socket.socket | None
+        :param allow_ssh_agent: use SSH Agent if available
+        :type allow_ssh_agent: bool
         :raises PasswordRequiredException: No password has been set, but required.
         :raises AuthenticationException: Authentication failed.
         """
@@ -196,6 +199,7 @@ def connect(
                     username=self.username,
                     password=self.__password,
                     key_filename=self.__key_filename,  # type: ignore[arg-type]  # types verified by not signature
+                    allow_agent=allow_ssh_agent,
                     **kwargs,
                 )
                 if index != self.__key_index:

test/test_ssh_client_init_basic.py

@@ -56,13 +56,21 @@ def __iter__(self):
     "username_password": {"host": host, "username": "user", "password": "password"},
     "auth": {
         "host": host,
-        "auth": exec_helpers.SSHAuth(username="user", password="password", key=gen_private_keys(1).pop()),
+        "auth": exec_helpers.SSHAuth(
+            username="user",
+            password="password",
+            key=gen_private_keys(1).pop(),
+        ),
     },
     "auth_break": {
         "host": host,
         "username": "Invalid",
         "password": "Invalid",
-        "auth": exec_helpers.SSHAuth(username="user", password="password", key=gen_private_keys(1).pop()),
+        "auth": exec_helpers.SSHAuth(
+            username="user",
+            password="password",
+            key=gen_private_keys(1).pop(),
+        ),
     },
 }
 
@@ -90,7 +98,12 @@ def run_parameters(request):
     return configs[request.param]
 
 
-def test_init_base(paramiko_ssh_client, auto_add_policy, run_parameters, ssh_auth_logger):
+def test_init_base(
+    paramiko_ssh_client,
+    auto_add_policy,
+    run_parameters,
+    ssh_auth_logger,
+):
     """Parametrized validation of SSH client initialization."""
     # Helper code
     _ssh = mock.call
@@ -117,6 +130,7 @@ def test_init_base(paramiko_ssh_client, auto_add_policy, run_parameters, ssh_aut
                 port=port,
                 username=username,
                 key_filename=(),
+                allow_agent=True,
             ),
             _ssh.get_transport(),
             _ssh.get_transport().set_keepalive(1),