diff --git a/tests/test_release_catalog_publication.py b/tests/test_release_catalog_publication.py index 70ac419..12fa3b0 100644 --- a/tests/test_release_catalog_publication.py +++ b/tests/test_release_catalog_publication.py @@ -2,6 +2,7 @@ from __future__ import annotations import json from pathlib import Path +import subprocess import sys import tempfile import unittest @@ -13,10 +14,204 @@ 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 publish_catalog_candidate # noqa: E402 +from govoplan_release.publisher import ( # noqa: E402 + publish_catalog_candidate, + verify_committed_publication, +) class ReleaseCatalogPublicationTests(unittest.TestCase): + 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_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"))) + (web_root / ".git").mkdir() + + 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), + ): + 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) @@ -94,6 +289,17 @@ class ReleaseCatalogPublicationTests(unittest.TestCase): ] } + @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() diff --git a/tests/test_release_module_directory.py b/tests/test_release_module_directory.py new file mode 100644 index 0000000..f4d0b58 --- /dev/null +++ b/tests/test_release_module_directory.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from pathlib import Path +import sys +import unittest +from unittest import mock + + +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.module_directory import ( # noqa: E402 + module_directory_payloads, + safe_path_part, +) + + +class ReleaseModuleDirectoryTests(unittest.TestCase): + def test_dot_segments_and_noncanonical_ids_never_become_paths(self) -> None: + for value in (".", "..", "../escape", "/absolute", "module/child"): + with self.subTest(value=value), self.assertRaises(ValueError): + safe_path_part(value) + + for module_id in ("..", "../escape", "UPPER", "bad.id"): + with self.subTest(module_id=module_id), mock.patch( + "govoplan_release.module_directory.module_rows", + return_value=[ + { + "module_id": module_id, + "version": "1.2.3", + } + ], + ): + with self.assertRaises(ValueError): + module_directory_payloads( + catalog_payload={}, + keyring_payload={}, + channel="stable", + ) + + def test_noncanonical_version_and_channel_fail_before_paths(self) -> None: + with mock.patch( + "govoplan_release.module_directory.module_rows", + return_value=[{"module_id": "demo", "version": "../1.2.3"}], + ): + with self.assertRaises(ValueError): + module_directory_payloads( + catalog_payload={}, keyring_payload={}, channel="stable" + ) + with self.assertRaises(ValueError): + module_directory_payloads( + catalog_payload={}, keyring_payload={}, channel="../stable" + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/release/govoplan_release/artifact_identity.py b/tools/release/govoplan_release/artifact_identity.py new file mode 100644 index 0000000..6fdd30a --- /dev/null +++ b/tools/release/govoplan_release/artifact_identity.py @@ -0,0 +1,397 @@ +"""Independent identities for immutable Python wheel release artifacts.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +from email.parser import BytesParser +from email.policy import compat32 +import hashlib +import json +import os +from pathlib import Path, PurePosixPath +import re +import stat +import zipfile + + +WHEEL_PAYLOAD_ALGORITHM = "govoplan-wheel-declared-payload-v1" +INSTALLED_PAYLOAD_ALGORITHM = "govoplan-installed-record-payload-v1" +MAX_WHEEL_BYTES = 512 * 1024 * 1024 +MAX_WHEEL_ENTRIES = 10_000 +MAX_WHEEL_UNCOMPRESSED_BYTES = 1024 * 1024 * 1024 +MAX_WHEEL_ENTRY_BYTES = 64 * 1024 * 1024 +MAX_METADATA_BYTES = 1024 * 1024 +_PACKAGE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") +_VERSION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+!-]{0,127}$") +_PYTHON_REF = re.compile( + r"/(?Pgovoplan-[a-z0-9-]+)[.]git@v(?P[^\s;]+)$" +) + + +class ArtifactIdentityError(ValueError): + """A built artifact cannot provide a bounded immutable identity.""" + + +@dataclass(frozen=True, slots=True) +class PythonWheelIdentity: + package_name: str + package_version: str + archive_sha256: str + archive_size: int + payload_sha256: str + payload_file_count: int + requires_installer_receipt: bool + + def catalog_payload(self) -> dict[str, object]: + return { + "artifact_kind": "python-wheel", + "package_name": self.package_name, + "package_version": self.package_version, + "archive_sha256": self.archive_sha256, + "archive_size": self.archive_size, + "installed_payload": { + "algorithm": WHEEL_PAYLOAD_ALGORITHM, + "sha256": self.payload_sha256, + "file_count": self.payload_file_count, + }, + "requires_installer_receipt": self.requires_installer_receipt, + } + + +def inspect_python_wheel(path: Path | str) -> PythonWheelIdentity: + """Hash a regular built wheel and derive its install-stable payload identity.""" + + wheel = Path(path).expanduser() + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(wheel, flags) + except OSError as exc: + raise ArtifactIdentityError("built artifact cannot be opened safely") from exc + try: + archive_sha256, archive_size, initial = _hash_open_artifact(descriptor) + os.lseek(descriptor, 0, os.SEEK_SET) + try: + with os.fdopen(os.dup(descriptor), "rb") as artifact_handle: + with zipfile.ZipFile(artifact_handle) as archive: + ( + package_name, + package_version, + payload_rows, + requires_receipt, + ) = _inspect_wheel_archive(archive) + except (OSError, zipfile.BadZipFile, RuntimeError) as exc: + raise ArtifactIdentityError( + "built Python artifact is not a readable wheel" + ) from exc + final = os.fstat(descriptor) + if _stat_fingerprint(final) != _stat_fingerprint(initial): + raise ArtifactIdentityError("built artifact changed while it was inspected") + finally: + os.close(descriptor) + + return PythonWheelIdentity( + package_name=package_name, + package_version=package_version, + archive_sha256=archive_sha256, + archive_size=archive_size, + payload_sha256=payload_identity_sha256( + algorithm=WHEEL_PAYLOAD_ALGORITHM, rows=payload_rows + ), + payload_file_count=len(payload_rows), + requires_installer_receipt=requires_receipt, + ) + + +def _inspect_wheel_archive( + archive: zipfile.ZipFile, +) -> tuple[str, str, list[dict[str, object]], bool]: + infos = archive.infolist() + if not infos or len(infos) > MAX_WHEEL_ENTRIES: + raise ArtifactIdentityError("wheel entry count is empty or exceeds its limit") + _validate_members(infos) + metadata_members = [ + item + for item in infos + if not item.is_dir() + and PurePosixPath(item.filename).name == "METADATA" + and PurePosixPath(item.filename).parent.name.endswith(".dist-info") + ] + if len(metadata_members) != 1: + raise ArtifactIdentityError( + "wheel must contain exactly one dist-info METADATA file" + ) + metadata_member = metadata_members[0] + dist_info = PurePosixPath(metadata_member.filename).parent + metadata_bytes = _read_member( + archive, metadata_member, max_bytes=MAX_METADATA_BYTES + ) + parsed = BytesParser(policy=compat32).parsebytes(metadata_bytes) + package_name = normalize_package_name(str(parsed.get("Name") or "")) + package_version = str(parsed.get("Version") or "").strip() + if _PACKAGE.fullmatch(package_name) is None: + raise ArtifactIdentityError("wheel METADATA has an invalid package name") + if _VERSION.fullmatch(package_version) is None: + raise ArtifactIdentityError("wheel METADATA has an invalid package version") + + payload_rows: list[dict[str, object]] = [] + requires_receipt = False + for info in infos: + if info.is_dir(): + continue + normalized, transformed = _wheel_payload_path( + PurePosixPath(info.filename), dist_info=dist_info + ) + requires_receipt = requires_receipt or transformed + if normalized is None: + continue + encoded = _read_member(archive, info, max_bytes=MAX_WHEEL_ENTRY_BYTES) + payload_rows.append( + { + "path": normalized, + "sha256": hashlib.sha256(encoded).hexdigest(), + "size": len(encoded), + } + ) + payload_rows.sort(key=lambda item: str(item["path"])) + if not payload_rows: + raise ArtifactIdentityError("wheel has no immutable payload entries") + if len({str(item["path"]) for item in payload_rows}) != len(payload_rows): + raise ArtifactIdentityError( + "wheel payload paths are ambiguous after installation mapping" + ) + entry_points = f"{dist_info.as_posix()}/entry_points.txt" + if any(item.filename == entry_points for item in infos): + entry_point_info = next(item for item in infos if item.filename == entry_points) + entry_point_text = _read_member( + archive, entry_point_info, max_bytes=MAX_METADATA_BYTES + ).decode("utf-8", errors="replace") + if re.search(r"(?m)^\s*\[(?:console|gui)_scripts\]\s*$", entry_point_text): + requires_receipt = True + return package_name, package_version, payload_rows, requires_receipt + + +def selected_artifact_identity_issues(payload: object) -> tuple[str, ...]: + """Report selected Python releases without a matching built-wheel identity.""" + + if not isinstance(payload, dict): + return ("candidate catalog must be a JSON object",) + release = payload.get("release") + selected_units = release.get("selected_units") if isinstance(release, dict) else None + artifacts = release.get("artifacts") if isinstance(release, dict) else None + if not isinstance(selected_units, list) or not selected_units: + return ("release.selected_units is missing or empty",) + entries: list[object] = [payload.get("core_release")] + modules = payload.get("modules") + if isinstance(modules, list): + entries.extend(modules) + package_by_repo: dict[str, tuple[str, str]] = {} + for entry in entries: + if not isinstance(entry, dict): + continue + python_ref = entry.get("python_ref") + match = _PYTHON_REF.search(python_ref) if isinstance(python_ref, str) else None + package_name = entry.get("python_package") + version = entry.get("version") + if ( + match is not None + and isinstance(package_name, str) + and _PACKAGE.fullmatch(package_name) is not None + and isinstance(version, str) + ): + package_by_repo[match.group("repo")] = (package_name, version) + artifacts_by_package: dict[str, dict[str, object]] = {} + malformed_artifacts = False + if not isinstance(artifacts, list): + artifacts = [] + for artifact in artifacts: + if not isinstance(artifact, dict): + malformed_artifacts = True + continue + package_name = artifact.get("package_name") + if not isinstance(package_name, str) or package_name in artifacts_by_package: + malformed_artifacts = True + continue + if not _catalog_artifact_shape_valid(artifact): + malformed_artifacts = True + continue + artifacts_by_package[package_name] = artifact + issues: list[str] = [] + if malformed_artifacts: + issues.append("release.artifacts contains malformed or duplicate identities") + seen_repos: set[str] = set() + for unit in selected_units: + if not isinstance(unit, dict): + issues.append("release.selected_units contains a malformed entry") + continue + repo = unit.get("repo") + if not isinstance(repo, str) or repo in seen_repos: + issues.append("release.selected_units contains a duplicate or invalid repository") + continue + seen_repos.add(repo) + expected = package_by_repo.get(repo) + if expected is None: + continue # Selected non-Python repositories have no wheel identity. + package_name, version = expected + artifact = artifacts_by_package.get(package_name) + if artifact is None: + issues.append( + f"selected Python repository {repo} has no built artifact identity for {package_name} {version}" + ) + elif artifact.get("package_version") != version: + issues.append( + f"selected Python repository {repo} artifact identity has version {artifact.get('package_version')!r}, expected {version!r}" + ) + return tuple(issues) + + +def _catalog_artifact_shape_valid(value: dict[str, object]) -> bool: + if set(value) != { + "artifact_kind", + "package_name", + "package_version", + "archive_sha256", + "archive_size", + "installed_payload", + "requires_installer_receipt", + }: + return False + package_name = value.get("package_name") + version = value.get("package_version") + archive_sha256 = value.get("archive_sha256") + archive_size = value.get("archive_size") + installed_payload = value.get("installed_payload") + return ( + value.get("artifact_kind") == "python-wheel" + and isinstance(package_name, str) + and _PACKAGE.fullmatch(package_name) is not None + and isinstance(version, str) + and _VERSION.fullmatch(version) is not None + and isinstance(archive_sha256, str) + and re.fullmatch(r"[0-9a-f]{64}", archive_sha256) is not None + and isinstance(archive_size, int) + and not isinstance(archive_size, bool) + and 0 < archive_size <= MAX_WHEEL_BYTES + and isinstance(value.get("requires_installer_receipt"), bool) + and isinstance(installed_payload, dict) + and set(installed_payload) == {"algorithm", "sha256", "file_count"} + and installed_payload.get("algorithm") == WHEEL_PAYLOAD_ALGORITHM + and isinstance(installed_payload.get("sha256"), str) + and re.fullmatch(r"[0-9a-f]{64}", str(installed_payload["sha256"])) + is not None + and isinstance(installed_payload.get("file_count"), int) + and not isinstance(installed_payload.get("file_count"), bool) + and 0 < int(installed_payload["file_count"]) <= MAX_WHEEL_ENTRIES + ) + + +def payload_identity_sha256( + *, algorithm: str, rows: list[dict[str, object]] +) -> str: + payload = {"algorithm": algorithm, "files": rows} + encoded = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def record_hash_hex(value: str) -> str | None: + try: + decoded = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + except (ValueError, TypeError): + return None + return decoded.hex() if len(decoded) == hashlib.sha256().digest_size else None + + +def normalize_package_name(value: str) -> str: + return re.sub(r"[-_.]+", "-", value.strip().lower()) + + +def _hash_open_artifact(descriptor: int) -> tuple[str, int, os.stat_result]: + digest = hashlib.sha256() + total = 0 + initial = os.fstat(descriptor) + if not stat.S_ISREG(initial.st_mode) or initial.st_size > MAX_WHEEL_BYTES: + raise ArtifactIdentityError("built artifact must be a bounded regular file") + while total < initial.st_size: + chunk = os.read(descriptor, min(1024 * 1024, initial.st_size - total)) + if not chunk: + break + total += len(chunk) + digest.update(chunk) + final = os.fstat(descriptor) + if total != initial.st_size or _stat_fingerprint(final) != _stat_fingerprint(initial): + raise ArtifactIdentityError("built artifact changed while it was hashed") + return digest.hexdigest(), total, initial + + +def _stat_fingerprint(value: os.stat_result) -> tuple[int, int, int, int, int]: + return ( + value.st_dev, + value.st_ino, + value.st_size, + value.st_mtime_ns, + value.st_ctime_ns, + ) + + +def _validate_members(infos: list[zipfile.ZipInfo]) -> None: + names: set[str] = set() + total = 0 + for info in infos: + path = PurePosixPath(info.filename.replace("\\", "/")) + if ( + not info.filename + or path.is_absolute() + or ".." in path.parts + or any(ord(character) < 32 for character in info.filename) + or info.flag_bits & 0x1 + ): + raise ArtifactIdentityError("wheel contains an unsafe member") + normalized = path.as_posix() + if normalized in names: + raise ArtifactIdentityError("wheel contains duplicate members") + names.add(normalized) + if _zip_member_is_symlink(info): + raise ArtifactIdentityError("wheel contains a symbolic-link member") + total += info.file_size + if info.file_size > MAX_WHEEL_ENTRY_BYTES or total > MAX_WHEEL_UNCOMPRESSED_BYTES: + raise ArtifactIdentityError("wheel uncompressed payload exceeds its limit") + + +def _zip_member_is_symlink(info: zipfile.ZipInfo) -> bool: + return stat.S_ISLNK((info.external_attr >> 16) & 0o177777) + + +def _read_member( + archive: zipfile.ZipFile, info: zipfile.ZipInfo, *, max_bytes: int +) -> bytes: + if info.file_size > max_bytes: + raise ArtifactIdentityError("wheel member exceeds its size limit") + with archive.open(info, "r") as handle: + encoded = handle.read(max_bytes + 1) + if len(encoded) > max_bytes or len(encoded) != info.file_size: + raise ArtifactIdentityError("wheel member size is inconsistent") + return encoded + + +def _wheel_payload_path( + path: PurePosixPath, *, dist_info: PurePosixPath +) -> tuple[str | None, bool]: + if path.parent == dist_info and path.name in {"RECORD", "RECORD.jws", "RECORD.p7s"}: + return None, False + parts = path.parts + data_prefix = dist_info.name.removesuffix(".dist-info") + ".data" + if parts and parts[0] == data_prefix: + if len(parts) < 3: + raise ArtifactIdentityError("wheel data member has no installation target") + scheme = parts[1] + remainder = PurePosixPath(*parts[2:]).as_posix() + if scheme in {"purelib", "platlib"}: + return remainder, False + if scheme == "data": + return f"@data/{remainder}", False + if scheme in {"scripts", "headers"}: + return None, True + raise ArtifactIdentityError("wheel uses an unsupported installation scheme") + return path.as_posix(), False diff --git a/tools/release/govoplan_release/module_directory.py b/tools/release/govoplan_release/module_directory.py index 596e3fa..7c0e356 100644 --- a/tools/release/govoplan_release/module_directory.py +++ b/tools/release/govoplan_release/module_directory.py @@ -9,6 +9,7 @@ import re from typing import Any from .catalog import DEFAULT_PUBLIC_BASE_URL, canonical_hash +from .candidate_artifact import validate_release_channel from .release_intelligence import compatibility_rows, module_rows, signature_rows @@ -42,6 +43,7 @@ def module_directory_payloads( channel: str, public_base_url: str = DEFAULT_PUBLIC_BASE_URL, ) -> tuple[tuple[Path, dict[str, Any]], ...]: + channel = validate_release_channel(channel) generated_at = json_datetime(datetime.now(tz=UTC)) modules = module_rows(catalog_payload) compatibility = compatibility_rows(modules) @@ -59,8 +61,8 @@ def module_directory_payloads( if not module_id: continue version = str(module.get("version") or "0.0.0") - module_slug = safe_path_part(module_id) - version_slug = safe_path_part(version) + module_slug = safe_module_id(module_id) + version_slug = safe_version(version) module_base = f"{base_url}/catalogs/v1/modules/{module_slug}" manifest_url = f"{module_base}/{version_slug}/manifest.json" module_index_url = f"{module_base}/index.json" @@ -139,7 +141,25 @@ def module_directory_payloads( def safe_path_part(value: str) -> str: - return re.sub(r"[^A-Za-z0-9_.-]+", "_", value.strip()) or "_" + if ( + not isinstance(value, str) + or value in {".", ".."} + or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.+!-]{0,127}", value) is None + ): + raise ValueError("module-directory path part is not canonical") + return value + + +def safe_module_id(value: str) -> str: + if re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", value) is None: + raise ValueError("module ID is not canonical") + return safe_path_part(value) + + +def safe_version(value: str) -> str: + if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+!-]{0,127}", value) is None: + raise ValueError("module version is not canonical") + return safe_path_part(value) def json_datetime(value: datetime) -> str: diff --git a/tools/release/govoplan_release/publisher.py b/tools/release/govoplan_release/publisher.py index c0a4ffc..9e55d7c 100644 --- a/tools/release/govoplan_release/publisher.py +++ b/tools/release/govoplan_release/publisher.py @@ -4,16 +4,23 @@ from __future__ import annotations from datetime import UTC, datetime import json +import os from pathlib import Path +import re +import secrets import shlex import shutil +import stat import subprocess +import tempfile from govoplan_core.core.module_package_catalog import validate_module_package_catalog from .catalog import canonical_hash +from .artifact_identity import selected_artifact_identity_issues +from .candidate_artifact import validate_release_channel from .model import CatalogPublishResult, CatalogPublishStep -from .module_directory import write_module_directory +from .module_directory import module_directory_payloads from .selective_catalog import trusted_keys_from_keyring from .source_provenance import catalog_source_selection, source_tag_provenance_issues from .version_alignment import candidate_catalog_version_issues @@ -38,6 +45,7 @@ def publish_catalog_candidate( tag_name: str | None = None, allow_dirty_website: bool = False, ) -> CatalogPublishResult: + channel = validate_release_channel(channel) workspace = resolve_workspace_root(workspace_root or DEFAULT_WORKSPACE_ROOT) candidate_root = Path(candidate_dir).expanduser() if not candidate_root.is_absolute(): @@ -75,6 +83,8 @@ def publish_catalog_candidate( 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") version_issues = candidate_catalog_version_issues(candidate_payload) blockers.extend( f"candidate version alignment: {issue.source}: {issue.actual!r}; " @@ -86,6 +96,16 @@ def publish_catalog_candidate( f"source provenance: {issue.describe()}" for issue in source_selection.issues ) + artifact_issues = selected_artifact_identity_issues(candidate_payload) + if apply or commit or tag or push: + blockers.extend( + f"release artifact identity: {issue}" for issue in artifact_issues + ) + else: + notes.extend( + f"Source-only preview; publication would block: {issue}" + for issue in artifact_issues + ) source_issues = source_tag_provenance_issues( repo_versions=source_selection.all_versions, workspace=workspace, @@ -117,8 +137,8 @@ def publish_catalog_candidate( blockers.append( "candidate keyring differs from the publication trust anchor without a matching signed keyring_sha256" ) - validation = validate_module_package_catalog( - candidate_catalog, + validation = validate_catalog_payload( + candidate_payload, require_trusted=True, approved_channels=(channel,), trusted_keys=publication_trust, @@ -264,23 +284,42 @@ def publish_catalog_candidate( notes=tuple(notes), ) - target_catalog.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(candidate_catalog, target_catalog) - shutil.copy2(candidate_keyring, target_keyring) - if target_modules.exists(): - shutil.rmtree(target_modules) - write_module_directory( + directory_payloads = module_directory_payloads( catalog_payload=candidate_payload, keyring_payload=keyring_payload, - output_root=target_catalog.parent.parent, channel=channel, public_base_url=catalog_public_base_url(candidate_payload), ) + _atomic_write_public_json( + target_catalog, candidate_payload, trusted_root=resolved_web_root + ) + _atomic_write_public_json( + target_keyring, keyring_payload, trusted_root=resolved_web_root + ) + if target_modules.exists(): + _remove_public_tree(target_modules, trusted_root=resolved_web_root) + for relative_path, payload in directory_payloads: + _atomic_write_public_json( + target_catalog.parent.parent / relative_path, + payload, + trusted_root=resolved_web_root, + ) + + expected_blobs = _expected_publication_blobs( + web_root=resolved_web_root, + target_catalog=target_catalog, + candidate_payload=candidate_payload, + target_keyring=target_keyring, + keyring_payload=keyring_payload, + module_root=target_catalog.parent.parent, + directory_payloads=directory_payloads, + ) 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] + 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(): @@ -292,15 +331,37 @@ def publish_catalog_candidate( 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] - if tag and effective_tag_name: - run_checked(["git", "-C", str(resolved_web_root), "tag", "-a", effective_tag_name, "-m", commit_message(candidate_payload, channel=channel)], cwd=resolved_web_root) + publication_commit = committed_publication_sha(resolved_web_root) + verify_committed_publication( + web_root=resolved_web_root, + commit_sha=publication_commit, + expected_blobs=expected_blobs, + module_root=target_modules, + ) + publication_tag_object: 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}" + ) + tagged_commit = git_text( + resolved_web_root, + "rev-parse", + "--verify", + f"refs/tags/{effective_tag_name}^{{commit}}", + ) + if ( + re.fullmatch(r"[0-9a-f]{40,64}", publication_tag_object) is None + or tagged_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] - if push and effective_branch: + if push and effective_branch and publication_commit: push_args = ["git", "-C", str(resolved_web_root), "push"] - if tag and effective_tag_name: - push_args.extend(["--atomic", remote, f"HEAD:refs/heads/{effective_branch}", f"refs/tags/{effective_tag_name}"]) + 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}"]) else: - push_args.extend([remote, f"HEAD:refs/heads/{effective_branch}"]) + 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] @@ -394,6 +455,300 @@ def read_json(path: Path) -> object: return json.loads(path.read_text(encoding="utf-8")) +def validate_catalog_payload( + payload: object, + *, + require_trusted: bool, + approved_channels: tuple[str, ...], + trusted_keys: dict[str, bytes], +) -> dict[str, object]: + """Validate exactly the in-memory object that publication will consume.""" + + encoded = _public_json_bytes(payload) + descriptor, temporary_name = tempfile.mkstemp( + prefix="govoplan-catalog-validation-", suffix=".json" + ) + temporary = Path(temporary_name) + try: + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, "wb", closefd=True) as handle: + handle.write(encoded) + handle.flush() + os.fsync(handle.fileno()) + return validate_module_package_catalog( + temporary, + require_trusted=require_trusted, + approved_channels=approved_channels, + trusted_keys=trusted_keys, + ) + finally: + try: + os.close(descriptor) + except OSError: + pass + temporary.unlink(missing_ok=True) + + +def _public_json_bytes(payload: object) -> bytes: + encoded = ( + json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=True) + "\n" + ).encode("utf-8") + if len(encoded) > 16 * 1024 * 1024: + raise ValueError("published catalog object exceeds its size limit") + return encoded + + +def _atomic_write_public_json( + path: Path, payload: object, *, trusted_root: Path +) -> None: + """Write through pinned directories from a trusted worktree root.""" + + encoded = _public_json_bytes(payload) + relative = _trusted_relative_path(path, trusted_root=trusted_root) + if len(relative.parts) < 2: + raise ValueError("published catalog target requires a parent directory") + parent_descriptor = _open_relative_directory( + trusted_root, relative.parts[:-1], create=True + ) + target_name = relative.parts[-1] + try: + try: + observed = os.stat( + target_name, dir_fd=parent_descriptor, follow_symlinks=False + ) + except FileNotFoundError: + observed = None + 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 + ) + replaced = False + try: + with os.fdopen(descriptor, "wb", closefd=True) as handle: + handle.write(encoded) + handle.flush() + os.fsync(handle.fileno()) + os.replace( + temporary_name, + target_name, + src_dir_fd=parent_descriptor, + dst_dir_fd=parent_descriptor, + ) + replaced = True + published = os.open( + target_name, + os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), + dir_fd=parent_descriptor, + ) + try: + if not stat.S_ISREG(os.fstat(published).st_mode): + raise ValueError("published catalog target must remain regular") + os.fchmod(published, 0o644) + finally: + os.close(published) + os.fsync(parent_descriptor) + finally: + if not replaced: + try: + os.close(descriptor) + except OSError: + pass + try: + os.unlink(temporary_name, dir_fd=parent_descriptor) + except FileNotFoundError: + pass + finally: + os.close(parent_descriptor) + + +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 + if not relative.parts or any(part in {"", ".", ".."} for part in relative.parts): + raise ValueError("published catalog target is not canonical") + return relative + + +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 + ) + descriptor = os.open(trusted_root, root_flags) + try: + if not stat.S_ISDIR(os.fstat(descriptor).st_mode): + raise ValueError("publication worktree root must be a directory") + for part in parts: + if part in {"", ".", ".."} or "/" in part: + raise ValueError("published catalog path is not canonical") + try: + child = os.open(part, root_flags, dir_fd=descriptor) + except FileNotFoundError: + if not create: + raise + os.mkdir(part, mode=0o755, dir_fd=descriptor) + child = os.open(part, root_flags, dir_fd=descriptor) + os.close(descriptor) + descriptor = child + return descriptor + except BaseException: + os.close(descriptor) + 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") + 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) + + +def _expected_publication_blobs( + *, + web_root: Path, + target_catalog: Path, + candidate_payload: object, + target_keyring: Path, + keyring_payload: object, + module_root: Path, + directory_payloads: tuple[tuple[Path, dict[str, object]], ...], +) -> dict[str, bytes]: + expected = { + _trusted_relative_path( + target_catalog, trusted_root=web_root + ).as_posix(): _public_json_bytes(candidate_payload), + _trusted_relative_path( + target_keyring, trusted_root=web_root + ).as_posix(): _public_json_bytes(keyring_payload), + } + for relative_path, payload in directory_payloads: + target = module_root / relative_path + repository_path = _trusted_relative_path( + target, trusted_root=web_root + ).as_posix() + if repository_path in expected: + raise ValueError("publication output paths are ambiguous") + expected[repository_path] = _public_json_bytes(payload) + return expected + + +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: + raise RuntimeError("could not resolve the publication commit") + return commit_sha + + +def verify_committed_publication( + *, + web_root: Path, + commit_sha: str, + expected_blobs: dict[str, bytes], + module_root: Path, +) -> None: + """Require the exact validated objects in one immutable commit tree.""" + + if re.fullmatch(r"[0-9a-f]{40,64}", commit_sha) is None: + raise RuntimeError("publication commit 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}/") + } + 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 + } + 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 + 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[0] != b"100644" + or fields[1] != b"blob" + or re.fullmatch(rb"[0-9a-f]{40,64}", fields[2]) is None + ): + 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 + 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 file_json_hash(path: Path) -> str | None: if not path.exists(): return None