diff --git a/web/pgadmin/utils/driver/psycopg3/server_manager.py b/web/pgadmin/utils/driver/psycopg3/server_manager.py index 00738355ee4..ea1b2995daf 100644 --- a/web/pgadmin/utils/driver/psycopg3/server_manager.py +++ b/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" diff --git a/web/pgadmin/utils/tests/test_ssh_tunnel_allow_agent.py b/web/pgadmin/utils/tests/test_ssh_tunnel_allow_agent.py new file mode 100644 index 00000000000..69210076fc0 --- /dev/null +++ b/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)