Enable .pgpass support for SSH tunnel connections

Preserve original hostname for .pgpass lookup using PostgreSQL's
host/hostaddr parameters: host keeps the original DB hostname (for
.pgpass and SSL), hostaddr gets 127.0.0.1 (the tunnel endpoint).

Changes:
- main.py: Use hostaddr instead of replacing host with 127.0.0.1
- main.py: Pass dsn in connect_uri() for proper .pgpass handling
- pgexecute.py: Simplify DSN filtering to keep dsn, password, hostaddr
- tests: Add 4 new tests, update existing to verify host preservation

Made with ❤️ and 🤖 Claude

Author: DiegoDAF

changelog.rst

@@ -4,6 +4,10 @@ Upcoming (TBD)
 Features:
 ---------
 * Add support for `\\T` prompt escape sequence to display transaction status (similar to psql's `%x`).
+* 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
 
 4.4.0 (2025-12-24)
 ==================

pgcli/main.py

@@ -616,7 +616,8 @@ def connect_uri(self, uri):
         kwargs = conninfo_to_dict(uri)
         remap = {"dbname": "database", "password": "passwd"}
         kwargs = {remap.get(k, k): v for k, v in kwargs.items()}
-        self.connect(**kwargs)
+        # Pass the original URI as dsn parameter for .pgpass support with SSH tunnels
+        self.connect(dsn=uri, **kwargs)
 
     def connect(self, database="", host="", user="", port="", passwd="", dsn="", **kwargs):
         # Connect to the database.
@@ -709,11 +710,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,11 @@ 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

@@ -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"]  # database
+    assert call_args[1] == db_params["user"]  # user
+    assert call_args[2] == db_params["passwd"]  # passwd
+    assert call_args[3] == db_params["host"]  # original host preserved
+    assert call_args[4] == pgcli.ssh_tunnel.local_bind_ports[0]  # tunnel port
+    assert call_kwargs.get("hostaddr") == "127.0.0.1"  # tunnel endpoint via hostaddr
     mock_ssh_tunnel_forwarder.reset_mock()
     mock_pgexecute.reset_mock()
 
@@ -86,15 +84,10 @@ 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, hostaddr has tunnel endpoint
+    assert call_args[3] == db_params["host"]  # original host preserved
+    assert call_args[4] == pgcli.ssh_tunnel.local_bind_ports[0]  # tunnel port
+    assert call_kwargs.get("hostaddr") == "127.0.0.1"
     mock_ssh_tunnel_forwarder.reset_mock()
     mock_pgexecute.reset_mock()
 
@@ -104,13 +97,78 @@ 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]  # dsn is 6th positional arg
+    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
+    # host parameter should be the original, not 127.0.0.1
+    assert call_args[3] == original_host
+    # hostaddr should be the tunnel endpoint
+    assert call_kwargs.get("hostaddr") == "127.0.0.1"
+
+
+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_connect_uri_without_ssh_tunnel(mock_pgexecute: MagicMock) -> None:
+    """connect_uri should work normally without SSH tunnel."""
+    pgcli = PGCli()
+    pgcli.connect_uri("postgresql://user:pass@localhost/mydb")
+
+    mock_pgexecute.assert_called_once()
+    call_args, call_kwargs = mock_pgexecute.call_args
+    assert "hostaddr" not in call_kwargs
+
+
+def test_connect_uri_passes_dsn(mock_pgexecute: MagicMock) -> None:
+    """connect_uri should pass the URI as dsn parameter."""
+    uri = "postgresql://user:pass@localhost/mydb"
+    pgcli = PGCli()
+    pgcli.connect_uri(uri)
+
+    call_args, call_kwargs = mock_pgexecute.call_args
+    # dsn is passed as the 6th positional arg to PGExecute.__init__
+    assert call_args[5] == uri
 
 
 def test_cli_with_tunnel() -> None: