Enable .pgpass support for SSH tunnel connections

changelog.rst

@@ -9,6 +9,10 @@ Features:
   Uses prompt_toolkit's ``ModalCursorShapeConfig``.
 * Add support of Python 3.14.
 * Drop support of Python 3.9.
+* Enable ``.pgpass`` support for SSH tunnel connections.
+    * Preserve original hostname for ``.pgpass`` lookup using PostgreSQL's ``hostaddr`` parameter
+    * SSH tunnel endpoint (``127.0.0.1``) is passed via ``hostaddr``, keeping ``host`` for ``.pgpass``
+    * Works with both DSN and host/port connection styles
 
 Bug fixes:
 ----------

pgcli/main.py

@@ -717,11 +717,15 @@ def should_ask_for_password(exc):
             self.logger.handlers = logger_handlers
 
             atexit.register(self.ssh_tunnel.stop)
-            host = "127.0.0.1"
+            # Preserve original host for .pgpass lookup and SSL certificate verification.
+            # Use hostaddr to specify the actual connection endpoint (SSH tunnel).
+            hostaddr = "127.0.0.1"
             port = self.ssh_tunnel.local_bind_ports[0]
 
             if dsn:
-                dsn = make_conninfo(dsn, host=host, port=port)
+                dsn = make_conninfo(dsn, host=host, hostaddr=hostaddr, port=port)
+            else:
+                kwargs["hostaddr"] = hostaddr
 
         # Attempt to connect to the database.
         # Note that passwd may be empty on the first attempt. If connection

pgcli/pgexecute.py

@@ -212,7 +212,8 @@ def connect(
         new_params.update(kwargs)
 
         if new_params["dsn"]:
-            new_params = {"dsn": new_params["dsn"], "password": new_params["password"]}
+            # When using DSN, only keep dsn, password, and hostaddr (for SSH tunnels)
+            new_params = {k: v for k, v in new_params.items() if k in ("dsn", "password", "hostaddr")}
 
             if new_params["password"]:
                 new_params["dsn"] = make_conninfo(new_params["dsn"], password=new_params.pop("password"))

tests/test_ssh_tunnel.py

@@ -6,7 +6,7 @@
 from click.testing import CliRunner
 from sshtunnel import SSHTunnelForwarder
 
-from pgcli.main import cli, notify_callback, PGCli
+from pgcli.main import cli, PGCli
 from pgcli.pgexecute import PGExecute
 
 
@@ -50,15 +50,13 @@ def test_ssh_tunnel(mock_ssh_tunnel_forwarder: MagicMock, mock_pgexecute: MagicM
     mock_pgexecute.assert_called_once()
 
     call_args, call_kwargs = mock_pgexecute.call_args
-    assert call_args == (
-        db_params["database"],
-        db_params["user"],
-        db_params["passwd"],
-        "127.0.0.1",
-        pgcli.ssh_tunnel.local_bind_ports[0],
-        "",
-        notify_callback,
-    )
+    # Original host is preserved for .pgpass lookup, hostaddr has tunnel endpoint
+    assert call_args[0] == db_params["database"]
+    assert call_args[1] == db_params["user"]
+    assert call_args[2] == db_params["passwd"]
+    assert call_args[3] == db_params["host"]  # original host preserved
+    assert call_args[4] == pgcli.ssh_tunnel.local_bind_ports[0]
+    assert call_kwargs.get("hostaddr") == "127.0.0.1"
     mock_ssh_tunnel_forwarder.reset_mock()
     mock_pgexecute.reset_mock()
 
@@ -86,15 +84,9 @@ def test_ssh_tunnel(mock_ssh_tunnel_forwarder: MagicMock, mock_pgexecute: MagicM
     mock_pgexecute.assert_called_once()
 
     call_args, call_kwargs = mock_pgexecute.call_args
-    assert call_args == (
-        db_params["database"],
-        db_params["user"],
-        db_params["passwd"],
-        "127.0.0.1",
-        pgcli.ssh_tunnel.local_bind_ports[0],
-        "",
-        notify_callback,
-    )
+    assert call_args[3] == db_params["host"]  # original host preserved
+    assert call_args[4] == pgcli.ssh_tunnel.local_bind_ports[0]
+    assert call_kwargs.get("hostaddr") == "127.0.0.1"
     mock_ssh_tunnel_forwarder.reset_mock()
     mock_pgexecute.reset_mock()
 
@@ -104,13 +96,51 @@ def test_ssh_tunnel(mock_ssh_tunnel_forwarder: MagicMock, mock_pgexecute: MagicM
     pgcli = PGCli(ssh_tunnel_url=tunnel_url)
     pgcli.connect(dsn=dsn)
 
-    expected_dsn = f"user={db_params['user']} password={db_params['passwd']} host=127.0.0.1 port={pgcli.ssh_tunnel.local_bind_ports[0]}"
-
     mock_ssh_tunnel_forwarder.assert_called_once_with(**expected_tunnel_params)
     mock_pgexecute.assert_called_once()
 
     call_args, call_kwargs = mock_pgexecute.call_args
-    assert expected_dsn in call_args
+    # DSN should contain original host AND hostaddr for tunnel
+    dsn_arg = call_args[5]
+    assert f"host={db_params['host']}" in dsn_arg
+    assert "hostaddr=127.0.0.1" in dsn_arg
+    assert f"port={pgcli.ssh_tunnel.local_bind_ports[0]}" in dsn_arg
+
+
+def test_ssh_tunnel_preserves_original_host_for_pgpass(mock_ssh_tunnel_forwarder: MagicMock, mock_pgexecute: MagicMock) -> None:
+    """Verify that the original hostname is preserved for .pgpass lookup."""
+    tunnel_url = "bastion.example.com"
+    original_host = "production.db.example.com"
+
+    pgcli = PGCli(ssh_tunnel_url=tunnel_url)
+    pgcli.connect(database="mydb", host=original_host, user="dbuser", passwd="dbpass")
+
+    call_args, call_kwargs = mock_pgexecute.call_args
+    assert call_args[3] == original_host  # host preserved
+    assert call_kwargs.get("hostaddr") == "127.0.0.1"  # tunnel endpoint
+
+
+def test_ssh_tunnel_with_dsn_preserves_host(mock_ssh_tunnel_forwarder: MagicMock, mock_pgexecute: MagicMock) -> None:
+    """DSN connections should include hostaddr for tunnel while preserving host."""
+    tunnel_url = "bastion.example.com"
+    dsn = "host=production.db.example.com port=5432 dbname=mydb user=dbuser"
+
+    pgcli = PGCli(ssh_tunnel_url=tunnel_url)
+    pgcli.connect(dsn=dsn)
+
+    call_args, call_kwargs = mock_pgexecute.call_args
+    dsn_arg = call_args[5]
+    assert "host=production.db.example.com" in dsn_arg
+    assert "hostaddr=127.0.0.1" in dsn_arg
+
+
+def test_no_ssh_tunnel_does_not_set_hostaddr(mock_pgexecute: MagicMock) -> None:
+    """Without SSH tunnel, hostaddr should not be set."""
+    pgcli = PGCli()
+    pgcli.connect(database="mydb", host="localhost", user="user", passwd="pass")
+
+    call_args, call_kwargs = mock_pgexecute.call_args
+    assert "hostaddr" not in call_kwargs
 
 
 def test_cli_with_tunnel() -> None: