Harden installer runtime secrets

src/govoplan_core/core/module_installer.py

@@ -15,7 +15,8 @@ import re
 import shlex
 import shutil
 import sqlite3
-import subprocess
+import stat
+import subprocess  # nosec B404 - installer commands are structured and policy-validated before execution.
 import sys
 import tomllib
 from typing import Any, Literal
@@ -579,7 +580,7 @@ def _prepare_module_install_run(
 ) -> _ModuleInstallRunState:
     run_id = _run_id()
     run_dir = effective_runtime_dir / "runs" / run_id
-    run_dir.mkdir(parents=True, exist_ok=False)
+    _create_private_installer_run_dir(run_dir)
     commands = structured_install_commands(
         plan,
         webui_root=webui_root,
@@ -2445,8 +2446,8 @@ def _migration_provider_modules(
     target_metadata: Mapping[str, Mapping[str, object]],
 ) -> dict[str, tuple[tuple[str, str], ...]]:
     providers: dict[str, list[tuple[str, str]]] = defaultdict(list)
-    for module_id, metadata in target_metadata.items():
+    for module_id, module_metadata in target_metadata.items():
-        for provided in metadata.get("provides_interfaces", ()):
+        for provided in module_metadata.get("provides_interfaces", ()):
             if not isinstance(provided, Mapping):
                 continue
             name = provided.get("name") if isinstance(provided.get("name"), str) else None
@@ -3828,14 +3829,47 @@ def _restore_external_database_snapshot(
 def _write_installer_secret(path: Path, value: str | None) -> str | None:
     if not value:
         return None
-    path.write_text(value, encoding="utf-8")
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+    flags |= getattr(os, "O_CLOEXEC", 0)
+    flags |= getattr(os, "O_NOFOLLOW", 0)
     try:
-        path.chmod(0o600)
+        descriptor = os.open(path, flags, 0o600)
     except OSError as exc:
-        logger.debug("Could not restrict installer secret file permissions for %s: %s", path, exc, exc_info=True)
+        raise ModuleInstallerError(f"Could not create private installer secret file: {path.name}") from exc
+    try:
+        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
+            handle.write(value)
+    except OSError as exc:
+        path.unlink(missing_ok=True)
+        raise ModuleInstallerError(f"Could not write private installer secret file: {path.name}") from exc
+
+    try:
+        mode = stat.S_IMODE(path.stat().st_mode)
+    except OSError as exc:
+        path.unlink(missing_ok=True)
+        raise ModuleInstallerError(f"Could not verify private installer secret file: {path.name}") from exc
+    if mode != 0o600:
+        path.unlink(missing_ok=True)
+        raise ModuleInstallerError(f"Installer secret file permissions are not private: {path.name}")
     return path.name
 
 
+def _create_private_installer_run_dir(path: Path) -> None:
+    path.mkdir(parents=True, mode=0o700, exist_ok=False)
+    try:
+        # mkdir's requested mode is still reduced by the process umask. Set the
+        # exact owner-only mode so the directory remains usable with a strict
+        # deployment umask while never retaining group/other access.
+        path.chmod(0o700)
+        mode = stat.S_IMODE(path.stat().st_mode)
+    except OSError as exc:
+        path.rmdir()
+        raise ModuleInstallerError("Could not secure the installer run directory") from exc
+    if mode != 0o700:
+        path.rmdir()
+        raise ModuleInstallerError("Installer run directory permissions are not exactly owner-only")
+
+
 def _database_url_from_external_snapshot(run_dir: Path, raw: Mapping[str, object]) -> str | None:
     secret_name = raw.get("database_url_secret")
     if isinstance(secret_name, str) and secret_name:

tests/test_module_system.py

@@ -8,6 +8,7 @@ import os
 import re
 import shlex
 import shutil
+import stat
 import subprocess
 import sys
 import tempfile
@@ -2562,9 +2563,33 @@ finally:
         self.assertEqual("external", database_backup["metadata"]["snapshot"])
         self.assertEqual("postgresql://govoplan:***@db.example.invalid/govoplan", database_backup["metadata"]["pgtools_url"])
         self.assertEqual("postgresql+psycopg://govoplan:***@db.example.invalid/govoplan", database_backup["stdout"].strip())
-        self.assertEqual(database_url, (result.record_path.parent / database_backup["database_url_secret"]).read_text(encoding="utf-8"))
+        secret_path = result.record_path.parent / database_backup["database_url_secret"]
+        self.assertEqual(database_url, secret_path.read_text(encoding="utf-8"))
+        self.assertEqual(0o700, stat.S_IMODE(result.record_path.parent.stat().st_mode))
+        self.assertEqual(0o600, stat.S_IMODE(secret_path.stat().st_mode))
         self.assertEqual("backup", (result.record_path.parent / "database.external.backup").read_text(encoding="utf-8"))
 
+    def test_installer_secret_creation_refuses_an_existing_path(self) -> None:
+        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-secret-existing-", dir=_TEST_ROOT))
+        secret_path = root / "database-url.secret"
+        secret_path.write_text("existing-value", encoding="utf-8")
+
+        with self.assertRaisesRegex(module_installer_module.ModuleInstallerError, "Could not create private installer secret file"):
+            module_installer_module._write_installer_secret(secret_path, "replacement-value")
+
+        self.assertEqual("existing-value", secret_path.read_text(encoding="utf-8"))
+
+    def test_installer_run_directory_remains_usable_with_a_restrictive_umask(self) -> None:
+        root = Path(tempfile.mkdtemp(prefix="govoplan-installer-run-umask-", dir=_TEST_ROOT))
+        run_dir = root / "run"
+        previous_umask = os.umask(0o777)
+        try:
+            module_installer_module._create_private_installer_run_dir(run_dir)
+        finally:
+            os.umask(previous_umask)
+
+        self.assertEqual(0o700, stat.S_IMODE(run_dir.stat().st_mode))
+
     def test_module_installer_rejects_unsafe_external_database_hook_command(self) -> None:
         root = Path(tempfile.mkdtemp(prefix="govoplan-installer-unsafe-hook-", dir=_TEST_ROOT))
         settings = _settings(root)