From 349a099e5a35edb9dcd6e65691fbfcf1bee6b0bc Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 21:20:08 +0200 Subject: [PATCH] Harden catalog publication transaction --- tests/test_release_catalog_publication.py | 377 ++++- tools/release/govoplan_release/model.py | 4 + tools/release/govoplan_release/publisher.py | 1499 +++++++++++++++++-- tools/release/release-catalog.py | 105 +- 4 files changed, 1807 insertions(+), 178 deletions(-) diff --git a/tests/test_release_catalog_publication.py b/tests/test_release_catalog_publication.py index 12fa3b0..da1e027 100644 --- a/tests/test_release_catalog_publication.py +++ b/tests/test_release_catalog_publication.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os from pathlib import Path import subprocess import sys @@ -15,8 +16,15 @@ 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, verify_committed_publication, + verify_remote_branch_head, + verify_remote_publication, ) @@ -26,13 +34,17 @@ class ReleaseCatalogPublicationTests(unittest.TestCase): 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" + 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', + (module_root / "index.json") + .relative_to(web_root) + .as_posix(): b'{"modules":[]}\n', } for relative, encoded in expected.items(): target = web_root / relative @@ -72,7 +84,339 @@ class ReleaseCatalogPublicationTests(unittest.TestCase): module_root=module_root, ) - def test_apply_writes_validated_objects_even_if_candidate_path_changes(self) -> None: + 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") @@ -112,7 +456,13 @@ class ReleaseCatalogPublicationTests(unittest.TestCase): 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() + 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 @@ -139,6 +489,14 @@ class ReleaseCatalogPublicationTests(unittest.TestCase): 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, @@ -150,12 +508,7 @@ class ReleaseCatalogPublicationTests(unittest.TestCase): published = json.loads( ( - web_root - / "public" - / "catalogs" - / "v1" - / "channels" - / "stable.json" + web_root / "public" / "catalogs" / "v1" / "channels" / "stable.json" ).read_text(encoding="utf-8") ) @@ -274,7 +627,9 @@ class ReleaseCatalogPublicationTests(unittest.TestCase): } ) ) - candidate.joinpath("keyring.json").write_text(json.dumps(ReleaseCatalogPublicationTests._keyring(key))) + candidate.joinpath("keyring.json").write_text( + json.dumps(ReleaseCatalogPublicationTests._keyring(key)) + ) return candidate @staticmethod diff --git a/tools/release/govoplan_release/model.py b/tools/release/govoplan_release/model.py index 3d4182a..8d9b5cc 100644 --- a/tools/release/govoplan_release/model.py +++ b/tools/release/govoplan_release/model.py @@ -334,6 +334,7 @@ class CatalogPublishResult: target_keyring_path: str branch: str | None tag_name: str | None + remote: str validation_valid: bool validation_error: str | None = None validation_warnings: tuple[str, ...] = () @@ -343,6 +344,9 @@ class CatalogPublishResult: target_keyring_hash_before: str | None = None catalog_changed: bool = False keyring_changed: bool = False + publication_commit_sha: str | None = None + publication_tag_object_sha: str | None = None + publication_tag_commit_sha: str | None = None steps: tuple[CatalogPublishStep, ...] = () notes: tuple[str, ...] = () diff --git a/tools/release/govoplan_release/publisher.py b/tools/release/govoplan_release/publisher.py index 9e55d7c..9320bc7 100644 --- a/tools/release/govoplan_release/publisher.py +++ b/tools/release/govoplan_release/publisher.py @@ -2,18 +2,22 @@ from __future__ import annotations +from dataclasses import dataclass from datetime import UTC, datetime +import hashlib import json import os from pathlib import Path import re import secrets import shlex -import shutil import stat import subprocess +import sys import tempfile +import cryptography +import govoplan_core from govoplan_core.core.module_package_catalog import validate_module_package_catalog from .catalog import canonical_hash @@ -21,10 +25,37 @@ from .artifact_identity import selected_artifact_identity_issues from .candidate_artifact import validate_release_channel from .model import CatalogPublishResult, CatalogPublishStep from .module_directory import module_directory_payloads -from .selective_catalog import trusted_keys_from_keyring +from .selective_catalog import read_bounded_json_source, trusted_keys_from_keyring from .source_provenance import catalog_source_selection, source_tag_provenance_issues from .version_alignment import candidate_catalog_version_issues -from .workspace import DEFAULT_WORKSPACE_ROOT, resolve_workspace_root, website_root +from .workspace import ( + DEFAULT_WORKSPACE_ROOT, + META_ROOT, + load_repository_specs, + resolve_workspace_root, + website_root, +) + + +_GIT_OBJECT = re.compile(r"^[0-9a-f]{40,64}$") +_SHA256 = re.compile(r"^[0-9a-f]{64}$") +_REMOTE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$") +_MAX_GIT_OUTPUT_BYTES = 16 * 1024 * 1024 +_MAX_REMOTE_METADATA_BYTES = 16 * 1024 +_MAX_REMOVED_TREE_ENTRIES = 100_000 +_MAX_REMOVED_TREE_DEPTH = 64 + + +class CatalogPublicationAmbiguousError(RuntimeError): + """A remote publication may have occurred but failed independent proof.""" + + +@dataclass(frozen=True, slots=True) +class FrozenPublicationRemote: + """One registered fetch/push endpoint frozen before publication.""" + + url: str + sha256: str def publish_catalog_candidate( @@ -44,16 +75,32 @@ def publish_catalog_candidate( npm: str = "npm", tag_name: str | None = None, allow_dirty_website: bool = False, + expected_website_head: str | None = None, + expected_website_branch: str | None = None, + expected_remote_sha256: str | None = None, ) -> CatalogPublishResult: channel = validate_release_channel(channel) + if _REMOTE_NAME.fullmatch(remote) is None: + raise ValueError("publication remote must be a bounded configured name") workspace = resolve_workspace_root(workspace_root or DEFAULT_WORKSPACE_ROOT) candidate_root = Path(candidate_dir).expanduser() if not candidate_root.is_absolute(): candidate_root = Path.cwd() / candidate_root - resolved_web_root = Path(web_root).expanduser().resolve() if web_root is not None else website_root(workspace) + resolved_web_root = ( + Path(web_root).expanduser().absolute() + if web_root is not None + else website_root(workspace).absolute() + ) candidate_catalog = candidate_root / "channels" / f"{channel}.json" candidate_keyring = candidate_root / "keyring.json" - target_catalog = resolved_web_root / "public" / "catalogs" / "v1" / "channels" / f"{channel}.json" + target_catalog = ( + resolved_web_root + / "public" + / "catalogs" + / "v1" + / "channels" + / f"{channel}.json" + ) target_keyring = resolved_web_root / "public" / "catalogs" / "v1" / "keyring.json" target_modules = resolved_web_root / "public" / "catalogs" / "v1" / "modules" @@ -68,23 +115,51 @@ def publish_catalog_candidate( if not resolved_web_root.exists(): blockers.append(f"website root is missing: {resolved_web_root}") if push: + tag = True commit = True if tag: commit = True - candidate_payload = read_json(candidate_catalog) if candidate_catalog.exists() else None - keyring_payload = read_json(candidate_keyring) if candidate_keyring.exists() else None - candidate_catalog_hash = canonical_hash(candidate_payload) if candidate_payload is not None else None - candidate_keyring_hash = canonical_hash(keyring_payload) if keyring_payload is not None else None - target_catalog_hash_before = file_json_hash(target_catalog) - target_keyring_hash_before = file_json_hash(target_keyring) + candidate_payload = ( + read_json(candidate_catalog) if candidate_catalog.exists() else None + ) + keyring_payload = ( + read_json(candidate_keyring) if candidate_keyring.exists() else None + ) + target_catalog_payload = ( + read_json(target_catalog) if target_catalog.exists() else None + ) + target_keyring_payload = ( + read_json(target_keyring) if target_keyring.exists() else None + ) + candidate_catalog_hash = ( + canonical_hash(candidate_payload) if candidate_payload is not None else None + ) + candidate_keyring_hash = ( + canonical_hash(keyring_payload) if keyring_payload is not None else None + ) + target_catalog_hash_before = ( + canonical_hash(target_catalog_payload) + if target_catalog_payload is not None + else None + ) + target_keyring_hash_before = ( + canonical_hash(target_keyring_payload) + if target_keyring_payload is not None + else None + ) validation_valid = False validation_error: str | None = None validation_warnings: tuple[str, ...] = () if candidate_payload is not None and keyring_payload is not None: - if not isinstance(candidate_payload, dict) or candidate_payload.get("channel") != channel: - blockers.append("candidate catalog channel does not match the publication channel") + if ( + not isinstance(candidate_payload, dict) + or candidate_payload.get("channel") != channel + ): + blockers.append( + "candidate catalog channel does not match the publication channel" + ) version_issues = candidate_catalog_version_issues(candidate_payload) blockers.extend( f"candidate version alignment: {issue.source}: {issue.actual!r}; " @@ -115,11 +190,11 @@ def publish_catalog_candidate( expected_tag_objects=source_selection.selected_tag_objects, ) blockers.extend( - f"source provenance: {issue.describe()}" - for issue in source_issues + f"source provenance: {issue.describe()}" for issue in source_issues + ) + candidate_keys = trusted_keys_from_keyring( + keyring_payload if isinstance(keyring_payload, dict) else {} ) - candidate_keys = trusted_keys_from_keyring(keyring_payload if isinstance(keyring_payload, dict) else {}) - target_keyring_payload = read_json(target_keyring) if target_keyring.exists() else None publication_trust = trusted_keys_from_keyring( target_keyring_payload if isinstance(target_keyring_payload, dict) else {} ) @@ -129,10 +204,18 @@ def publish_catalog_candidate( ) for key_id in sorted(set(candidate_keys) & set(publication_trust)): if candidate_keys[key_id] != publication_trust[key_id]: - blockers.append(f"candidate keyring changes the public key for trusted key id {key_id!r}") + blockers.append( + f"candidate keyring changes the public key for trusted key id {key_id!r}" + ) if candidate_keyring_hash != target_keyring_hash_before: - release = candidate_payload.get("release") if isinstance(candidate_payload, dict) else None - signed_keyring_hash = release.get("keyring_sha256") if isinstance(release, dict) else None + release = ( + candidate_payload.get("release") + if isinstance(candidate_payload, dict) + else None + ) + signed_keyring_hash = ( + release.get("keyring_sha256") if isinstance(release, dict) else None + ) if signed_keyring_hash != candidate_keyring_hash: blockers.append( "candidate keyring differs from the publication trust anchor without a matching signed keyring_sha256" @@ -145,29 +228,154 @@ def publish_catalog_candidate( ) validation_valid = validation.get("valid") is True validation_error = str(validation["error"]) if validation.get("error") else None - validation_warnings = tuple(str(item) for item in validation.get("warnings") or ()) + validation_warnings = tuple( + str(item) for item in validation.get("warnings") or () + ) if not validation_valid: blockers.append(validation_error or "candidate catalog validation failed") catalog_changed = candidate_catalog_hash != target_catalog_hash_before keyring_changed = candidate_keyring_hash != target_keyring_hash_before if not catalog_changed and not keyring_changed: - notes.append("Candidate catalog and keyring already match the website repository.") + notes.append( + "Candidate catalog and keyring already match the website repository." + ) if apply or commit or tag or push: if not (resolved_web_root / ".git").exists(): - blockers.append(f"website root is not a Git repository: {resolved_web_root}") + blockers.append( + f"website root is not a Git repository: {resolved_web_root}" + ) if not allow_dirty_website and website_dirty(resolved_web_root): blockers.append("website repository has uncommitted changes") + if apply: + blockers.extend(publication_runtime_trust_issues()) + blockers.extend( + publication_mutation_trust_issues( + web_root=resolved_web_root, + candidate_catalog=candidate_catalog, + candidate_keyring=candidate_keyring, + target_catalog=target_catalog, + target_keyring=target_keyring, + target_modules=target_modules, + ) + ) + if not git_success(resolved_web_root, "diff", "--cached", "--quiet"): + blockers.append( + "website Git index has staged changes; publication requires a clean index" + ) - effective_branch = branch or git_text(resolved_web_root, "branch", "--show-current") or None - effective_tag_name = tag_name or default_tag_name(candidate_payload, channel=channel) + effective_branch = ( + branch or git_text(resolved_web_root, "branch", "--show-current") or None + ) + effective_tag_name = tag_name or default_tag_name( + candidate_payload, channel=channel + ) + frozen_website_head: str | None = None + frozen_remote: FrozenPublicationRemote | None = None + expected_binding = ( + expected_website_head, + expected_website_branch, + expected_remote_sha256, + ) + if any(value is not None for value in expected_binding) and not all( + isinstance(value, str) and value for value in expected_binding + ): + blockers.append( + "immutable website head, branch, and remote binding must be supplied together" + ) + elif all(value is not None for value in expected_binding) and ( + _GIT_OBJECT.fullmatch(expected_website_head or "") is None + or _SHA256.fullmatch(expected_remote_sha256 or "") is None + ): + blockers.append("immutable website release binding is malformed") if (tag or push) and not effective_tag_name: blockers.append("could not infer catalog tag name; pass --tag-name") if (commit or push) and not effective_branch: blockers.append("could not infer website branch; pass --branch") - if tag and effective_tag_name and git_success(resolved_web_root, "rev-parse", "--quiet", "--verify", f"refs/tags/{effective_tag_name}"): + for value, prefix, label in ( + (effective_branch, "refs/heads", "website branch"), + (effective_tag_name, "refs/tags", "website publication tag"), + ): + if value: + try: + if value.startswith("-"): + raise ValueError(f"{label} must not begin with an option prefix") + _validated_git_ref( + resolved_web_root, + f"{prefix}/{value}", + label=label, + ) + except (RuntimeError, subprocess.SubprocessError, ValueError) as exc: + blockers.append(f"{label} is not safe: {exc}") + if ( + tag + and effective_tag_name + and git_success( + resolved_web_root, + "rev-parse", + "--quiet", + "--verify", + f"refs/tags/{effective_tag_name}", + ) + ): blockers.append(f"website tag already exists: {effective_tag_name}") + if apply and commit and effective_branch: + frozen_website_head = git_text( + resolved_web_root, "rev-parse", "--verify", "HEAD^{commit}" + ) + frozen_branch_head = git_text( + resolved_web_root, + "rev-parse", + "--verify", + f"refs/heads/{effective_branch}^{{commit}}", + ) + if ( + _GIT_OBJECT.fullmatch(frozen_website_head) is None + or frozen_branch_head != frozen_website_head + ): + blockers.append( + "website branch and HEAD do not identify one frozen publication parent" + ) + if ( + expected_website_head is not None + and frozen_website_head != expected_website_head + ): + blockers.append("website HEAD differs from the immutable release binding") + if ( + expected_website_branch is not None + and effective_branch != expected_website_branch + ): + blockers.append("website branch differs from the immutable release binding") + if apply: + try: + frozen_remote = bind_publication_remote( + web_root=resolved_web_root, + remote=remote, + registered_remote=registered_website_remote(), + ) + if ( + expected_remote_sha256 is not None + and frozen_remote.sha256 != expected_remote_sha256 + ): + raise RuntimeError( + "website remote differs from the immutable release binding" + ) + if commit and effective_branch and frozen_website_head: + verify_remote_branch_head( + web_root=resolved_web_root, + remote_url=frozen_remote.url, + branch=effective_branch, + expected_commit=frozen_website_head, + ) + except ( + OSError, + RuntimeError, + subprocess.SubprocessError, + UnicodeError, + ValueError, + ) as exc: + blockers.append(f"website publication remote cannot be frozen: {exc}") steps.append( CatalogPublishStep( @@ -192,7 +400,9 @@ def publish_catalog_candidate( id="build-web", title="Build website", detail="Run the website build after copying catalog files.", - command=shlex.join([npm, "--prefix", str(resolved_web_root), "run", "build"]), + command=shlex.join( + [npm, "--prefix", str(resolved_web_root), "run", "build"] + ), mutating=True, status="planned" if not apply else "blocked" if blockers else "pending", ) @@ -203,7 +413,16 @@ def publish_catalog_candidate( id="commit", title="Commit website catalog update", detail="Commit generated catalog/keyring files in the website repository.", - command=shlex.join(["git", "-C", str(resolved_web_root), "commit", "-m", commit_message(candidate_payload, channel=channel)]), + command=shlex.join( + [ + "git", + "-C", + str(resolved_web_root), + "commit", + "-m", + commit_message(candidate_payload, channel=channel), + ] + ), mutating=True, status="planned" if not apply else "blocked" if blockers else "pending", ) @@ -214,7 +433,18 @@ def publish_catalog_candidate( id="tag", title=f"Tag website catalog publication {effective_tag_name}", detail="Create an annotated website publication tag.", - command=shlex.join(["git", "-C", str(resolved_web_root), "tag", "-a", effective_tag_name, "-m", commit_message(candidate_payload, channel=channel)]), + command=shlex.join( + [ + "git", + "-C", + str(resolved_web_root), + "tag", + "-a", + effective_tag_name, + "-m", + commit_message(candidate_payload, channel=channel), + ] + ), mutating=True, status="planned" if not apply else "blocked" if blockers else "pending", ) @@ -225,7 +455,12 @@ def publish_catalog_candidate( id="push", title="Push website catalog publication", detail="Push website branch and tag if a tag was requested.", - command=push_command(resolved_web_root, remote=remote, branch=effective_branch, tag_name=effective_tag_name if tag else None), + command=push_command( + resolved_web_root, + remote=remote, + branch=effective_branch, + tag_name=effective_tag_name if tag else None, + ), mutating=True, status="planned" if not apply else "blocked" if blockers else "pending", ) @@ -244,6 +479,7 @@ def publish_catalog_candidate( channel=channel, branch=effective_branch, tag_name=effective_tag_name, + remote=remote, validation_valid=validation_valid, validation_error=validation_error, validation_warnings=validation_warnings, @@ -258,7 +494,9 @@ def publish_catalog_candidate( ) if not apply: - notes.append("Dry run only. Pass --apply to copy files; pass --commit/--tag/--push for Git publication steps.") + notes.append( + "Dry run only. Pass --apply to copy files; pass --commit/--tag/--push for Git publication steps." + ) return result( status="ready", applied=False, @@ -271,6 +509,7 @@ def publish_catalog_candidate( channel=channel, branch=effective_branch, tag_name=effective_tag_name, + remote=remote, validation_valid=validation_valid, validation_error=validation_error, validation_warnings=validation_warnings, @@ -315,55 +554,126 @@ def publish_catalog_candidate( directory_payloads=directory_payloads, ) - completed_steps = [replace_status(step, "done") if step.id == "copy" else step for step in steps] + completed_steps = [ + replace_status(step, "done") if step.id == "copy" else step for step in steps + ] if build_web: - run_checked([npm, "--prefix", str(resolved_web_root), "run", "build"], cwd=resolved_web_root) - completed_steps = [replace_status(step, "done") if step.id == "build-web" else step for step in completed_steps] + run_checked( + [npm, "--prefix", str(resolved_web_root), "run", "build"], + cwd=resolved_web_root, + ) + completed_steps = [ + replace_status(step, "done") if step.id == "build-web" else step + for step in completed_steps + ] publication_commit: str | None = None if commit: - add_paths = ["git", "-C", str(resolved_web_root), "add", str(target_catalog), str(target_keyring)] - if target_modules.exists(): - add_paths.append(str(target_modules)) - run_checked(add_paths, cwd=resolved_web_root) - if git_success(resolved_web_root, "diff", "--cached", "--quiet"): - notes.append("No website catalog changes were staged for commit.") - completed_steps = [replace_status(step, "skipped") if step.id == "commit" else step for step in completed_steps] - else: - run_checked(["git", "-C", str(resolved_web_root), "commit", "-m", commit_message(candidate_payload, channel=channel)], cwd=resolved_web_root) - completed_steps = [replace_status(step, "done") if step.id == "commit" else step for step in completed_steps] - publication_commit = committed_publication_sha(resolved_web_root) - verify_committed_publication( + if frozen_website_head is None or effective_branch is None: + raise RuntimeError("publication parent was not frozen before mutation") + publication_commit = commit_publication_tree( web_root=resolved_web_root, - commit_sha=publication_commit, + frozen_head=frozen_website_head, + branch=effective_branch, expected_blobs=expected_blobs, module_root=target_modules, + message=commit_message(candidate_payload, channel=channel), ) + completed_steps = [ + replace_status(step, "done") if step.id == "commit" else step + for step in completed_steps + ] publication_tag_object: str | None = None + publication_tag_commit: str | None = None if tag and effective_tag_name and publication_commit: - run_checked(["git", "-C", str(resolved_web_root), "tag", "-a", effective_tag_name, publication_commit, "-m", commit_message(candidate_payload, channel=channel)], cwd=resolved_web_root) - publication_tag_object = git_text( - resolved_web_root, "rev-parse", "--verify", f"refs/tags/{effective_tag_name}" + _git_checked( + resolved_web_root, + "tag", + "--no-sign", + "-a", + "-m", + commit_message(candidate_payload, channel=channel), + "--", + effective_tag_name, + publication_commit, ) - tagged_commit = git_text( + publication_tag_object = git_text( + resolved_web_root, + "rev-parse", + "--verify", + f"refs/tags/{effective_tag_name}", + ) + publication_tag_commit = git_text( resolved_web_root, "rev-parse", "--verify", f"refs/tags/{effective_tag_name}^{{commit}}", ) + publication_tag_type = git_text( + resolved_web_root, + "cat-file", + "-t", + f"refs/tags/{effective_tag_name}", + ) if ( - re.fullmatch(r"[0-9a-f]{40,64}", publication_tag_object) is None - or tagged_commit != publication_commit + _GIT_OBJECT.fullmatch(publication_tag_object) is None + or publication_tag_type != "tag" + or publication_tag_commit != publication_commit ): - raise RuntimeError("website publication tag does not bind the verified commit") - completed_steps = [replace_status(step, "done") if step.id == "tag" else step for step in completed_steps] + raise RuntimeError( + "website publication tag does not bind the verified commit" + ) + completed_steps = [ + replace_status(step, "done") if step.id == "tag" else step + for step in completed_steps + ] if push and effective_branch and publication_commit: - push_args = ["git", "-C", str(resolved_web_root), "push"] + if frozen_remote is None: + raise RuntimeError("publication remote was not frozen before mutation") + push_args = ["-c", "core.hooksPath=/dev/null", "push"] if tag and effective_tag_name and publication_tag_object: - push_args.extend(["--atomic", remote, f"{publication_commit}:refs/heads/{effective_branch}", f"{publication_tag_object}:refs/tags/{effective_tag_name}"]) + push_args.extend( + [ + "--atomic", + frozen_remote.url, + f"{publication_commit}:refs/heads/{effective_branch}", + f"{publication_tag_object}:refs/tags/{effective_tag_name}", + ] + ) else: - push_args.extend([remote, f"{publication_commit}:refs/heads/{effective_branch}"]) - run_checked(push_args, cwd=resolved_web_root) - completed_steps = [replace_status(step, "done") if step.id == "push" else step for step in completed_steps] + push_args.extend( + [ + frozen_remote.url, + f"{publication_commit}:refs/heads/{effective_branch}", + ] + ) + try: + _git_checked( + resolved_web_root, + *push_args, + isolate_config=True, + ) + if ( + effective_tag_name is None + or publication_tag_object is None + or publication_tag_commit is None + ): + raise RuntimeError("published release has no complete tag identity") + verify_remote_publication( + web_root=resolved_web_root, + remote_url=frozen_remote.url, + branch=effective_branch, + tag_name=effective_tag_name, + expected_commit=publication_commit, + expected_tag_object=publication_tag_object, + ) + except Exception as exc: + raise CatalogPublicationAmbiguousError( + "website push may have completed but remote identity verification failed" + ) from exc + completed_steps = [ + replace_status(step, "done") if step.id == "push" else step + for step in completed_steps + ] return result( status="published" if push else "applied", @@ -377,6 +687,7 @@ def publish_catalog_candidate( channel=channel, branch=effective_branch, tag_name=effective_tag_name, + remote=remote, validation_valid=validation_valid, validation_error=validation_error, validation_warnings=validation_warnings, @@ -386,6 +697,9 @@ def publish_catalog_candidate( target_keyring_hash_before=target_keyring_hash_before, catalog_changed=catalog_changed, keyring_changed=keyring_changed, + publication_commit_sha=publication_commit, + publication_tag_object_sha=publication_tag_object, + publication_tag_commit_sha=publication_tag_commit, steps=tuple(completed_steps), notes=tuple(notes), ) @@ -412,6 +726,7 @@ def result( channel: str, branch: str | None, tag_name: str | None, + remote: str, validation_valid: bool, validation_error: str | None, validation_warnings: tuple[str, ...], @@ -421,11 +736,28 @@ def result( target_keyring_hash_before: str | None, catalog_changed: bool, keyring_changed: bool, + publication_commit_sha: str | None = None, + publication_tag_object_sha: str | None = None, + publication_tag_commit_sha: str | None = None, steps: tuple[CatalogPublishStep, ...], notes: tuple[str, ...], ) -> CatalogPublishResult: + if status == "published" and ( + _GIT_OBJECT.fullmatch(publication_commit_sha or "") is None + or _GIT_OBJECT.fullmatch(publication_tag_object_sha or "") is None + or publication_tag_commit_sha != publication_commit_sha + or _SHA256.fullmatch(candidate_catalog_hash or "") is None + or _SHA256.fullmatch(candidate_keyring_hash or "") is None + or not branch + or not tag_name + or _REMOTE_NAME.fullmatch(remote) is None + ): + raise RuntimeError("published catalog result has no complete remote identity") return CatalogPublishResult( - generated_at=datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"), + generated_at=datetime.now(tz=UTC) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z"), channel=channel, status=status, applied=applied, @@ -437,6 +769,7 @@ def result( target_keyring_path=str(target_keyring), branch=branch, tag_name=tag_name, + remote=remote, validation_valid=validation_valid, validation_error=validation_error, validation_warnings=validation_warnings, @@ -446,13 +779,16 @@ def result( target_keyring_hash_before=target_keyring_hash_before, catalog_changed=catalog_changed, keyring_changed=keyring_changed, + publication_commit_sha=publication_commit_sha, + publication_tag_object_sha=publication_tag_object_sha, + publication_tag_commit_sha=publication_tag_commit_sha, steps=steps, notes=notes, ) def read_json(path: Path) -> object: - return json.loads(path.read_text(encoding="utf-8")) + return read_bounded_json_source(path, label="catalog JSON") def validate_catalog_payload( @@ -521,15 +857,8 @@ def _atomic_write_public_json( if observed is not None and not stat.S_ISREG(observed.st_mode): raise ValueError("published catalog target must be a regular file") temporary_name = f".{target_name}.{secrets.token_hex(16)}.tmp" - flags = ( - os.O_WRONLY - | os.O_CREAT - | os.O_EXCL - | getattr(os, "O_NOFOLLOW", 0) - ) - descriptor = os.open( - temporary_name, flags, 0o600, dir_fd=parent_descriptor - ) + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(temporary_name, flags, 0o600, dir_fd=parent_descriptor) replaced = False try: with os.fdopen(descriptor, "wb", closefd=True) as handle: @@ -573,7 +902,9 @@ def _trusted_relative_path(path: Path, *, trusted_root: Path) -> Path: try: relative = path.absolute().relative_to(trusted_root.absolute()) except ValueError as exc: - raise ValueError("published catalog target leaves the trusted worktree") from exc + raise ValueError( + "published catalog target leaves the trusted worktree" + ) from exc if not relative.parts or any(part in {"", ".", ".."} for part in relative.parts): raise ValueError("published catalog target is not canonical") return relative @@ -582,8 +913,8 @@ def _trusted_relative_path(path: Path, *, trusted_root: Path) -> Path: def _open_relative_directory( trusted_root: Path, parts: tuple[str, ...], *, create: bool ) -> int: - root_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr( - os, "O_NOFOLLOW", 0 + root_flags = ( + os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) ) descriptor = os.open(trusted_root, root_flags) try: @@ -607,17 +938,331 @@ def _open_relative_directory( raise -def _remove_public_tree(path: Path, *, trusted_root: Path) -> None: - _trusted_relative_path(path, trusted_root=trusted_root) - if not shutil.rmtree.avoids_symlink_attacks: - raise RuntimeError("safe directory removal is unavailable on this platform") +def publication_mutation_trust_issues( + *, + web_root: Path, + candidate_catalog: Path, + candidate_keyring: Path, + target_catalog: Path, + target_keyring: Path, + target_modules: Path, +) -> tuple[str, ...]: + """Reject mutable trust inputs and unsafe website/Git ownership boundaries.""" + + issues: list[str] = [] + for path, label in ( + (web_root, "website root"), + (candidate_catalog.parent, "candidate catalog root"), + (candidate_keyring.parent, "candidate keyring root"), + ): + issue = _operator_controlled_ancestor_issue(path, allow_sticky=True) + if issue: + issues.append(f"{label}: {issue}") + + checks: list[tuple[Path, bool, bool, str]] = [ + (web_root, True, True, "website root"), + (web_root / ".git", True, True, "website Git directory"), + (web_root / ".git" / "config", False, True, "website Git config"), + (candidate_catalog, False, True, "candidate catalog"), + (candidate_keyring, False, True, "candidate keyring"), + (target_keyring, False, True, "published trust keyring"), + (target_catalog, False, False, "published channel catalog"), + (web_root / ".git" / "HEAD", False, False, "website Git HEAD"), + (web_root / ".git" / "index", False, False, "website Git index"), + (web_root / ".git" / "objects", True, False, "website Git objects"), + (web_root / ".git" / "refs", True, False, "website Git refs"), + ] + for target in (target_catalog.parent, target_keyring.parent, target_modules.parent): + try: + relative = target.absolute().relative_to(web_root.absolute()) + except ValueError: + return ("publication target leaves the owned website root",) + current = web_root + for part in relative.parts: + current /= part + checks.append((current, True, False, "website publication directory")) + seen: set[Path] = set() + for path, directory, required, label in checks: + if path in seen: + continue + seen.add(path) + try: + observed = path.lstat() + except FileNotFoundError: + if required: + issues.append(f"{label} is missing") + continue + except OSError: + issues.append(f"{label} cannot be inspected safely") + continue + expected_type = stat.S_ISDIR if directory else stat.S_ISREG + if stat.S_ISLNK(observed.st_mode) or not expected_type(observed.st_mode): + issues.append( + f"{label} must be a real {'directory' if directory else 'file'}" + ) + elif observed.st_uid != os.geteuid(): + issues.append(f"{label} is not owned by the current operator") + elif stat.S_IMODE(observed.st_mode) & 0o022: + issues.append(f"{label} is writable by another user") + return tuple(issues) + + +def publication_runtime_trust_issues() -> tuple[str, ...]: + """Require the running meta release code/config to be operator-controlled.""" + + release_root = META_ROOT / "tools" / "release" + runtime_owners = {0, os.geteuid()} + package_roots: tuple[tuple[Path, str], ...] = ( + (_loaded_package_root(govoplan_core, label="govoplan_core"), "GovOPlaN Core"), + (_loaded_package_root(cryptography, label="cryptography"), "cryptography"), + ) + for path, label in ( + (META_ROOT, "meta repository"), + (Path(sys.prefix), "Python environment"), + *package_roots, + ): + ancestor_issue = _operator_controlled_ancestor_issue(path) + if ancestor_issue: + return (f"{label}: {ancestor_issue}",) + executable_issue = _trusted_runtime_executable_issue( + Path(sys.executable), permitted_owners=runtime_owners + ) + if executable_issue: + return (executable_issue,) + required = ( + (META_ROOT, True, "meta repository root"), + (META_ROOT / "repositories.json", False, "repository registry"), + (release_root, True, "release tooling root"), + ) + issues: list[str] = [] + for path, directory, label in required: + issue = _operator_control_issue( + path, + directory=directory, + label=label, + permitted_owners=runtime_owners, + ) + if issue: + issues.append(issue) + if issues: + return tuple(issues) + + pending = [release_root, *(path for path, _label in package_roots)] + inspected = 0 + while pending: + parent = pending.pop() + try: + entries = tuple(os.scandir(parent)) + except OSError: + return ("release tooling tree cannot be inspected safely",) + for entry in entries: + inspected += 1 + if inspected > 4_096: + return ("release tooling tree exceeds its trust inspection bound",) + path = Path(entry.path) + try: + observed = path.lstat() + except OSError: + return ("release tooling tree changed during trust inspection",) + if stat.S_ISDIR(observed.st_mode): + pending.append(path) + directory = True + elif stat.S_ISREG(observed.st_mode): + directory = False + else: + return ( + f"release tooling path is not a real file or directory: {path}", + ) + issue = _operator_control_issue( + path, + directory=directory, + label="release tooling path", + observed=observed, + permitted_owners=runtime_owners, + ) + if issue: + return (issue,) + return () + + +def _loaded_package_root(module: object, *, label: str) -> Path: + source = getattr(module, "__file__", None) + if not isinstance(source, str) or not source: + raise RuntimeError(f"{label} package origin cannot be identified") + return Path(source).absolute().parent + + +def _trusted_runtime_executable_issue( + path: Path, *, permitted_owners: set[int] +) -> str | None: + executable = path.absolute() + ancestor_issue = _operator_controlled_ancestor_issue(executable.parent) + if ancestor_issue: + return f"Python executable: {ancestor_issue}" try: - mode = path.lstat().st_mode - except FileNotFoundError: - return - if stat.S_ISLNK(mode) or not stat.S_ISDIR(mode): - raise ValueError("published module target must be a real directory") - shutil.rmtree(path) + link_metadata = executable.lstat() + except OSError: + return f"Python executable cannot be inspected safely: {executable}" + if stat.S_ISLNK(link_metadata.st_mode): + if link_metadata.st_uid not in permitted_owners: + return f"Python executable symlink has an untrusted owner: {executable}" + try: + target = executable.resolve(strict=True) + except OSError: + return f"Python executable symlink cannot be resolved safely: {executable}" + else: + target = executable + ancestor_issue = _operator_controlled_ancestor_issue(target.parent) + if ancestor_issue: + return f"Python executable target: {ancestor_issue}" + return _operator_control_issue( + target, + directory=False, + label="Python executable target", + permitted_owners=permitted_owners, + ) + + +def _operator_controlled_ancestor_issue( + path: Path, *, allow_sticky: bool = False +) -> str | None: + """Reject a trust root that an ancestor directory can be replaced beneath.""" + + current = path.absolute() + permitted_owners = {0, os.geteuid()} + while True: + try: + observed = current.lstat() + except OSError: + return f"release trust-path ancestor cannot be inspected safely: {current}" + if stat.S_ISLNK(observed.st_mode) or not stat.S_ISDIR(observed.st_mode): + return f"release trust-path ancestor is not a real directory: {current}" + if observed.st_uid not in permitted_owners: + return f"release trust-path ancestor has an untrusted owner: {current}" + writable_by_others = stat.S_IMODE(observed.st_mode) & 0o022 + sticky_boundary = allow_sticky and bool(observed.st_mode & stat.S_ISVTX) + if writable_by_others and not sticky_boundary: + return f"release trust-path ancestor is group/world writable: {current}" + if current.parent == current: + break + current = current.parent + return None + + +def _operator_control_issue( + path: Path, + *, + directory: bool, + label: str, + observed: os.stat_result | None = None, + permitted_owners: set[int] | None = None, +) -> str | None: + try: + metadata = observed or path.lstat() + except OSError: + return f"{label} cannot be inspected safely: {path}" + expected_type = stat.S_ISDIR if directory else stat.S_ISREG + if stat.S_ISLNK(metadata.st_mode) or not expected_type(metadata.st_mode): + return f"{label} must be a real {'directory' if directory else 'file'}: {path}" + owners = permitted_owners or {os.geteuid()} + if metadata.st_uid not in owners: + return f"{label} is not owned by the current operator: {path}" + if stat.S_IMODE(metadata.st_mode) & 0o022: + return f"{label} is group/world writable: {path}" + return None + + +def _remove_public_tree(path: Path, *, trusted_root: Path) -> None: + relative = _trusted_relative_path(path, trusted_root=trusted_root) + if len(relative.parts) < 2: + raise ValueError("published module target requires a pinned parent") + parent_descriptor = _open_relative_directory( + trusted_root, relative.parts[:-1], create=False + ) + target_name = relative.parts[-1] + trash_name = f".{target_name}.{secrets.token_hex(16)}.trash" + try: + try: + observed = os.stat( + target_name, dir_fd=parent_descriptor, follow_symlinks=False + ) + except FileNotFoundError: + return + if not stat.S_ISDIR(observed.st_mode): + raise ValueError("published module target must be a real directory") + os.rename( + target_name, + trash_name, + src_dir_fd=parent_descriptor, + dst_dir_fd=parent_descriptor, + ) + trash_descriptor = os.open( + trash_name, + os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), + dir_fd=parent_descriptor, + ) + try: + trash_device = os.fstat(trash_descriptor).st_dev + os.fchmod(trash_descriptor, 0o700) + finally: + os.close(trash_descriptor) + _remove_tree_at( + parent_descriptor, + trash_name, + expected_device=trash_device, + remaining=[_MAX_REMOVED_TREE_ENTRIES], + depth=0, + ) + os.fsync(parent_descriptor) + finally: + os.close(parent_descriptor) + + +def _remove_tree_at( + parent_descriptor: int, + name: str, + *, + expected_device: int, + remaining: list[int], + depth: int, +) -> None: + """Remove a renamed private tree without resolving a mutable parent path.""" + + directory_flags = ( + os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) + ) + if depth > _MAX_REMOVED_TREE_DEPTH: + raise RuntimeError("published module tree exceeds its removal depth bound") + descriptor = os.open(name, directory_flags, dir_fd=parent_descriptor) + try: + if os.fstat(descriptor).st_dev != expected_device: + raise RuntimeError("published module tree crosses a mounted filesystem") + with os.scandir(descriptor) as entries: + for entry in entries: + remaining[0] -= 1 + if remaining[0] < 0: + raise RuntimeError( + "published module tree exceeds its removal entry bound" + ) + child_name = entry.name + observed = os.stat(child_name, dir_fd=descriptor, follow_symlinks=False) + if stat.S_ISDIR(observed.st_mode): + if observed.st_dev != expected_device: + raise RuntimeError( + "published module tree crosses a mounted filesystem" + ) + _remove_tree_at( + descriptor, + child_name, + expected_device=expected_device, + remaining=remaining, + depth=depth + 1, + ) + else: + os.unlink(child_name, dir_fd=descriptor) + finally: + os.close(descriptor) + os.rmdir(name, dir_fd=parent_descriptor) def _expected_publication_blobs( @@ -651,11 +1296,133 @@ def _expected_publication_blobs( def committed_publication_sha(web_root: Path) -> str: commit_sha = git_text(web_root, "rev-parse", "--verify", "HEAD^{commit}") - if re.fullmatch(r"[0-9a-f]{40,64}", commit_sha) is None: + if _GIT_OBJECT.fullmatch(commit_sha) is None: raise RuntimeError("could not resolve the publication commit") return commit_sha +def commit_publication_tree( + *, + web_root: Path, + frozen_head: str, + branch: str, + expected_blobs: dict[str, bytes], + module_root: Path, + message: str, +) -> str: + """Build one exact commit from memory through a private index and no hooks.""" + + if _GIT_OBJECT.fullmatch(frozen_head) is None: + raise RuntimeError("publication parent identity is malformed") + module_prefix = _trusted_relative_path( + module_root, trusted_root=web_root + ).as_posix() + expected_modules = { + path for path in expected_blobs if path.startswith(f"{module_prefix}/") + } + old_modules = _git_tree_paths(web_root, frozen_head, module_prefix) + with tempfile.TemporaryDirectory( + prefix="govoplan-publication-index-" + ) as temporary_directory: + index_name = str(Path(temporary_directory) / "index") + environment = _sanitized_git_environment(private_index=index_name) + _git_checked( + web_root, + "read-tree", + frozen_head, + environment=environment, + ) + for path, encoded in sorted(expected_blobs.items()): + object_id = ( + _git_bytes( + web_root, + "hash-object", + "-w", + "--stdin", + input_bytes=encoded, + ) + .decode("ascii") + .strip() + ) + if _GIT_OBJECT.fullmatch(object_id) is None: + raise RuntimeError("Git returned an invalid publication blob identity") + _git_checked( + web_root, + "update-index", + "--add", + "--cacheinfo", + "100644", + object_id, + path, + environment=environment, + ) + for path in sorted(old_modules - expected_modules): + _git_checked( + web_root, + "update-index", + "--force-remove", + "--", + path, + environment=environment, + ) + tree_sha = _git_text_checked(web_root, "write-tree", environment=environment) + if _GIT_OBJECT.fullmatch(tree_sha) is None: + raise RuntimeError("Git returned an invalid publication tree identity") + _verify_treeish_publication( + web_root=web_root, + treeish=tree_sha, + expected_blobs=expected_blobs, + module_root=module_root, + ) + expected_delta = _expected_catalog_delta_paths( + web_root=web_root, + frozen_head=frozen_head, + expected_blobs=expected_blobs, + old_modules=old_modules, + expected_modules=expected_modules, + ) + observed_delta = _git_diff_paths(web_root, frozen_head, tree_sha) + if observed_delta != expected_delta: + raise RuntimeError( + "publication tree contains paths outside the computed catalog delta" + ) + commit_sha = _git_text_checked( + web_root, + "-c", + "commit.gpgSign=false", + "commit-tree", + tree_sha, + "-p", + frozen_head, + input_bytes=(message + "\n").encode("utf-8"), + ) + if _GIT_OBJECT.fullmatch(commit_sha) is None: + raise RuntimeError("Git returned an invalid publication commit identity") + commit_line = _git_text_checked( + web_root, "rev-list", "--parents", "-n", "1", commit_sha + ) + if commit_line.split() != [commit_sha, frozen_head]: + raise RuntimeError("publication commit does not have the sole frozen parent") + commit_tree = _git_text_checked(web_root, "show", "-s", "--format=%T", commit_sha) + if commit_tree != tree_sha: + raise RuntimeError("publication commit tree differs from the verified tree") + symbolic_head = _git_text_checked(web_root, "symbolic-ref", "--quiet", "HEAD") + current_head = _git_text_checked(web_root, "rev-parse", "--verify", "HEAD^{commit}") + if symbolic_head != f"refs/heads/{branch}" or current_head != frozen_head: + raise RuntimeError("website HEAD changed after publication was frozen") + _git_checked( + web_root, + "-c", + "core.hooksPath=/dev/null", + "update-ref", + f"refs/heads/{branch}", + commit_sha, + frozen_head, + ) + _git_checked(web_root, "read-tree", "--reset", commit_sha) + return commit_sha + + def verify_committed_publication( *, web_root: Path, @@ -665,54 +1432,53 @@ def verify_committed_publication( ) -> None: """Require the exact validated objects in one immutable commit tree.""" - if re.fullmatch(r"[0-9a-f]{40,64}", commit_sha) is None: + if _GIT_OBJECT.fullmatch(commit_sha) is None: raise RuntimeError("publication commit identity is malformed") + _verify_treeish_publication( + web_root=web_root, + treeish=commit_sha, + expected_blobs=expected_blobs, + module_root=module_root, + ) + + +def _verify_treeish_publication( + *, + web_root: Path, + treeish: str, + expected_blobs: dict[str, bytes], + module_root: Path, +) -> None: module_prefix = _trusted_relative_path( module_root, trusted_root=web_root ).as_posix() expected_modules = { path for path in expected_blobs if path.startswith(f"{module_prefix}/") } - tree = subprocess.run( - [ - "git", - "-C", - str(web_root), - "ls-tree", - "-r", - "--name-only", - "-z", - commit_sha, - "--", - module_prefix, - ], - check=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ).stdout - observed_modules = { - item.decode("utf-8") for item in tree.split(b"\0") if item - } + tree = _git_bytes( + web_root, + "ls-tree", + "-r", + "--name-only", + "-z", + treeish, + "--", + module_prefix, + ) + observed_modules = {item.decode("utf-8") for item in tree.split(b"\0") if item} if observed_modules != expected_modules: raise RuntimeError( "committed module directory differs from validated publication outputs" ) for repository_path, expected in expected_blobs.items(): - entry = subprocess.run( - [ - "git", - "-C", - str(web_root), - "ls-tree", - "-z", - commit_sha, - "--", - repository_path, - ], - check=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ).stdout + entry = _git_bytes( + web_root, + "ls-tree", + "-z", + treeish, + "--", + repository_path, + ) metadata, separator, encoded_path = entry.rstrip(b"\0").partition(b"\t") fields = metadata.split() if ( @@ -726,33 +1492,286 @@ def verify_committed_publication( raise RuntimeError( f"committed publication path is not one regular blob: {repository_path}" ) - observed = subprocess.run( - ["git", "-C", str(web_root), "cat-file", "blob", fields[2].decode()], - check=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ).stdout + observed = _git_bytes(web_root, "cat-file", "blob", fields[2].decode("ascii")) if observed != expected: raise RuntimeError( f"committed publication blob differs from validated output: {repository_path}" ) -def _reject_symlink_components(path: Path) -> None: - absolute = path.absolute() - for component in reversed((absolute, *absolute.parents)): - try: - mode = component.lstat().st_mode - except FileNotFoundError: - continue - if stat.S_ISLNK(mode): - raise ValueError("published catalog path must not traverse symlinks") +def _expected_catalog_delta_paths( + *, + web_root: Path, + frozen_head: str, + expected_blobs: dict[str, bytes], + old_modules: set[str], + expected_modules: set[str], +) -> set[str]: + changed = set(old_modules - expected_modules) + for path, expected in expected_blobs.items(): + observed = _git_tree_blob(web_root, frozen_head, path) + if observed is None or observed != (b"100644", expected): + changed.add(path) + return changed -def file_json_hash(path: Path) -> str | None: - if not path.exists(): +def _git_tree_paths(web_root: Path, treeish: str, prefix: str) -> set[str]: + encoded = _git_bytes( + web_root, + "ls-tree", + "-r", + "--name-only", + "-z", + treeish, + "--", + prefix, + ) + return {item.decode("utf-8") for item in encoded.split(b"\0") if item} + + +def _git_tree_blob( + web_root: Path, treeish: str, repository_path: str +) -> tuple[bytes, bytes] | None: + entry = _git_bytes(web_root, "ls-tree", "-z", treeish, "--", repository_path) + if not entry: return None - return canonical_hash(read_json(path)) + metadata, separator, encoded_path = entry.rstrip(b"\0").partition(b"\t") + fields = metadata.split() + if ( + not separator + or encoded_path.decode("utf-8") != repository_path + or len(fields) != 3 + or fields[1] != b"blob" + or re.fullmatch(rb"[0-9a-f]{40,64}", fields[2]) is None + ): + return None + return fields[0], _git_bytes( + web_root, "cat-file", "blob", fields[2].decode("ascii") + ) + + +def _git_diff_paths(web_root: Path, before: str, after: str) -> set[str]: + encoded = _git_bytes( + web_root, + "diff-tree", + "--no-renames", + "--no-ext-diff", + "--no-commit-id", + "--name-only", + "-r", + "-z", + before, + after, + ) + return {item.decode("utf-8") for item in encoded.split(b"\0") if item} + + +def registered_website_remote() -> str: + """Return the sole website endpoint registered for release publication.""" + + websites = tuple( + spec + for spec in load_repository_specs(include_website=True) + if spec.category == "website" + ) + if len(websites) != 1: + raise RuntimeError("release configuration must register one website repository") + return _validated_remote_url(websites[0].remote) + + +def bind_publication_remote( + *, web_root: Path, remote: str, registered_remote: str +) -> FrozenPublicationRemote: + """Bind one local remote name to the exact registered fetch/push URL.""" + + registered = _validated_remote_url(registered_remote) + fetch_urls = _configured_remote_urls(web_root, remote=remote, push=False) + push_urls = _configured_remote_urls(web_root, remote=remote, push=True) + if fetch_urls != (registered,) or push_urls != (registered,): + raise RuntimeError( + "configured website fetch and push URLs do not match the registered release remote" + ) + digest = hashlib.sha256( + json.dumps( + {"fetch": fetch_urls, "push": push_urls}, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("utf-8") + ).hexdigest() + return FrozenPublicationRemote(url=registered, sha256=digest) + + +def verify_remote_branch_head( + *, web_root: Path, remote_url: str, branch: str, expected_commit: str +) -> None: + """Prove the remote branch still identifies the frozen publication parent.""" + + if _GIT_OBJECT.fullmatch(expected_commit) is None: + raise RuntimeError("frozen website commit identity is malformed") + branch_ref = _validated_git_ref( + web_root, f"refs/heads/{branch}", label="website branch" + ) + observed = _remote_ref_identities( + web_root=web_root, + remote_url=remote_url, + refs=(branch_ref,), + ) + if observed != {branch_ref: expected_commit}: + raise RuntimeError( + "website HEAD is not identical to the remote publication branch" + ) + + +def remote_publication_identity( + *, web_root: Path, remote_url: str, branch: str, tag_name: str +) -> dict[str, str]: + """Read branch, annotated-tag object, and peeled commit from the remote.""" + + branch_ref = _validated_git_ref( + web_root, f"refs/heads/{branch}", label="website branch" + ) + tag_ref = _validated_git_ref( + web_root, f"refs/tags/{tag_name}", label="website publication tag" + ) + peeled_ref = f"{tag_ref}^{{}}" + observed = _remote_ref_identities( + web_root=web_root, + remote_url=remote_url, + refs=(branch_ref, tag_ref, peeled_ref), + ) + if set(observed) != {branch_ref, tag_ref, peeled_ref}: + raise RuntimeError( + "remote publication does not expose one annotated tag identity" + ) + return { + "publication_commit_sha": observed[branch_ref], + "publication_tag_object_sha": observed[tag_ref], + "publication_tag_commit_sha": observed[peeled_ref], + } + + +def verify_remote_publication( + *, + web_root: Path, + remote_url: str, + branch: str, + tag_name: str, + expected_commit: str, + expected_tag_object: str, +) -> dict[str, str]: + """Independently prove the exact branch and annotated tag after a push.""" + + if ( + _GIT_OBJECT.fullmatch(expected_commit) is None + or _GIT_OBJECT.fullmatch(expected_tag_object) is None + ): + raise RuntimeError("expected publication identity is malformed") + identity = remote_publication_identity( + web_root=web_root, + remote_url=remote_url, + branch=branch, + tag_name=tag_name, + ) + if identity != { + "publication_commit_sha": expected_commit, + "publication_tag_object_sha": expected_tag_object, + "publication_tag_commit_sha": expected_commit, + }: + raise RuntimeError( + "remote publication identity differs from the verified objects" + ) + return identity + + +def _configured_remote_urls( + web_root: Path, *, remote: str, push: bool +) -> tuple[str, ...]: + arguments = ["remote", "get-url"] + if push: + arguments.append("--push") + arguments.extend(["--all", remote]) + encoded = _git_bytes( + web_root, + *arguments, + max_output_bytes=_MAX_REMOTE_METADATA_BYTES, + ) + try: + values = tuple( + _validated_remote_url(line) + for line in encoded.decode("utf-8", errors="strict").splitlines() + if line + ) + except UnicodeError as exc: + raise RuntimeError("configured website remote is not valid UTF-8") from exc + if not values or len(values) > 16: + raise RuntimeError("configured website remote URL set is not bounded") + return values + + +def _validated_remote_url(value: str) -> str: + if not isinstance(value, str): + raise RuntimeError("website remote URL must be text") + try: + encoded = value.encode("utf-8", errors="strict") + except UnicodeError as exc: + raise RuntimeError("website remote URL is not valid UTF-8") from exc + if ( + not value + or value.startswith("-") + or value != value.strip() + or len(encoded) > _MAX_REMOTE_METADATA_BYTES + or any( + character.isspace() or not character.isprintable() for character in value + ) + ): + raise RuntimeError("website remote URL is not a bounded printable endpoint") + return value + + +def _validated_git_ref(web_root: Path, value: str, *, label: str) -> str: + if len(value.encode("utf-8")) > 512: + raise RuntimeError(f"{label.capitalize()} is not a bounded Git ref") + _git_checked( + web_root, + "check-ref-format", + value, + max_output_bytes=_MAX_REMOTE_METADATA_BYTES, + ) + return value + + +def _remote_ref_identities( + *, web_root: Path, remote_url: str, refs: tuple[str, ...] +) -> dict[str, str]: + endpoint = _validated_remote_url(remote_url) + encoded = _git_bytes( + web_root, + "ls-remote", + endpoint, + *refs, + max_output_bytes=_MAX_REMOTE_METADATA_BYTES, + isolate_config=True, + ) + observed: dict[str, str] = {} + try: + lines = encoded.decode("utf-8", errors="strict").splitlines() + except UnicodeError as exc: + raise RuntimeError("remote Git identities are not valid UTF-8") from exc + if len(lines) > max(16, len(refs) * 2): + raise RuntimeError("remote Git identity response is not bounded") + allowed = set(refs) + for line in lines: + object_id, separator, ref = line.partition("\t") + if ( + separator != "\t" + or ref not in allowed + or ref in observed + or _GIT_OBJECT.fullmatch(object_id) is None + ): + raise RuntimeError("remote Git identity response is ambiguous") + observed[ref] = object_id + return observed def website_dirty(path: Path) -> bool: @@ -766,8 +1785,14 @@ def default_tag_name(payload: object, *, channel: str) -> str | None: return None sequence = payload.get("sequence") release = payload.get("release") - selected_units = release.get("selected_units") if isinstance(release, dict) else None - if isinstance(selected_units, list) and len(selected_units) == 1 and isinstance(selected_units[0], dict): + selected_units = ( + release.get("selected_units") if isinstance(release, dict) else None + ) + if ( + isinstance(selected_units, list) + and len(selected_units) == 1 + and isinstance(selected_units[0], dict) + ): repo = str(selected_units[0].get("repo") or "catalog") version = str(selected_units[0].get("version") or "").removeprefix("v") if version: @@ -778,7 +1803,9 @@ def default_tag_name(payload: object, *, channel: str) -> str | None: def commit_message(payload: object, *, channel: str) -> str: if isinstance(payload, dict): release = payload.get("release") - selected_units = release.get("selected_units") if isinstance(release, dict) else None + selected_units = ( + release.get("selected_units") if isinstance(release, dict) else None + ) if isinstance(selected_units, list) and selected_units: units = ", ".join( f"{item.get('repo')} v{str(item.get('version') or '').removeprefix('v')}" @@ -790,10 +1817,32 @@ def commit_message(payload: object, *, channel: str) -> str: return f"Publish {channel} catalog" -def push_command(web_root: Path, *, remote: str, branch: str | None, tag_name: str | None) -> str: +def push_command( + web_root: Path, *, remote: str, branch: str | None, tag_name: str | None +) -> str: if tag_name: - return shlex.join(["git", "-C", str(web_root), "push", "--atomic", remote, f"HEAD:refs/heads/{branch or ''}", f"refs/tags/{tag_name}"]) - return shlex.join(["git", "-C", str(web_root), "push", remote, f"HEAD:refs/heads/{branch or ''}"]) + return shlex.join( + [ + "git", + "-C", + str(web_root), + "push", + "--atomic", + remote, + f"HEAD:refs/heads/{branch or ''}", + f"refs/tags/{tag_name}", + ] + ) + return shlex.join( + [ + "git", + "-C", + str(web_root), + "push", + remote, + f"HEAD:refs/heads/{branch or ''}", + ] + ) def replace_status(step: CatalogPublishStep, status: str) -> CatalogPublishStep: @@ -807,14 +1856,166 @@ def replace_status(step: CatalogPublishStep, status: str) -> CatalogPublishStep: ) +def _sanitized_git_environment( + *, + private_index: str | None = None, + isolate_config: bool = False, + base: dict[str, str] | None = None, +) -> dict[str, str]: + """Remove caller-controlled Git redirection before invoking Git.""" + + source = base if base is not None else os.environ + environment = { + key: value for key, value in source.items() if not key.startswith("GIT_") + } + environment.update( + { + "GIT_CONFIG_COUNT": "0", + "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_NO_REPLACE_OBJECTS": "1", + "GIT_PAGER": "cat", + "GIT_TERMINAL_PROMPT": "0", + } + ) + if isolate_config: + environment["GIT_CONFIG"] = os.devnull + if private_index is not None: + index_path = Path(private_index) + if not index_path.is_absolute(): + raise RuntimeError("private publication index path must be absolute") + parent = index_path.parent.lstat() + if ( + not stat.S_ISDIR(parent.st_mode) + or parent.st_uid != os.geteuid() + or stat.S_IMODE(parent.st_mode) & 0o077 + ): + raise RuntimeError("private publication index directory is not private") + environment["GIT_INDEX_FILE"] = str(index_path) + return environment + + +def _git_bytes( + path: Path, + *args: str, + input_bytes: bytes | None = None, + environment: dict[str, str] | None = None, + max_output_bytes: int = _MAX_GIT_OUTPUT_BYTES, + isolate_config: bool = False, +) -> bytes: + private_index = environment.get("GIT_INDEX_FILE") if environment else None + process_environment = _sanitized_git_environment( + private_index=private_index, + isolate_config=isolate_config, + base=environment, + ) + result = subprocess.run( + [ + "git", + "-C", + str(path), + "-c", + "core.fsmonitor=false", + "-c", + "core.hooksPath=/dev/null", + *args, + ], + check=False, + input=input_bytes, + env=process_environment, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=120, + ) + if len(result.stdout) > max_output_bytes or len(result.stderr) > max_output_bytes: + raise RuntimeError("Git command output exceeded its safety bound") + if result.returncode != 0: + raise subprocess.CalledProcessError( + result.returncode, + result.args, + output=result.stdout, + stderr=result.stderr, + ) + return result.stdout + + +def _git_checked( + path: Path, + *args: str, + input_bytes: bytes | None = None, + environment: dict[str, str] | None = None, + max_output_bytes: int = _MAX_GIT_OUTPUT_BYTES, + isolate_config: bool = False, +) -> None: + _git_bytes( + path, + *args, + input_bytes=input_bytes, + environment=environment, + max_output_bytes=max_output_bytes, + isolate_config=isolate_config, + ) + + +def _git_text_checked( + path: Path, + *args: str, + input_bytes: bytes | None = None, + environment: dict[str, str] | None = None, + max_output_bytes: int = _MAX_GIT_OUTPUT_BYTES, + isolate_config: bool = False, +) -> str: + try: + return ( + _git_bytes( + path, + *args, + input_bytes=input_bytes, + environment=environment, + max_output_bytes=max_output_bytes, + isolate_config=isolate_config, + ) + .decode("utf-8", errors="strict") + .strip() + ) + except UnicodeError as exc: + raise RuntimeError("Git command output is not valid UTF-8") from exc + + def git_text(path: Path, *args: str) -> str: - result = subprocess.run(["git", *args], cwd=path, check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - return result.stdout.strip() if result.returncode == 0 else "" + try: + return _git_text_checked(path, *args) + except (OSError, RuntimeError, subprocess.SubprocessError, UnicodeError): + return "" def git_success(path: Path, *args: str) -> bool: - return subprocess.run(["git", *args], cwd=path, check=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0 + try: + _git_checked(path, *args) + except (OSError, RuntimeError, subprocess.SubprocessError, UnicodeError): + return False + return True def run_checked(args: list[str], *, cwd: Path) -> None: - subprocess.run(args, cwd=cwd, check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + result = subprocess.run( + args, + cwd=cwd, + env=_sanitized_git_environment(), + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=600, + ) + if ( + len(result.stdout) > _MAX_GIT_OUTPUT_BYTES + or len(result.stderr) > _MAX_GIT_OUTPUT_BYTES + ): + raise RuntimeError("publication command output exceeded its safety bound") + if result.returncode != 0: + raise subprocess.CalledProcessError( + result.returncode, + result.args, + output=result.stdout, + stderr=result.stderr, + ) diff --git a/tools/release/release-catalog.py b/tools/release/release-catalog.py index 2abe8e6..691a02f 100644 --- a/tools/release/release-catalog.py +++ b/tools/release/release-catalog.py @@ -9,7 +9,10 @@ from pathlib import Path from govoplan_release.module_directory import write_module_directory from govoplan_release.model import to_jsonable -from govoplan_release.publisher import publish_catalog_candidate +from govoplan_release.publisher import ( + publication_runtime_trust_issues, + publish_catalog_candidate, +) from govoplan_release.selective_catalog import build_selective_catalog_candidate from govoplan_release.workspace import DEFAULT_WORKSPACE_ROOT @@ -18,11 +21,27 @@ def main() -> int: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) - selective = subparsers.add_parser("selective", help="Build a signed selective channel catalog candidate.") - selective.add_argument("--workspace-root", type=Path, default=DEFAULT_WORKSPACE_ROOT) + selective = subparsers.add_parser( + "selective", help="Build a signed selective channel catalog candidate." + ) + selective.add_argument( + "--workspace-root", type=Path, default=DEFAULT_WORKSPACE_ROOT + ) selective.add_argument("--channel", default="stable") - selective.add_argument("--repo-version", action="append", default=[], metavar="REPO=VERSION", required=True) - selective.add_argument("--catalog-signing-key", action="append", default=[], metavar="KEY_ID=PRIVATE_KEY", required=True) + selective.add_argument( + "--repo-version", + action="append", + default=[], + metavar="REPO=VERSION", + required=True, + ) + selective.add_argument( + "--catalog-signing-key", + action="append", + default=[], + metavar="KEY_ID=PRIVATE_KEY", + required=True, + ) selective.add_argument( "--python-artifact", action="append", @@ -56,41 +75,76 @@ def main() -> int: ), ) selective.add_argument("--public-base-url", default="https://govoplan.add-ideas.de") - selective.add_argument("--repository-base", default="git+ssh://git@git.add-ideas.de/add-ideas") - selective.add_argument("--source-remote", default="origin", help="Configured Git remote containing immutable source tags.") + selective.add_argument( + "--repository-base", default="git+ssh://git@git.add-ideas.de/add-ideas" + ) + selective.add_argument( + "--source-remote", + default="origin", + help="Configured Git remote containing immutable source tags.", + ) selective.add_argument("--expires-days", type=int, default=90) selective.add_argument("--sequence", type=int) selective.add_argument("--skip-public-check", action="store_true") - selective.add_argument("--json", action="store_true", help="Print machine-readable summary JSON.") + selective.add_argument( + "--json", action="store_true", help="Print machine-readable summary JSON." + ) - publish = subparsers.add_parser("publish-candidate", help="Publish a reviewed catalog candidate into the website repo.") + publish = subparsers.add_parser( + "publish-candidate", + help="Publish a reviewed catalog candidate into the website repo.", + ) publish.add_argument("--candidate-dir", type=Path, required=True) publish.add_argument("--workspace-root", type=Path, default=DEFAULT_WORKSPACE_ROOT) publish.add_argument("--web-root", type=Path) publish.add_argument("--channel", default="stable") - publish.add_argument("--apply", action="store_true", help="Copy candidate files into the website repo. Without this, only preview.") + publish.add_argument( + "--apply", + action="store_true", + help="Copy candidate files into the website repo. Without this, only preview.", + ) publish.add_argument("--build-web", action="store_true") publish.add_argument("--commit", action="store_true") - publish.add_argument("--tag", action="store_true", help="Create a website catalog publication tag. Implies --commit.") - publish.add_argument("--push", action="store_true", help="Push the website branch and tag. Implies --commit.") + publish.add_argument( + "--tag", + action="store_true", + help="Create a website catalog publication tag. Implies --commit.", + ) + publish.add_argument( + "--push", + action="store_true", + help="Push the website branch and tag. Implies --commit.", + ) publish.add_argument("--remote", default="origin") - publish.add_argument("--source-remote", default="origin", help="Configured Git remote containing catalog source tags.") + publish.add_argument( + "--source-remote", + default="origin", + help="Configured Git remote containing catalog source tags.", + ) publish.add_argument("--branch") publish.add_argument("--tag-name") publish.add_argument("--npm", default="npm") publish.add_argument("--allow-dirty-website", action="store_true") - publish.add_argument("--json", action="store_true", help="Print machine-readable summary JSON.") + publish.add_argument( + "--json", action="store_true", help="Print machine-readable summary JSON." + ) - directory = subparsers.add_parser("module-directory", help="Generate browsable module-directory files from a catalog and keyring.") + directory = subparsers.add_parser( + "module-directory", + help="Generate browsable module-directory files from a catalog and keyring.", + ) directory.add_argument("--catalog", type=Path, required=True) directory.add_argument("--keyring", type=Path, required=True) directory.add_argument("--output-dir", type=Path, required=True) directory.add_argument("--channel", default="stable") directory.add_argument("--public-base-url", default="https://govoplan.add-ideas.de") - directory.add_argument("--json", action="store_true", help="Print machine-readable summary JSON.") + directory.add_argument( + "--json", action="store_true", help="Print machine-readable summary JSON." + ) args = parser.parse_args() if args.command == "selective": + require_release_runtime_trust() result = build_selective_catalog_candidate( repo_versions=parse_repo_versions(tuple(args.repo_version)), channel=args.channel, @@ -114,6 +168,8 @@ def main() -> int: print_text_summary(payload) return 0 if result.status == "ready" else 1 if args.command == "publish-candidate": + if args.apply: + require_release_runtime_trust() result = publish_catalog_candidate( candidate_dir=args.candidate_dir, workspace_root=args.workspace_root, @@ -138,6 +194,7 @@ def main() -> int: print_publish_summary(payload) return 0 if result.status in {"ready", "applied", "published"} else 1 if args.command == "module-directory": + require_release_runtime_trust() catalog_payload = json.loads(args.catalog.read_text(encoding="utf-8")) keyring_payload = json.loads(args.keyring.read_text(encoding="utf-8")) files = write_module_directory( @@ -147,7 +204,11 @@ def main() -> int: channel=args.channel, public_base_url=args.public_base_url, ) - payload = {"status": "generated", "output_dir": str(args.output_dir), "files": [str(path) for path in files]} + payload = { + "status": "generated", + "output_dir": str(args.output_dir), + "files": [str(path) for path in files], + } if args.json: print(json.dumps(payload, indent=2, sort_keys=True)) else: @@ -162,6 +223,12 @@ def main() -> int: return 2 +def require_release_runtime_trust() -> None: + issues = publication_runtime_trust_issues() + if issues: + raise SystemExit("Release tooling trust gate failed: " + "; ".join(issues)) + + def parse_repo_versions(values: tuple[str, ...]) -> dict[str, str]: result: dict[str, str] = {} for value in values: @@ -185,7 +252,9 @@ def parse_repo_paths(values: tuple[str, ...]) -> dict[str, Path]: repo = repo.strip() path_text = path_text.strip() if not repo or not path_text or repo in result: - raise SystemExit(f"--python-artifact must uniquely use REPO=/path/to/wheel: {value}") + raise SystemExit( + f"--python-artifact must uniquely use REPO=/path/to/wheel: {value}" + ) result[repo] = Path(path_text).expanduser() return result