from __future__ import annotations from pathlib import Path import os import subprocess import sys import tempfile from types import SimpleNamespace import unittest from unittest.mock import patch META_ROOT = Path(__file__).resolve().parents[1] RELEASE_ROOT = META_ROOT / "tools" / "release" if str(RELEASE_ROOT) not in sys.path: sys.path.insert(0, str(RELEASE_ROOT)) from govoplan_release.model import RepositorySpec # noqa: E402 from govoplan_release.git_state import sanitized_git_environment # noqa: E402 from govoplan_release.release_execution import ( # noqa: E402 ExecutorSpec, ReleaseExecutionAmbiguous, ReleaseExecutionBlocked, build_selected_wheels, execute_repository_step, executor_spec, reconciled_repository_receipt, repository_state_receipt, require_trusted_release_runtime, verify_frozen_repository_tag_receipt, verify_repository_preflight_binding, _clone_catalog_sources, _checkout_frozen_source, _flatpak_proxy_environment, _run_isolated_wheel_builder, _trusted_flatpak_builder_proxy, _trusted_host_bubblewrap, _verify_exact_publication_delta, isolated_builder_launcher, ) from govoplan_release.selective_catalog import ( # noqa: E402 enforce_frozen_selected_source_provenance, enforce_selected_source_provenance, ) class ReleaseExecutionTests(unittest.TestCase): def test_flatpak_proxy_environment_drops_unrelated_values(self) -> None: with patch.dict( os.environ, { "DBUS_SESSION_BUS_ADDRESS": "unix:path=/attacker/bus", "XDG_RUNTIME_DIR": "/run/user/1000", "PYTHONPATH": "/attacker", "GIT_DIR": "/attacker/repository", }, clear=True, ): environment = _flatpak_proxy_environment() self.assertEqual("/usr/bin:/bin", environment["PATH"]) self.assertEqual( "unix:path=/run/flatpak/bus", environment["DBUS_SESSION_BUS_ADDRESS"], ) self.assertNotIn("XDG_RUNTIME_DIR", environment) self.assertNotIn("PYTHONPATH", environment) self.assertNotIn("GIT_DIR", environment) def test_host_bubblewrap_preflight_requires_root_owned_safe_binary(self) -> None: with patch( "govoplan_release.release_execution.subprocess.run", return_value=SimpleNamespace( returncode=0, stdout="81ed|0\n", ), ): self.assertTrue(_trusted_host_bubblewrap()) with patch( "govoplan_release.release_execution.subprocess.run", return_value=SimpleNamespace( returncode=0, stdout="81ff|1000\n", ), ): self.assertFalse(_trusted_host_bubblewrap()) def test_builder_launcher_uses_only_preflighted_flatpak_proxy(self) -> None: with patch( "govoplan_release.release_execution._trusted_flatpak_builder_proxy", return_value=True, ): self.assertEqual( ("/usr/bin/flatpak-spawn", "--host", "/usr/bin/bwrap"), isolated_builder_launcher(), ) def test_flatpak_proxy_requires_recognized_container(self) -> None: with patch.dict(os.environ, {}, clear=True): self.assertFalse(_trusted_flatpak_builder_proxy()) def test_flatpak_builder_launch_uses_sanitized_environment(self) -> None: process = SimpleNamespace(wait=lambda timeout: 0) command = ( "/usr/bin/flatpak-spawn", "--host", "/usr/bin/bwrap", ) with ( patch.dict( os.environ, { "DBUS_SESSION_BUS_ADDRESS": "unix:path=/attacker/bus", "PYTHONPATH": "/attacker", }, clear=True, ), patch( "govoplan_release.release_execution.subprocess.Popen", return_value=process, ) as popen, ): self.assertEqual(0, _run_isolated_wheel_builder(command)) environment = popen.call_args.kwargs["env"] self.assertEqual( "unix:path=/run/flatpak/bus", environment["DBUS_SESSION_BUS_ADDRESS"], ) self.assertNotIn("PYTHONPATH", environment) def test_git_environment_ignores_inherited_redirection_and_commands(self) -> None: with patch.dict( "os.environ", { "GIT_DIR": "/attacker/repository", "GIT_CONFIG_GLOBAL": "/attacker/config", "GIT_SSH_COMMAND": "/attacker/ssh", "PATH": "/attacker/bin", "SSH_AUTH_SOCK": "/operator/agent.sock", }, clear=False, ): environment = sanitized_git_environment() self.assertNotIn("GIT_DIR", environment) self.assertEqual("/dev/null", environment["GIT_CONFIG_GLOBAL"]) self.assertEqual("/usr/bin:/bin", environment["PATH"]) self.assertEqual( "/usr/bin/ssh -o BatchMode=yes -o ConnectTimeout=8", environment["GIT_SSH_COMMAND"], ) self.assertEqual("/operator/agent.sock", environment["SSH_AUTH_SOCK"]) def test_publication_reconciliation_rejects_unrelated_commit_delta(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: website = Path(temp_dir) / "website" _git(Path(temp_dir), "init", "-b", "main", str(website)) _git(website, "config", "user.name", "Release Test") _git(website, "config", "user.email", "release@example.test") catalog = website / "public/catalogs/v1/channels/stable.json" keyring = website / "public/catalogs/v1/keyring.json" modules = website / "public/catalogs/v1/modules" catalog.parent.mkdir(parents=True) modules.mkdir(parents=True) catalog.write_text("{}\n", encoding="utf-8") keyring.write_text("{}\n", encoding="utf-8") (website / "README.md").write_text("before\n", encoding="utf-8") _git(website, "add", ".") _git(website, "commit", "-m", "Base") parent = _git_output(website, "rev-parse", "HEAD") expected = { "public/catalogs/v1/channels/stable.json": b'{"new": true}\n', "public/catalogs/v1/keyring.json": b"{}\n", } catalog.write_bytes(expected["public/catalogs/v1/channels/stable.json"]) (website / "README.md").write_text("unrelated\n", encoding="utf-8") _git(website, "add", ".") _git(website, "commit", "-m", "Catalog plus unrelated change") commit = _git_output(website, "rev-parse", "HEAD") with self.assertRaisesRegex( ReleaseExecutionBlocked, "outside the exact derived delta", ): _verify_exact_publication_delta( web_root=website, frozen_parent=parent, commit=commit, expected_blobs=expected, module_root=modules, ) def test_runtime_trust_rejects_replaceable_workspace_boundary(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) meta = root / "govoplan" workspace = root / "workspace" _git(root, "init", "-b", "main", str(meta)) (meta / "tools" / "release").mkdir(parents=True) (meta / "tools" / "checks").mkdir(parents=True) (meta / "tools" / "release" / "tool.py").write_text( "# trusted\n", encoding="utf-8" ) (meta / "tools" / "checks" / "check.py").write_text( "# trusted\n", encoding="utf-8" ) registry = meta / "repositories.json" registry.write_text('{"repositories": []}\n', encoding="utf-8") workspace.mkdir(mode=0o700) with ( patch( "govoplan_release.release_execution.META_ROOT", meta, ), patch( "govoplan_release.release_execution.REPOSITORIES_FILE", registry, ), patch( "govoplan_release.release_execution.publication_runtime_trust_issues", return_value=(), ), ): require_trusted_release_runtime(workspace_root=workspace) workspace.chmod(0o777) with self.assertRaisesRegex( ReleaseExecutionBlocked, "operator-owned" ): require_trusted_release_runtime(workspace_root=workspace) def test_durable_binding_rejects_group_writable_git_configuration(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) repository = workspace / "govoplan-files" remote = workspace / "files.git" _git(workspace, "init", "--bare", str(remote)) _git(workspace, "init", "-b", "main", str(repository)) _git(repository, "config", "user.name", "Release Test") _git(repository, "config", "user.email", "release@example.test") (repository / "README.md").write_text("release\n", encoding="utf-8") _git(repository, "add", "README.md") _git(repository, "commit", "-m", "Initial") _git(repository, "remote", "add", "origin", str(remote)) (repository / ".git" / "config").chmod(0o664) with patch( "govoplan_release.release_execution.load_repository_specs", return_value=(_repository_spec(remote),), ): with self.assertRaisesRegex( ReleaseExecutionBlocked, "group/world writable" ): repository_state_receipt( repo="govoplan-files", target_tag="v1.2.3", workspace_root=workspace, ) def test_durable_binding_rejects_nested_writable_git_authority(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) repository = workspace / "govoplan-files" remote = workspace / "files.git" _git(workspace, "init", "--bare", str(remote)) _git(workspace, "init", "-b", "main", str(repository)) _git(repository, "config", "user.name", "Release Test") _git(repository, "config", "user.email", "release@example.test") (repository / "README.md").write_text("release\n", encoding="utf-8") _git(repository, "add", "README.md") _git(repository, "commit", "-m", "Initial") _git(repository, "remote", "add", "origin", str(remote)) (repository / ".git" / "refs" / "heads").chmod(0o777) with patch( "govoplan_release.release_execution.load_repository_specs", return_value=(_repository_spec(remote),), ): with self.assertRaisesRegex( ReleaseExecutionBlocked, "Git metadata directory", ): repository_state_receipt( repo="govoplan-files", target_tag="v1.2.3", workspace_root=workspace, ) def test_creation_binding_rejects_head_and_push_remote_drift(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) repository = workspace / "govoplan-files" remote = workspace / "files.git" _git(workspace, "init", "--bare", str(remote)) _git(workspace, "init", "-b", "main", str(repository)) _git(repository, "config", "user.name", "Release Test") _git(repository, "config", "user.email", "release@example.test") (repository / "README.md").write_text("one\n", encoding="utf-8") _git(repository, "add", "README.md") _git(repository, "commit", "-m", "Initial") _git(repository, "remote", "add", "origin", str(remote)) _git(repository, "push", "-u", "origin", "main") spec = _repository_spec(remote) with patch( "govoplan_release.release_execution.load_repository_specs", return_value=(spec,), ): frozen = repository_state_receipt( repo="govoplan-files", target_tag="v1.2.3", workspace_root=workspace, ) step = { "repo": "govoplan-files", "source_binding": frozen, } (repository / "README.md").write_text("two\n", encoding="utf-8") _git(repository, "add", "README.md") _git(repository, "commit", "-m", "Drift") with self.assertRaisesRegex( ReleaseExecutionBlocked, "state drifted" ): verify_repository_preflight_binding( plan_step=step, version="1.2.3", workspace_root=workspace, ) _git( repository, "remote", "set-url", "--add", "--push", "origin", str(workspace / "other.git"), ) with self.assertRaisesRegex( ReleaseExecutionBlocked, "registered release remote" ): repository_state_receipt( repo="govoplan-files", target_tag="v1.2.3", workspace_root=workspace, ) def test_reconciliation_proves_annotated_local_and_remote_tag(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) repository = workspace / "govoplan-files" remote = workspace / "files.git" _git(workspace, "init", "--bare", str(remote)) _git(workspace, "init", "-b", "main", str(repository)) _git(repository, "config", "user.name", "Release Test") _git(repository, "config", "user.email", "release@example.test") (repository / "README.md").write_text("release\n", encoding="utf-8") _git(repository, "add", "README.md") _git(repository, "commit", "-m", "Initial") _git(repository, "remote", "add", "origin", str(remote)) _git(repository, "push", "-u", "origin", "main") spec = _repository_spec(remote) plan_step = { "repo": "govoplan-files", "source_binding": {}, } tag_spec = ExecutorSpec("tag", "TAG", "tag_created", True) push_spec = ExecutorSpec("push", "PUBLISH", "tag_published", True) with patch( "govoplan_release.release_execution.load_repository_specs", return_value=(spec,), ): before = repository_state_receipt( repo="govoplan-files", target_tag="v1.2.3", workspace_root=workspace, ) _git(repository, "tag", "-a", "v1.2.3", "-m", "Release") tagged = reconciled_repository_receipt( spec=tag_spec, plan_step=plan_step, version="1.2.3", workspace_root=workspace, expected_receipt=before, ) self.assertIsNotNone(tagged["tag_object"]) with self.assertRaisesRegex( ReleaseExecutionBlocked, "Remote annotated tag" ): reconciled_repository_receipt( spec=push_spec, plan_step=plan_step, version="1.2.3", workspace_root=workspace, expected_receipt=tagged, ) _git( repository, "push", "--atomic", "origin", "HEAD:refs/heads/main", "refs/tags/v1.2.3", ) published = reconciled_repository_receipt( spec=push_spec, plan_step=plan_step, version="1.2.3", workspace_root=workspace, expected_receipt=tagged, ) self.assertEqual(tagged, published) (repository / "README.md").write_text( "uncommitted drift\n", encoding="utf-8" ) self.assertEqual( tagged, verify_frozen_repository_tag_receipt( receipt=tagged, workspace_root=workspace, ), ) (repository / "README.md").write_text( "remote branch drift\n", encoding="utf-8" ) _git(repository, "add", "README.md") _git(repository, "commit", "-m", "Remote branch drift") drift_commit = _git_output(repository, "rev-parse", "HEAD") _git( repository, "push", "origin", f"{drift_commit}:refs/heads/drift-source", ) _git(remote, "update-ref", "refs/heads/main", drift_commit) _git(repository, "reset", "--hard", str(tagged["head"])) with self.assertRaisesRegex( ReleaseExecutionBlocked, "Remote branch", ): reconciled_repository_receipt( spec=push_spec, plan_step=plan_step, version="1.2.3", workspace_root=workspace, expected_receipt=tagged, ) def test_post_tag_receipt_failure_is_ambiguous(self) -> None: spec = ExecutorSpec("tag", "TAG", "tag_created", True) plan_step = { "id": "govoplan-files:tag", "repo": "govoplan-files", "source_binding": {"kind": "repository_state"}, } with ( patch( "govoplan_release.release_execution.require_trusted_release_runtime" ), patch( "govoplan_release.release_execution.verify_repository_step_precondition" ), patch( "govoplan_release.release_execution.tag_repositories", return_value={ "status": "tagged", "repositories": [{"status": "tagged"}], }, ), patch( "govoplan_release.release_execution.repository_state_receipt", side_effect=ReleaseExecutionBlocked("post-effect probe failed"), ), ): with self.assertRaisesRegex( ReleaseExecutionAmbiguous, "post-effect repository receipt" ): execute_repository_step( spec=spec, plan_step=plan_step, repo_versions={"govoplan-files": "1.2.3"}, workspace_root=Path("/workspace"), remote="origin", expected_receipt={"kind": "repository_state"}, ) def test_commit_step_has_no_durable_executor(self) -> None: self.assertIsNone( executor_spec( { "id": "govoplan-files:commit", "status": "planned", "mutating": True, "repo": "govoplan-files", "source_binding": { "kind": "repository_state", }, } ) ) def test_frozen_checkout_uses_receipt_commit_not_live_worktree(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) repository = workspace / "govoplan-files" repository.mkdir() _git(repository, "init", "-b", "main") _git(repository, "config", "user.name", "Release Test") _git(repository, "config", "user.email", "release@example.test") (repository / "version.txt").write_text("frozen\n", encoding="utf-8") _git(repository, "add", "version.txt") _git(repository, "commit", "-m", "Frozen") frozen_commit = _git_output(repository, "rev-parse", "HEAD") _git(repository, "tag", "-a", "v1.2.3", "-m", "Release") tag_object = _git_output(repository, "rev-parse", "refs/tags/v1.2.3") (repository / "version.txt").write_text("live drift\n", encoding="utf-8") source_root = workspace / "isolated" source_root.mkdir() checkout = _checkout_frozen_source( repo="govoplan-files", source_remote=repository, commit=frozen_commit, tag="v1.2.3", tag_object=tag_object, source_root=source_root, ) self.assertEqual( "frozen\n", (checkout / "version.txt").read_text(encoding="utf-8") ) def test_frozen_provenance_accepts_only_exact_clean_detached_tag(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) repository = workspace / "govoplan-files" remote = workspace / "files.git" _git(workspace, "init", "--bare", str(remote)) _git(workspace, "init", "-b", "main", str(repository)) _git(repository, "config", "user.name", "Release Test") _git(repository, "config", "user.email", "release@example.test") (repository / "pyproject.toml").write_text( '[project]\nname = "govoplan-files"\nversion = "1.2.3"\n', encoding="utf-8", ) _git(repository, "add", "pyproject.toml") _git(repository, "commit", "-m", "Release") commit = _git_output(repository, "rev-parse", "HEAD") _git(repository, "tag", "-a", "v1.2.3", "-m", "Release") tag_object = _git_output( repository, "rev-parse", "refs/tags/v1.2.3" ) _git(repository, "remote", "add", "origin", str(remote)) _git(repository, "push", "origin", "main", "refs/tags/v1.2.3") _git(repository, "checkout", "--detach", commit) spec = _repository_spec(remote) with ( patch( "govoplan_release.source_provenance.load_repository_specs", return_value=(spec,), ), patch( "govoplan_release.selective_catalog.load_repository_specs", return_value=(spec,), ), ): with self.assertRaisesRegex(ValueError, "named branch"): enforce_selected_source_provenance( repo_versions={"govoplan-files": "1.2.3"}, workspace=workspace, ) observed = enforce_frozen_selected_source_provenance( repo_versions={"govoplan-files": "1.2.3"}, expected_provenance={ "govoplan-files": { "commit_sha": commit, "tag_object_sha": tag_object, } }, workspace=workspace, ) self.assertEqual(commit, observed["govoplan-files"]["commit_sha"]) self.assertEqual( tag_object, observed["govoplan-files"]["tag_object_sha"] ) def test_wheel_staging_skips_selected_tag_only_repository(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) python_repo = workspace / "govoplan-files" tag_only_repo = workspace / "govoplan-meta-notes" python_repo.mkdir() tag_only_repo.mkdir() (python_repo / "pyproject.toml").write_text( '[project]\nname = "govoplan-files"\nversion = "1.2.3"\n', encoding="utf-8", ) _git(workspace, "init", "-b", "main", str(python_repo)) _git(python_repo, "config", "user.name", "Release Test") _git(python_repo, "config", "user.email", "release@example.test") _git(python_repo, "add", "pyproject.toml") _git(python_repo, "commit", "-m", "Source") specs = ( RepositorySpec( name="govoplan-files", category="module", subtype="infrastructure", remote="git@example.test/files.git", path="govoplan-files", ), RepositorySpec( name="govoplan-meta-notes", category="system", subtype="metadata", remote="git@example.test/notes.git", path="govoplan-meta-notes", ), ) candidate = workspace / "candidate" def build(command: tuple[str, ...]) -> int: wheel_dir = Path(command[command.index("--bind") + 1]) second_bind = command.index("--bind", command.index("--bind") + 1) source_dir = Path(command[second_bind + 1]) (source_dir / "build-mutated.txt").write_text( "isolated copy", encoding="utf-8", ) ( wheel_dir / "govoplan_files-1.2.3-py3-none-any.whl" ).write_bytes(b"wheel") return 0 with ( patch( "govoplan_release.release_execution.load_repository_specs", return_value=specs, ), patch( "govoplan_release.release_execution.isolated_builder_launcher", return_value=("/usr/bin/bwrap",), ), patch( "govoplan_release.release_execution._run_isolated_wheel_builder", side_effect=build, ) as runner, patch( "govoplan_release.release_execution.inspect_python_wheel", return_value=SimpleNamespace( package_name="govoplan-files", package_version="1.2.3", ), ), ): artifacts = build_selected_wheels( repo_versions={ "govoplan-files": "1.2.3", "govoplan-meta-notes": "1.0.0", }, workspace_root=workspace, candidate_output=candidate, source_receipts={ "govoplan-files": { "head": "a" * 40, "target_tag": "v1.2.3", "tag_object": "c" * 40, }, "govoplan-meta-notes": { "head": "b" * 40, "target_tag": "v1.0.0", "tag_object": "d" * 40, }, }, frozen_sources={ "govoplan-files": python_repo, "govoplan-meta-notes": tag_only_repo, }, ) self.assertFalse((python_repo / "build-mutated.txt").exists()) self.assertEqual({"govoplan-files"}, set(artifacts)) self.assertEqual(1, runner.call_count) command = runner.call_args.args[0] self.assertNotIn(str(python_repo), command) self.assertTrue( any( value.startswith(str(candidate / "artifacts")) and value.endswith("/source") for value in command ) ) self.assertIn("--unshare-all", command) self.assertIn("PIP_NO_INDEX", command) self.assertNotIn(str(Path.home()), command) def test_catalog_synthesis_clones_selected_and_unchanged_base_sources( self, ) -> None: with tempfile.TemporaryDirectory() as temp_dir: workspace = Path(temp_dir) files_repo, files_remote = _tagged_repository( workspace, repo="govoplan-files", bare="files.git", version="1.2.3", ) core_repo, core_remote = _tagged_repository( workspace, repo="govoplan-core", bare="core.git", version="1.0.0", ) specs = ( _repository_spec(files_remote), RepositorySpec( name="govoplan-core", category="core", subtype="core", remote=str(core_remote), path="govoplan-core", ), ) destination = workspace / "synthesis" destination.mkdir() with patch( "govoplan_release.release_execution.load_repository_specs", return_value=specs, ): receipt = repository_state_receipt( repo="govoplan-files", target_tag="v1.2.3", workspace_root=workspace, ) cloned = _clone_catalog_sources( source_versions={ "govoplan-core": "1.0.0", "govoplan-files": "1.2.3", }, repo_versions={"govoplan-files": "1.2.3"}, source_receipts={"govoplan-files": receipt}, workspace_root=workspace, destination=destination, ) self.assertEqual( _git_output(files_repo, "rev-parse", "refs/tags/v1.2.3^{commit}"), _git_output(cloned["govoplan-files"], "rev-parse", "HEAD"), ) self.assertEqual( _git_output(core_repo, "rev-parse", "refs/tags/v1.0.0^{commit}"), _git_output(cloned["govoplan-core"], "rev-parse", "HEAD"), ) self.assertEqual( "HEAD", _git_output(cloned["govoplan-core"], "rev-parse", "--abbrev-ref", "HEAD"), ) def _repository_spec(remote: Path) -> RepositorySpec: return RepositorySpec( name="govoplan-files", category="module", subtype="infrastructure", remote=str(remote), path="govoplan-files", ) def _tagged_repository( workspace: Path, *, repo: str, bare: str, version: str ) -> tuple[Path, Path]: repository = workspace / repo remote = workspace / bare _git(workspace, "init", "--bare", str(remote)) _git(workspace, "init", "-b", "main", str(repository)) _git(repository, "config", "user.name", "Release Test") _git(repository, "config", "user.email", "release@example.test") (repository / "pyproject.toml").write_text( f'[project]\nname = "{repo}"\nversion = "{version}"\n', encoding="utf-8", ) _git(repository, "add", "pyproject.toml") _git(repository, "commit", "-m", "Release") _git(repository, "tag", "-a", f"v{version}", "-m", "Release") _git(repository, "remote", "add", "origin", str(remote)) _git(repository, "push", "-u", "origin", "main", f"refs/tags/v{version}") return repository, remote def _git(cwd: Path, *arguments: str) -> None: subprocess.run( # noqa: S603 ("git", *arguments), cwd=cwd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) def _git_output(cwd: Path, *arguments: str) -> str: return subprocess.run( # noqa: S603 ("git", *arguments), cwd=cwd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ).stdout.strip() if __name__ == "__main__": unittest.main()