Disable SSH agent when a tunnel identity file/password is provided (#9814)

web/pgadmin/utils/driver/psycopg3/server_manager.py

@@ -587,12 +587,23 @@ def create_ssh_tunnel(self, tunnel_password):
                 ssh_logger.setLevel(logging.DEBUG)
                 for h in current_app.logger.handlers:
                     ssh_logger.addHandler(h)
+            # Keep sshtunnel away from the agent whenever the user has
+            # configured a credential of their own (#9814). The identity file
+            # check uses the configured path, not the resolved one, so a
+            # missing key file is reported as a failure rather than silently
+            # authenticating with some other key from the agent. With no
+            # identity file or password configured the agent stays enabled,
+            # preserving the previous behaviour for anyone relying on
+            # "Prompt for Password?" or on the agent itself.
             if self.tunnel_authentication == 1:
+                tunnel_identity_file = get_complete_file_path(
+                    self.tunnel_identity_file)
                 self.tunnel_object = SSHTunnelForwarder(
                     (self.tunnel_host, int(self.tunnel_port)),
                     ssh_username=self.tunnel_username,
-                    ssh_pkey=get_complete_file_path(self.tunnel_identity_file),
+                    ssh_pkey=tunnel_identity_file,
                     ssh_private_key_password=tunnel_password,
+                    allow_agent=not self.tunnel_identity_file,
                     remote_bind_address=(self.host, self.port),
                     logger=ssh_logger,
                     set_keepalive=int(self.tunnel_keep_alive)
@@ -602,6 +613,7 @@ def create_ssh_tunnel(self, tunnel_password):
                     (self.tunnel_host, int(self.tunnel_port)),
                     ssh_username=self.tunnel_username,
                     ssh_password=tunnel_password,
+                    allow_agent=not tunnel_password,
                     remote_bind_address=(self.host, self.port),
                     logger=ssh_logger,
                     set_keepalive=int(self.tunnel_keep_alive)
@@ -610,7 +622,11 @@ def create_ssh_tunnel(self, tunnel_password):
             self.tunnel_object.daemon_forward_servers = True
             self.tunnel_object.start()
             self.tunnel_created = True
-        except BaseSSHTunnelForwarderError as e:
+        except (BaseSSHTunnelForwarderError, ValueError) as e:
+            # sshtunnel raises a bare ValueError rather than one of its own
+            # exceptions when it finds no credential to authenticate with, so
+            # catch that too and report it the same way instead of letting it
+            # escape as an unhandled exception.
             current_app.logger.exception(e)
             return False, gettext(
                 "Failed to create the SSH tunnel. Possible causes:\n"

web/pgadmin/utils/tests/test_ssh_tunnel_allow_agent.py

@@ -0,0 +1,145 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""Tests for the SSH agent handling in ServerManager.create_ssh_tunnel().
+
+Probing the agent when pgAdmin already has an identity file or a password of
+its own produces repeated prompts or denials (#9814), so the agent is disabled
+whenever one is configured, even if the identity file turns out to be missing,
+since falling back to the agent would authenticate with some other key. It
+must stay enabled when no credential is configured at all, because
+sshtunnel raises ValueError from _consolidate_auth() if it is left with
+nothing at all to authenticate with, and that would be a worse failure than
+the one being fixed: ValueError is not a BaseSSHTunnelForwarderError, so it
+would escape the handler that turns tunnel failures into a friendly message.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import config
+from pgadmin.utils.route import BaseTestGenerator
+
+import pgadmin.utils.driver.psycopg3.server_manager as server_manager
+
+
+class SSHTunnelAllowAgentTestCase(BaseTestGenerator):
+    """allow_agent must follow whether a credential is actually present."""
+
+    scenarios = [
+        ('Identity file present disables the agent', dict(
+            tunnel_authentication=1,
+            resolved_identity_file='/tmp/id_rsa',
+            stored_password=None,
+            expected_allow_agent=False,
+        )),
+        ('Missing identity file still disables the agent', dict(
+            tunnel_authentication=1,
+            resolved_identity_file=None,
+            stored_password=None,
+            expected_allow_agent=False,
+        )),
+        ('No identity file configured leaves the agent enabled', dict(
+            tunnel_authentication=1,
+            configured_identity_file=None,
+            resolved_identity_file=None,
+            stored_password=None,
+            expected_allow_agent=True,
+        )),
+        ('Tunnel password disables the agent', dict(
+            tunnel_authentication=0,
+            resolved_identity_file=None,
+            stored_password='encrypted',
+            expected_allow_agent=False,
+        )),
+        ('No credential at all leaves the agent enabled', dict(
+            tunnel_authentication=0,
+            resolved_identity_file=None,
+            stored_password=None,
+            expected_allow_agent=True,
+        )),
+        ('An empty tunnel password leaves the agent enabled', dict(
+            tunnel_authentication=0,
+            resolved_identity_file=None,
+            stored_password='',
+            expected_allow_agent=True,
+        )),
+        ('A missing credential is reported, not raised', dict(
+            tunnel_authentication=0,
+            resolved_identity_file=None,
+            stored_password=None,
+            expected_allow_agent=True,
+            forwarder_error=ValueError(
+                'No password or public key available!'),
+        )),
+    ]
+
+    # Overridden per scenario where the forwarder is meant to fail, or where
+    # no identity file is configured.
+    forwarder_error = None
+    configured_identity_file = 'id_rsa'
+
+    def setUp(self):
+        # Deliberately no server connection: this exercises the argument
+        # marshalling in create_ssh_tunnel(), not a real tunnel.
+        if not config.SUPPORT_SSH_TUNNEL:
+            self.skipTest('SSH tunnelling is disabled in this configuration.')
+
+    def _make_manager(self):
+        manager = server_manager.ServerManager.__new__(
+            server_manager.ServerManager)
+        manager.tunnel_authentication = self.tunnel_authentication
+        manager.tunnel_host = 'tunnel.example.com'
+        manager.tunnel_port = 22
+        manager.tunnel_username = 'tunneluser'
+        manager.tunnel_identity_file = self.configured_identity_file
+        manager.tunnel_keep_alive = 0
+        manager.host = 'db.example.com'
+        manager.port = 5432
+        manager.tunnel_object = None
+        manager.tunnel_created = False
+        return manager
+
+    def runTest(self):
+        manager = self._make_manager()
+        forwarder = MagicMock(side_effect=self.forwarder_error) \
+            if self.forwarder_error else MagicMock()
+
+        # A request context, not just an app context: the failure path calls
+        # gettext(), and pgAdmin's locale selector reads the request.
+        with self.app.test_request_context(), \
+            patch.object(server_manager, 'SSHTunnelForwarder', forwarder), \
+            patch.object(server_manager, 'User', MagicMock()), \
+            patch.object(server_manager, 'current_user', MagicMock()), \
+            patch.object(server_manager, 'get_complete_file_path',
+                         return_value=self.resolved_identity_file), \
+            patch.object(server_manager, 'get_crypt_key',
+                         return_value=(True, 'crypt-key')), \
+                patch.object(server_manager, 'decrypt',
+                             return_value=b'tunnelpassword'):
+            success, error = manager.create_ssh_tunnel(self.stored_password)
+
+        forwarder.assert_called_once()
+        kwargs = forwarder.call_args.kwargs
+        self.assertEqual(kwargs['allow_agent'], self.expected_allow_agent)
+
+        # The credential itself must reach sshtunnel too, not just the flag.
+        if self.tunnel_authentication == 1:
+            self.assertEqual(kwargs['ssh_pkey'], self.resolved_identity_file)
+        else:
+            # A stored password is decrypted; an absent or empty one is
+            # passed through untouched.
+            expected_password = 'tunnelpassword' \
+                if self.stored_password else self.stored_password
+            self.assertEqual(kwargs['ssh_password'], expected_password)
+
+        if self.forwarder_error:
+            self.assertFalse(success)
+            self.assertIn('Failed to create the SSH tunnel', error)
+        else:
+            self.assertTrue(success, msg=error)