from __future__ import annotations import hashlib import json import os from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch META_ROOT = Path(__file__).resolve().parents[1] RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" if str(RELEASE_TOOLS_ROOT) not in sys.path: sys.path.insert(0, str(RELEASE_TOOLS_ROOT)) from govoplan_release.publisher import ( # noqa: E402 FrozenPublicationRemote, bind_publication_remote, commit_publication_tree, publication_mutation_trust_issues, publish_catalog_candidate, remote_publication_identity, run_checked, verify_committed_publication, verify_remote_branch_head, verify_remote_publication, _git_bytes, _sanitized_git_environment, _seal_git_metadata_file, _trusted_npm_command, ) class ReleaseCatalogPublicationTests(unittest.TestCase): def test_publication_git_writes_ignore_permissive_operator_umask(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: repository = Path(temp_dir) / "website" repository.mkdir() self._git(repository, "init", "--quiet") payload_number = 0 while True: payload = f"private publication object {payload_number}\n".encode() header = f"blob {len(payload)}\0".encode() expected_object = hashlib.sha1( header + payload, usedforsecurity=False, ).hexdigest() object_parent = repository / ".git" / "objects" / expected_object[:2] if not object_parent.exists(): break payload_number += 1 previous_umask = os.umask(0o002) try: object_id = ( _git_bytes( repository, "hash-object", "-w", "--stdin", input_bytes=payload, ) .decode("ascii") .strip() ) finally: os.umask(previous_umask) object_path = object_parent / expected_object[2:] self.assertEqual(expected_object, object_id) self.assertEqual(os.geteuid(), object_parent.stat().st_uid) self.assertEqual(0o700, object_parent.stat().st_mode & 0o777) self.assertEqual(os.geteuid(), object_path.stat().st_uid) self.assertEqual(0o400, object_path.stat().st_mode & 0o777) def test_publication_git_metadata_is_sealed_after_git_writes(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: metadata = Path(temp_dir) / "index" metadata.write_bytes(b"index") metadata.chmod(0o664) _seal_git_metadata_file(metadata, label="test index") self.assertEqual(0o600, metadata.stat().st_mode & 0o777) def test_publication_git_identity_is_fixed_and_non_personal(self) -> None: with patch.dict( os.environ, { "GIT_AUTHOR_NAME": "Attacker", "GIT_AUTHOR_EMAIL": "attacker@example.test", "GIT_COMMITTER_NAME": "Attacker", "GIT_COMMITTER_EMAIL": "attacker@example.test", }, clear=False, ): environment = _sanitized_git_environment() self.assertEqual( "GovOPlaN Release Automation", environment["GIT_AUTHOR_NAME"], ) self.assertEqual( "release@govoplan.invalid", environment["GIT_AUTHOR_EMAIL"], ) self.assertEqual( environment["GIT_AUTHOR_NAME"], environment["GIT_COMMITTER_NAME"], ) self.assertEqual( environment["GIT_AUTHOR_EMAIL"], environment["GIT_COMMITTER_EMAIL"], ) def test_build_command_uses_its_pinned_node_directory(self) -> None: completed = subprocess.CompletedProcess( ["/trusted/node/bin/npm"], 0, stdout=b"", stderr=b"", ) with patch( "govoplan_release.publisher.subprocess.run", return_value=completed, ) as runner: run_checked( ["/trusted/node/bin/npm", "run", "build"], cwd=Path("/trusted/website"), ) environment = runner.call_args.kwargs["env"] self.assertEqual( "/trusted/node/bin:/usr/bin:/bin", environment["PATH"], ) def test_npm_command_requires_trusted_npm_and_sibling_node(self) -> None: with ( patch( "govoplan_release.publisher.shutil.which", return_value="/trusted/node/bin/npm", ), patch( "govoplan_release.publisher._trusted_runtime_executable_issue", side_effect=(None, None), ) as trust_check, ): self.assertEqual( "/trusted/node/bin/npm", _trusted_npm_command("npm"), ) self.assertEqual(Path("/trusted/node/bin/npm"), trust_check.call_args_list[0].args[0]) self.assertEqual(Path("/trusted/node/bin/node"), trust_check.call_args_list[1].args[0]) def test_npm_command_rejects_caller_relative_path(self) -> None: with ( patch("govoplan_release.publisher.shutil.which") as which, self.assertRaisesRegex(RuntimeError, "absolute path or the bare name"), ): _trusted_npm_command("./npm") which.assert_not_called() def test_committed_publication_must_match_validated_blobs_exactly(self) -> None: with tempfile.TemporaryDirectory() as tmp: web_root = Path(tmp) / "website" module_root = web_root / "public" / "catalogs" / "v1" / "modules" module_root.mkdir(parents=True) catalog = ( web_root / "public" / "catalogs" / "v1" / "channels" / "stable.json" ) keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json" catalog.parent.mkdir(parents=True) expected = { catalog.relative_to(web_root).as_posix(): b'{"catalog":true}\n', keyring.relative_to(web_root).as_posix(): b'{"keys":[]}\n', (module_root / "index.json") .relative_to(web_root) .as_posix(): b'{"modules":[]}\n', } for relative, encoded in expected.items(): target = web_root / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(encoded) self._git(web_root, "init", "--quiet") self._git(web_root, "config", "user.email", "test@example.test") self._git(web_root, "config", "user.name", "Test") self._git(web_root, "add", ".") self._git(web_root, "commit", "--quiet", "-m", "publication") commit_sha = self._git(web_root, "rev-parse", "HEAD").strip() verify_committed_publication( web_root=web_root, commit_sha=commit_sha, expected_blobs=expected, module_root=module_root, ) changed = dict(expected) changed[catalog.relative_to(web_root).as_posix()] = b'{"catalog":false}\n' with self.assertRaisesRegex(RuntimeError, "differs"): verify_committed_publication( web_root=web_root, commit_sha=commit_sha, expected_blobs=changed, module_root=module_root, ) catalog_path = catalog.relative_to(web_root).as_posix() self._git(web_root, "update-index", "--chmod=+x", catalog_path) self._git(web_root, "commit", "--quiet", "-m", "unsafe mode") executable_commit = self._git(web_root, "rev-parse", "HEAD").strip() with self.assertRaisesRegex(RuntimeError, "regular blob"): verify_committed_publication( web_root=web_root, commit_sha=executable_commit, expected_blobs=expected, module_root=module_root, ) def test_private_index_commit_has_one_parent_and_only_computed_delta(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) web_root = root / "website" module_root = web_root / "public" / "catalogs" / "v1" / "modules" channel = ( web_root / "public" / "catalogs" / "v1" / "channels" / "stable.json" ) keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json" old_module = module_root / "obsolete.json" unrelated = web_root / "unrelated.txt" for path, encoded in ( (channel, b'{"old":true}\n'), (keyring, b'{"keys":[]}\n'), (old_module, b'{"obsolete":true}\n'), (unrelated, b"keep\n"), ): path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(encoded) self._git(web_root, "init", "--quiet") self._git(web_root, "config", "user.email", "test@example.test") self._git(web_root, "config", "user.name", "Test") self._git(web_root, "add", ".") self._git(web_root, "commit", "--quiet", "-m", "base") frozen_head = self._git(web_root, "rev-parse", "HEAD").strip() branch = self._git(web_root, "branch", "--show-current").strip() malicious = web_root / "not-in-publication.txt" malicious.write_text("staged but excluded\n", encoding="utf-8") self._git(web_root, "add", malicious.name) marker = root / "reference-hook-ran" hook = web_root / ".git" / "hooks" / "reference-transaction" hook.write_text(f"#!/bin/sh\ntouch {marker}\n", encoding="utf-8") hook.chmod(0o755) expected = { channel.relative_to(web_root).as_posix(): b'{"new":true}\n', keyring.relative_to(web_root).as_posix(): b'{"keys":["release"]}\n', (module_root / "index.json") .relative_to(web_root) .as_posix(): b'{"modules":[]}\n', } redirected = root / "attacker-index" with patch.dict( os.environ, { "GIT_DIR": str(root / "attacker-git-dir"), "GIT_INDEX_FILE": str(redirected), "GIT_OBJECT_DIRECTORY": str(root / "attacker-objects"), "GIT_CONFIG_GLOBAL": str(root / "attacker-config"), }, ): commit_sha = commit_publication_tree( web_root=web_root, frozen_head=frozen_head, branch=branch, expected_blobs=expected, module_root=module_root, message="Exact publication", ) parents = self._git( web_root, "rev-list", "--parents", "-n", "1", commit_sha ).split() changed = set( filter( None, self._git( web_root, "diff-tree", "--no-commit-id", "--name-only", "-r", frozen_head, commit_sha, ).splitlines(), ) ) hook_ran = marker.exists() inherited_index_used = redirected.exists() commit_paths = self._git( web_root, "ls-tree", "-r", "--name-only", commit_sha ) self.assertEqual([commit_sha, frozen_head], parents) self.assertEqual( { "public/catalogs/v1/channels/stable.json", "public/catalogs/v1/keyring.json", "public/catalogs/v1/modules/index.json", "public/catalogs/v1/modules/obsolete.json", }, changed, ) self.assertFalse(hook_ran) self.assertFalse(inherited_index_used) self.assertNotIn("not-in-publication.txt", commit_paths) def test_remote_binding_and_annotated_publication_identity_are_exact(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) remote_root = root / "website.git" remote_root.mkdir() self._git(remote_root, "init", "--bare", "--quiet") web_root = root / "website" web_root.mkdir() self._git(web_root, "init", "--quiet") self._git(web_root, "config", "user.email", "test@example.test") self._git(web_root, "config", "user.name", "Test") self._git(web_root, "branch", "-M", "main") web_root.joinpath("base.txt").write_text("base\n", encoding="utf-8") self._git(web_root, "add", ".") self._git(web_root, "commit", "--quiet", "-m", "base") self._git(web_root, "remote", "add", "origin", str(remote_root)) self._git(web_root, "push", "--quiet", "-u", "origin", "main") base_commit = self._git(web_root, "rev-parse", "HEAD").strip() frozen = bind_publication_remote( web_root=web_root, remote="origin", registered_remote=str(remote_root), ) self.assertIsInstance(frozen, FrozenPublicationRemote) verify_remote_branch_head( web_root=web_root, remote_url=frozen.url, branch="main", expected_commit=base_commit, ) web_root.joinpath("catalog.json").write_text("{}\n", encoding="utf-8") self._git(web_root, "add", "catalog.json") self._git(web_root, "commit", "--quiet", "-m", "publication") commit_sha = self._git(web_root, "rev-parse", "HEAD").strip() tag_name = "catalog-test" self._git(web_root, "tag", "-a", tag_name, "-m", "publication") tag_object = self._git( web_root, "rev-parse", f"refs/tags/{tag_name}" ).strip() self._git( web_root, "push", "--quiet", "--atomic", "origin", f"{commit_sha}:refs/heads/main", f"{tag_object}:refs/tags/{tag_name}", ) identity = remote_publication_identity( web_root=web_root, remote_url=frozen.url, branch="main", tag_name=tag_name, ) verified = verify_remote_publication( web_root=web_root, remote_url=frozen.url, branch="main", tag_name=tag_name, expected_commit=commit_sha, expected_tag_object=tag_object, ) self._git( web_root, "remote", "set-url", "--add", "--push", "origin", str(root / "other.git"), ) with self.assertRaisesRegex(RuntimeError, "do not match"): bind_publication_remote( web_root=web_root, remote="origin", registered_remote=str(remote_root), ) self.assertEqual(identity, verified) self.assertEqual(commit_sha, identity["publication_commit_sha"]) self.assertEqual(tag_object, identity["publication_tag_object_sha"]) self.assertEqual(commit_sha, identity["publication_tag_commit_sha"]) def test_mutation_rejects_writable_website_and_git_configuration(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) web_root = root / "website" web_root.mkdir() self._git(web_root, "init", "--quiet") candidate = self._candidate(root, key="trusted-key") target_root = web_root / "public" / "catalogs" / "v1" target_catalog = target_root / "channels" / "stable.json" target_keyring = target_root / "keyring.json" target_keyring.parent.mkdir(parents=True) target_keyring.write_text("{}\n", encoding="utf-8") web_root.chmod(0o777) root_issues = publication_mutation_trust_issues( web_root=web_root, candidate_catalog=candidate / "channels" / "stable.json", candidate_keyring=candidate / "keyring.json", target_catalog=target_catalog, target_keyring=target_keyring, target_modules=target_root / "modules", ) web_root.chmod(0o755) config = web_root / ".git" / "config" config.chmod(0o666) config_issues = publication_mutation_trust_issues( web_root=web_root, candidate_catalog=candidate / "channels" / "stable.json", candidate_keyring=candidate / "keyring.json", target_catalog=target_catalog, target_keyring=target_keyring, target_modules=target_root / "modules", ) self.assertIn("website root is writable by another user", root_issues) self.assertIn("website Git config is writable by another user", config_issues) def test_push_returns_only_independently_verified_publication_receipt(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) candidate = self._candidate(root, key="trusted-key") catalog_path = candidate / "channels" / "stable.json" catalog = json.loads(catalog_path.read_text(encoding="utf-8")) catalog["sequence"] = 7 catalog["core_release"]["python_package"] = "govoplan-core" catalog["release"] = { "selected_units": [ { "repo": "govoplan-core", "version": "1.2.3", "tag": "v1.2.3", "commit_sha": "a" * 40, "tag_object_sha": "b" * 40, } ], "artifacts": [ { "artifact_kind": "python-wheel", "package_name": "govoplan-core", "package_version": "1.2.3", "archive_sha256": "c" * 64, "archive_size": 100, "installed_payload": { "algorithm": "govoplan-wheel-declared-payload-v1", "sha256": "d" * 64, "file_count": 1, }, "requires_installer_receipt": True, } ], } catalog["signatures"] = [{}] catalog_path.write_text(json.dumps(catalog), encoding="utf-8") remote_root = root / "website.git" remote_root.mkdir() self._git(remote_root, "init", "--bare", "--quiet") web_root = root / "website" target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json" target_keyring.parent.mkdir(parents=True) target_keyring.write_text(json.dumps(self._keyring("trusted-key"))) self._git(web_root, "init", "--quiet") self._git(web_root, "config", "user.email", "test@example.test") self._git(web_root, "config", "user.name", "Test") self._git(web_root, "branch", "-M", "main") self._git(web_root, "remote", "add", "origin", str(remote_root)) self._git(web_root, "add", ".") self._git(web_root, "commit", "--quiet", "-m", "base") self._git(web_root, "push", "--quiet", "-u", "origin", "main") frozen_head = self._git(web_root, "rev-parse", "HEAD").strip() frozen_remote = bind_publication_remote( web_root=web_root, remote="origin", registered_remote=str(remote_root), ) with ( patch( "govoplan_release.publisher.validate_module_package_catalog", return_value={"valid": True, "warnings": [], "error": None}, ), patch( "govoplan_release.publisher.source_tag_provenance_issues", return_value=(), ), patch( "govoplan_release.publisher.publication_runtime_trust_issues", return_value=(), ), patch( "govoplan_release.publisher.registered_website_remote", return_value=str(remote_root), ), ): result = publish_catalog_candidate( candidate_dir=candidate, web_root=web_root, workspace_root=root, apply=True, push=True, branch="main", tag_name="catalog-test", expected_website_head=frozen_head, expected_website_branch="main", expected_remote_sha256=frozen_remote.sha256, ) remote_identity = remote_publication_identity( web_root=web_root, remote_url=str(remote_root), branch="main", tag_name="catalog-test", ) self.assertEqual("published", result.status) self.assertEqual("origin", result.remote) self.assertEqual( remote_identity["publication_commit_sha"], result.publication_commit_sha ) self.assertEqual( remote_identity["publication_tag_object_sha"], result.publication_tag_object_sha, ) self.assertEqual( remote_identity["publication_tag_commit_sha"], result.publication_tag_commit_sha, ) def test_apply_writes_validated_objects_even_if_candidate_path_changes( self, ) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) candidate = self._candidate(root, key="trusted-key") catalog_path = candidate / "channels" / "stable.json" catalog = json.loads(catalog_path.read_text(encoding="utf-8")) catalog.update({"channel": "stable", "sequence": 1}) catalog["core_release"]["python_package"] = "govoplan-core" catalog["release"] = { "selected_units": [ { "repo": "govoplan-core", "version": "1.2.3", "tag": "v1.2.3", "commit_sha": "a" * 40, "tag_object_sha": "b" * 40, } ], "artifacts": [ { "artifact_kind": "python-wheel", "package_name": "govoplan-core", "package_version": "1.2.3", "archive_sha256": "c" * 64, "archive_size": 100, "installed_payload": { "algorithm": "govoplan-wheel-declared-payload-v1", "sha256": "d" * 64, "file_count": 1, }, "requires_installer_receipt": True, } ], } catalog["signatures"] = [{}] catalog_path.write_text(json.dumps(catalog), encoding="utf-8") web_root = root / "website" target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json" target_keyring.parent.mkdir(parents=True) target_keyring.write_text(json.dumps(self._keyring("trusted-key"))) registered_remote = "ssh://example.test/release/website.git" self._git(web_root, "init", "--quiet") self._git(web_root, "config", "user.email", "test@example.test") self._git(web_root, "config", "user.name", "Test") self._git(web_root, "remote", "add", "origin", registered_remote) self._git(web_root, "add", ".") self._git(web_root, "commit", "--quiet", "-m", "base") def validate_and_swap(*args, **kwargs): del args, kwargs catalog_path.write_text( json.dumps( { "channel": "stable", "sequence": 999, "malicious": True, "signatures": [{}], } ), encoding="utf-8", ) return {"valid": True, "warnings": [], "error": None} with ( patch( "govoplan_release.publisher.validate_module_package_catalog", side_effect=validate_and_swap, ), patch( "govoplan_release.publisher.source_tag_provenance_issues", return_value=(), ), patch("govoplan_release.publisher.website_dirty", return_value=False), patch( "govoplan_release.publisher.publication_runtime_trust_issues", return_value=(), ), patch( "govoplan_release.publisher.registered_website_remote", return_value=registered_remote, ), ): result = publish_catalog_candidate( candidate_dir=candidate, web_root=web_root, workspace_root=root, apply=True, allow_dirty_website=True, ) published = json.loads( ( web_root / "public" / "catalogs" / "v1" / "channels" / "stable.json" ).read_text(encoding="utf-8") ) self.assertEqual("applied", result.status) self.assertEqual(1, published["sequence"]) self.assertNotIn("malicious", published) def test_apply_blocks_selected_python_release_without_built_identity(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) candidate = self._candidate(root, key="trusted-key") catalog_path = candidate / "channels" / "stable.json" catalog = json.loads(catalog_path.read_text(encoding="utf-8")) catalog["core_release"]["python_package"] = "govoplan-core" catalog["release"] = { "selected_units": [ { "repo": "govoplan-core", "version": "1.2.3", "tag": "v1.2.3", "commit_sha": "a" * 40, "tag_object_sha": "b" * 40, } ] } catalog_path.write_text(json.dumps(catalog), encoding="utf-8") web_root = root / "website" target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json" target_keyring.parent.mkdir(parents=True) target_keyring.write_text(json.dumps(self._keyring("trusted-key"))) (web_root / ".git").mkdir() with ( patch( "govoplan_release.publisher.validate_module_package_catalog", return_value={"valid": True, "warnings": [], "error": None}, ), patch( "govoplan_release.publisher.source_tag_provenance_issues", return_value=(), ), patch("govoplan_release.publisher.website_dirty", return_value=False), ): result = publish_catalog_candidate( candidate_dir=candidate, web_root=web_root, workspace_root=root, apply=True, ) self.assertEqual("blocked", result.status) self.assertIn( "selected Python repository govoplan-core has no built artifact identity", " ".join(result.notes), ) def test_publication_requires_an_existing_trust_anchor(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) candidate = self._candidate(root, key="candidate-key") web_root = root / "website" web_root.mkdir() with patch( "govoplan_release.publisher.validate_module_package_catalog", return_value={"valid": True, "warnings": [], "error": None}, ): result = publish_catalog_candidate( candidate_dir=candidate, web_root=web_root, workspace_root=root, ) self.assertEqual("blocked", result.status) self.assertIn("publication trust anchor is missing", " ".join(result.notes)) def test_publication_rejects_rebinding_an_existing_key_id(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) candidate = self._candidate(root, key="replacement-key") web_root = root / "website" target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json" target_keyring.parent.mkdir(parents=True) target_keyring.write_text(json.dumps(self._keyring("trusted-key"))) with patch( "govoplan_release.publisher.validate_module_package_catalog", return_value={"valid": True, "warnings": [], "error": None}, ): result = publish_catalog_candidate( candidate_dir=candidate, web_root=web_root, workspace_root=root, ) self.assertEqual("blocked", result.status) self.assertIn("changes the public key", " ".join(result.notes)) @staticmethod def _candidate(root: Path, *, key: str) -> Path: candidate = root / "candidate" channel = candidate / "channels" channel.mkdir(parents=True) channel.joinpath("stable.json").write_text( json.dumps( { "channel": "stable", "core_release": { "version": "1.2.3", "python_ref": ( "govoplan-core @ git+ssh://git@example.test/acme/" "govoplan-core.git@v1.2.3" ), }, "modules": [], } ) ) candidate.joinpath("keyring.json").write_text( json.dumps(ReleaseCatalogPublicationTests._keyring(key)) ) return candidate @staticmethod def _keyring(key: str) -> dict[str, object]: return { "keys": [ { "key_id": "release-key", "status": "active", "public_key": key, } ] } @staticmethod def _git(root: Path, *args: str) -> str: return subprocess.run( ["git", *args], cwd=root, check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ).stdout if __name__ == "__main__": unittest.main()