fix: do not leak pg password to command line (#560)

Author: liambuchanan

dbbackup/db/postgresql.py

@@ -7,20 +7,18 @@
 logger = logging.getLogger("dbbackup.command")
 
 
-def create_postgres_uri(self):
+def create_postgres_dbname_and_env(self):
     host = self.settings.get("HOST", "localhost")
     dbname = self.settings.get("NAME", "")
     user = quote(self.settings.get("USER") or "")
-    password = self.settings.get("PASSWORD", "")
-    password = f":{quote(password)}" if password else ""
-    if not user:
-        password = ""
-    else:
+    if user:
         host = "@" + host
-
     port = ":{}".format(self.settings.get("PORT")) if self.settings.get("PORT") else ""
-    dbname = f"--dbname=postgresql://{user}{password}{host}{port}/{dbname}"
-    return dbname
+    dbname = f"--dbname=postgresql://{user}{host}{port}/{dbname}"
+    env = {}
+    if self.settings.get("PASSWORD"):
+        env["PGPASSWORD"] = self.settings.get("PASSWORD")
+    return dbname, env
 
 
 class PgDumpConnector(BaseCommandDBConnector):
@@ -38,7 +36,8 @@ class PgDumpConnector(BaseCommandDBConnector):
 
     def _create_dump(self):
         cmd = f"{self.dump_cmd} "
-        cmd = cmd + create_postgres_uri(self)
+        dbname, pg_env = create_postgres_dbname_and_env(self)
+        cmd = cmd + dbname
 
         for table in self.exclude:
             cmd += f" --exclude-table-data={table}"
@@ -52,12 +51,13 @@ def _create_dump(self):
             cmd += " -n " + " -n ".join(self.schemas)
 
         cmd = f"{self.dump_prefix} {cmd} {self.dump_suffix}"
-        stdout, stderr = self.run_command(cmd, env=self.dump_env)
+        stdout, stderr = self.run_command(cmd, env={**self.dump_env, **pg_env})
         return stdout
 
     def _restore_dump(self, dump):
         cmd = f"{self.restore_cmd} "
-        cmd = cmd + create_postgres_uri(self)
+        dbname, pg_env = create_postgres_dbname_and_env(self)
+        cmd = cmd + dbname
 
         # without this, psql terminates with an exit value of 0 regardless of errors
         cmd += " --set ON_ERROR_STOP=on"
@@ -70,7 +70,9 @@ def _restore_dump(self, dump):
 
         cmd += " {}".format(self.settings["NAME"])
         cmd = f"{self.restore_prefix} {cmd} {self.restore_suffix}"
-        stdout, stderr = self.run_command(cmd, stdin=dump, env=self.restore_env)
+        stdout, stderr = self.run_command(
+            cmd, stdin=dump, env={**self.restore_env, **pg_env}
+        )
         return stdout, stderr
 
 
@@ -117,7 +119,8 @@ class PgDumpBinaryConnector(PgDumpConnector):
 
     def _create_dump(self):
         cmd = f"{self.dump_cmd} "
-        cmd = cmd + create_postgres_uri(self)
+        dbname, pg_env = create_postgres_dbname_and_env(self)
+        cmd = cmd + dbname
 
         cmd += " --format=custom"
         for table in self.exclude:
@@ -127,7 +130,7 @@ def _create_dump(self):
             cmd += " -n " + " -n ".join(self.schemas)
 
         cmd = f"{self.dump_prefix} {cmd} {self.dump_suffix}"
-        stdout, _ = self.run_command(cmd, env=self.dump_env)
+        stdout, _ = self.run_command(cmd, env={**self.dump_env, **pg_env})
         return stdout
 
     def _restore_dump(self, dump: str):
@@ -140,7 +143,7 @@ def _restore_dump(self, dump: str):
         Builds the command as a list.
         """
 
-        dbname = create_postgres_uri(self)
+        dbname, pg_env = create_postgres_dbname_and_env(self)
         cmd = []
 
         # Flatten optional values
@@ -188,6 +191,8 @@ def _restore_dump(self, dump: str):
             )
 
         cmd_str = " ".join(cmd)
-        stdout, _ = self.run_command(cmd_str, stdin=dump, env=self.dump_env)
+        stdout, _ = self.run_command(
+            cmd_str, stdin=dump, env={**self.dump_env, **pg_env}
+        )
 
         return stdout

dbbackup/tests/test_connectors/test_postgresql.py

@@ -21,15 +21,10 @@ def setUp(self):
         self.connector.settings["NAME"] = "dbname"
         self.connector.settings["HOST"] = "hostname"
 
-    def test_user_password_uses_special_characters(self, mock_dump_cmd):
-        self.connector.settings["PASSWORD"] = "@!"
+    def test_user_uses_special_characters(self, mock_dump_cmd):
         self.connector.settings["USER"] = "@"
-
         self.connector.create_dump()
-
-        self.assertIn(
-            "postgresql://%40:%40%21@hostname/dbname", mock_dump_cmd.call_args[0][0]
-        )
+        self.assertIn("postgresql://%40@hostname/dbname", mock_dump_cmd.call_args[0][0])
 
     def test_create_dump(self, mock_dump_cmd):
         dump = self.connector.create_dump()
@@ -166,6 +161,17 @@ def test_restore_dump_schema(self, mock_dump_cmd, mock_restore_cmd):
         self.assertIn(" -n public", mock_restore_cmd.call_args[0][0])
         self.assertIn(" -n foo", mock_restore_cmd.call_args[0][0])
 
+    def test_create_dump_password_excluded(self, mock_dump_cmd):
+        self.connector.settings["PASSWORD"] = "secret"
+        self.connector.create_dump()
+        self.assertNotIn("secret", mock_dump_cmd.call_args[0][0])
+
+    def test_restore_dump_password_excluded(self, mock_dump_cmd):
+        self.connector.settings["PASSWORD"] = "secret"
+        dump = self.connector.create_dump()
+        self.connector.restore_dump(dump)
+        self.assertNotIn("secret", mock_dump_cmd.call_args[0][0])
+
 
 @patch(
     "dbbackup.db.postgresql.PgDumpBinaryConnector.run_command",
@@ -289,6 +295,17 @@ def test_restore_dump_schema(self, mock_dump_cmd, mock_restore_cmd):
         self.assertIn(" -n public", mock_restore_cmd.call_args[0][0])
         self.assertIn(" -n foo", mock_restore_cmd.call_args[0][0])
 
+    def test_create_dump_password_excluded(self, mock_dump_cmd):
+        self.connector.settings["PASSWORD"] = "secret"
+        self.connector.create_dump()
+        self.assertNotIn("secret", mock_dump_cmd.call_args[0][0])
+
+    def test_restore_dump_password_excluded(self, mock_dump_cmd):
+        self.connector.settings["PASSWORD"] = "secret"
+        dump = self.connector.create_dump()
+        self.connector.restore_dump(dump)
+        self.assertNotIn("secret", mock_dump_cmd.call_args[0][0])
+
 
 @patch(
     "dbbackup.db.postgresql.PgDumpGisConnector.run_command",
@@ -365,6 +382,8 @@ def test_run_command_with_password(self, mock_popen):
         connector.settings["PASSWORD"] = "foo"
         connector.create_dump()
         self.assertEqual(mock_popen.call_args[0][0][0], "pg_dump")
+        self.assertNotIn("foo", " ".join(mock_popen.call_args[0][0]))
+        self.assertEqual("foo", mock_popen.call_args[1]["env"]["PGPASSWORD"])
 
     def test_run_command_with_password_and_other(self, mock_popen):
         connector = PgDumpConnector(env={"foo": "bar"})
@@ -374,3 +393,5 @@ def test_run_command_with_password_and_other(self, mock_popen):
         self.assertEqual(mock_popen.call_args[0][0][0], "pg_dump")
         self.assertIn("foo", mock_popen.call_args[1]["env"])
         self.assertEqual("bar", mock_popen.call_args[1]["env"]["foo"])
+        self.assertNotIn("foo", " ".join(mock_popen.call_args[0][0]))
+        self.assertEqual("foo", mock_popen.call_args[1]["env"]["PGPASSWORD"])

docs/changelog.rst

@@ -4,7 +4,7 @@ Changelog
 Unreleased
 ----------
 
-* Nothing (yet)!
+* Use environment variable for PostgreSQL password to prevent password leakage in logs/emails
 
 4.3.0 (2025-05-09)
 ----------