refactor(backup)!: run argv lists, never a shell

The backup built command strings and handed them to shell=True, so four interpolated values per dump - user, password, container, database - were each a way out of the command. validate_database covered one of them since the previous commit; now there is nothing to cover: every command is an argv list, and a value can only ever be an argument.

execute_to_file absorbs the atomic dump write. The shell redirect into <file>.tmp and the separate mv process become a Python file handle and os.replace, and a failing dump deletes its partial file instead of leaving it. PGPASSWORD moves out of the command string into the child's environment, where a process listing does not show it.

docker exec is built in one place, docker_exec_argv; db.py's three hand-built copies and the probe use it. The dead docker_volume_exists goes - never called, and the restore side owns the living twin. The rsync quoting in --link-dest falls away: inside an argv it would have become part of the path.

The snapshot module's injected runner changes type with it, which the three e2e drivers implement - the first conversion missed them, btrfs ran with no arguments, and the e2e caught it. Marked breaking for that contract: any external runner injected into volume_snapshot must now accept a list.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tests/unit/backup/test_db_mariadb_dump.py

@@ -13,16 +13,16 @@ def _df(rows):
     )
 
 
-def _capture_commands(*, db_type, rows, container, dump_tool="mariadb-dump"):
+def _capture_dumps(*, db_type, rows, container, dump_tool="mariadb-dump"):
+    """Every (argv, env) the dump path would have run."""
     captured = []
 
-    def _capture(cmd):
-        captured.append(cmd)
-        return []
+    def _capture(command, out_file, *, env=None):
+        captured.append((list(command), env))
 
     with (
         tempfile.TemporaryDirectory() as td,
-        patch.object(db_mod, "execute_shell_command", side_effect=_capture),
+        patch.object(db_mod, "execute_to_file", side_effect=_capture),
     ):
         db_mod.backup_database(
             container=container,
@@ -41,45 +41,64 @@ class TestMariaDBDumpUsesTCP(unittest.TestCase):
     # the connection is auth-matched against '%' instead of socket->localhost.
 
     def test_mariadb_dump_forces_tcp_loopback(self):
-        captured = _capture_commands(
+        captured = _capture_dumps(
             db_type="mariadb",
             rows=[("mariadb", "appdb", "appuser", "s3cret")],
             container="mariadb",
         )
-        dump_cmds = [c for c in captured if "mariadb-dump" in c]
-        self.assertEqual(
-            len(dump_cmds), 1, f"expected one dump command, got: {captured}"
-        )
+        self.assertEqual(len(captured), 1, f"expected one dump, got: {captured}")
 
-        cmd = dump_cmds[0]
-        self.assertIn("-h 127.0.0.1", cmd)
-        self.assertIn("--protocol=tcp", cmd)
-        self.assertIn("-u appuser", cmd)
-        self.assertIn("-ps3cret", cmd)
-        self.assertIn(" appdb", cmd)
+        argv, env = captured[0]
+        self.assertEqual(argv[:3], ["docker", "exec", "mariadb"])
+        self.assertIn("--protocol=tcp", argv)
+        self.assertEqual(argv[argv.index("-h") + 1], "127.0.0.1")
+        self.assertEqual(argv[argv.index("-u") + 1], "appuser")
+        self.assertIn("-ps3cret", argv)
+        self.assertEqual(argv[-1], "appdb")
+        self.assertIsNone(env)
 
     def test_the_probed_client_is_the_one_invoked(self):
-        captured = _capture_commands(
+        captured = _capture_dumps(
             db_type="mariadb",
             rows=[("mariadb", "appdb", "appuser", "s3cret")],
             container="mariadb",
             dump_tool="mysqldump",
         )
-        dump_cmds = [c for c in captured if "mysqldump" in c]
-        self.assertEqual(
-            len(dump_cmds), 1, f"expected one dump command, got: {captured}"
-        )
-        self.assertNotIn("mariadb-dump", dump_cmds[0])
+        argv, _env = captured[0]
+        self.assertIn("mysqldump", argv)
+        self.assertNotIn("mariadb-dump", argv)
 
     def test_postgres_dump_unaffected(self):
-        captured = _capture_commands(
+        captured = _capture_dumps(
             db_type="postgres",
             rows=[("pg", "appdb", "appuser", "s3cret")],
             container="pg",
         )
-        dump_cmds = [c for c in captured if "pg_dump" in c and "pg_dumpall" not in c]
-        self.assertEqual(len(dump_cmds), 1)
-        self.assertNotIn("--protocol=tcp", dump_cmds[0])
+        argv, _env = captured[0]
+        self.assertIn("pg_dump", argv)
+        self.assertNotIn("--protocol=tcp", argv)
+
+    def test_the_password_travels_in_the_environment_not_the_argv(self):
+        """A process listing shows argv; PGPASSWORD must not be in it."""
+        captured = _capture_dumps(
+            db_type="postgres",
+            rows=[("pg", "appdb", "appuser", "s3cret")],
+            container="pg",
+        )
+        argv, env = captured[0]
+        self.assertEqual(env, {"PGPASSWORD": "s3cret"})
+        self.assertNotIn("s3cret", argv)
+
+
+class TestNoShellReachesTheDump(unittest.TestCase):
+    def test_a_hostile_database_name_never_reaches_a_command(self):
+        """validate_database refuses it, so no argv is built at all."""
+        with self.assertRaises(ValueError):
+            _capture_dumps(
+                db_type="postgres",
+                rows=[("pg", "app;rm -rf /", "appuser", "s3cret")],
+                container="pg",
+            )
 
 
 if __name__ == "__main__":