Harden release artifact publication

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()

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()

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"/(?P<repo>govoplan-[a-z0-9-]+)[.]git@v(?P<version>[^\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

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)
+        module_slug = safe_module_id(module_id)
-        version_slug = safe_path_part(version)
+        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:

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(
+        validation = validate_catalog_payload(
-            candidate_catalog,
+            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)
+    directory_payloads = module_directory_payloads(
-    shutil.copy2(candidate_catalog, target_catalog)
-    shutil.copy2(candidate_keyring, target_keyring)
-    if target_modules.exists():
-        shutil.rmtree(target_modules)
-    write_module_directory(
         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:
+        publication_commit = committed_publication_sha(resolved_web_root)
-        run_checked(["git", "-C", str(resolved_web_root), "tag", "-a", effective_tag_name, "-m", commit_message(candidate_payload, channel=channel)], cwd=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:
+        if tag and effective_tag_name and publication_tag_object:
-            push_args.extend(["--atomic", remote, f"HEAD:refs/heads/{effective_branch}", f"refs/tags/{effective_tag_name}"])
+            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