From a98475f7bcd923dd4cc33776c956dbca6b08fc51 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 03:18:08 +0200 Subject: [PATCH] Harden installer runtime secrets --- src/govoplan_core/core/module_installer.py | 48 ++++++++++++++++++---- tests/test_module_system.py | 27 +++++++++++- 2 files changed, 67 insertions(+), 8 deletions(-) diff --git a/src/govoplan_core/core/module_installer.py b/src/govoplan_core/core/module_installer.py index 590c661..12a44e8 100644 --- a/src/govoplan_core/core/module_installer.py +++ b/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 provided in metadata.get("provides_interfaces", ()): + for module_id, module_metadata in target_metadata.items(): + 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: diff --git a/tests/test_module_system.py b/tests/test_module_system.py index 4753884..8be33dd 100644 --- a/tests/test_module_system.py +++ b/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)