Enable .pgpass support for SSH tunnel connections

changelog.rst

@@ -9,6 +9,11 @@ Features:
     * Support dsn specific init-command in the config file
 * Add suggestion when setting the search_path
 * Allow per dsn_alias ssh tunnel selection
+* Enable .pgpass support for SSH tunnel connections
+    * Preserve original hostname for .pgpass lookup when using SSH tunnels
+    * Use PostgreSQL's `hostaddr` parameter to specify tunnel endpoint
+    * Add SSH configuration options (ssh_config_file, allow_agent, compression)
+    * `.pgpass` file now works seamlessly with `--ssh-tunnel` option
 
 Internal:
 ---------

pgcli/main.py

@@ -594,7 +594,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.
@@ -667,6 +668,9 @@ def should_ask_for_password(exc):
                 "remote_bind_address": (host, int(port or 5432)),
                 "ssh_address_or_host": (tunnel_info.hostname, tunnel_info.port or 22),
                 "logger": self.logger,
+                "ssh_config_file": "~/.ssh/config",  # Use SSH config for host settings
+                "allow_agent": True,  # Allow SSH agent for authentication
+                "compression": False,  # Disable compression for better performance
             }
             if tunnel_info.username:
                 params["ssh_username"] = tunnel_info.username
@@ -687,11 +691,16 @@ 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:
+                # For non-DSN connections, pass hostaddr via kwargs
+                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"]}
+            # Preserve hostaddr when using DSN (needed for SSH tunnels with .pgpass)
+            preserved_params = {"dsn": new_params["dsn"], "password": new_params["password"]}
+            if "hostaddr" in new_params:
+                preserved_params["hostaddr"] = new_params["hostaddr"]
+            new_params = preserved_params
 
             if new_params["password"]:
                 new_params["dsn"] = make_conninfo(new_params["dsn"], password=new_params.pop("password"))