policy-nexus/tools/verify_release.py

#!/usr/bin/env python3
"""Fail closed unless a built policy site is safe to publish as an OCI release."""

from __future__ import annotations

import argparse
import hashlib
import html
import json
from pathlib import Path, PurePosixPath
import re
from typing import Any


HEX_DIGEST = re.compile(r"^[a-f0-9]{64}$")
CLEAN_GIT_REVISION = re.compile(r"^[a-f0-9]{40}$")
SOURCE_REVISION = re.compile(
    r'<meta name="policy-source-revision" content="([^"]+)">'
)
SOURCE_DIGEST = re.compile(r'<meta name="policy-source-digest" content="([^"]+)">')


def _safe_relative(value: str) -> PurePosixPath:
    path = PurePosixPath(value)
    if path.is_absolute() or ".." in path.parts or not path.parts:
        raise ValueError(f"unsafe release path: {value!r}")
    return path


def _page_provenance(path: Path) -> tuple[str, str]:
    page = path.read_text(encoding="utf-8")
    revision = SOURCE_REVISION.search(page)
    digest = SOURCE_DIGEST.search(page)
    if not revision or not digest:
        raise ValueError(f"{path}: missing source provenance metadata")
    return html.unescape(revision.group(1)), html.unescape(digest.group(1))


def verify(build: Path) -> dict[str, Any]:
    build = build.resolve()
    if not build.is_dir():
        raise ValueError(f"release directory does not exist: {build}")
    for path in build.rglob("*"):
        if path.is_symlink():
            raise ValueError(f"release tree contains a symlink: {path.relative_to(build)}")

    index = build / "index.html"
    manifest_path = build / "publication-manifest.json"
    inventory_path = build / "source-inventory.json"
    if not index.is_file() or not manifest_path.is_file() or not inventory_path.is_file():
        raise ValueError(
            "release requires index.html, publication-manifest.json and source-inventory.json"
        )

    manifest_bytes = manifest_path.read_bytes()
    manifest = json.loads(manifest_bytes)
    if manifest.get("schema_version") != 1:
        raise ValueError("publication manifest schema_version must be 1")
    if not manifest.get("generated_as_of"):
        raise ValueError("publication manifest generated_as_of is required")
    documents = manifest.get("documents")
    if not isinstance(documents, list) or not documents:
        raise ValueError("publication manifest must contain at least one document")

    inventory_bytes = inventory_path.read_bytes()
    inventory = json.loads(inventory_bytes)
    if inventory.get("schema_version") != "policy-nexus-source-inventory/v1":
        raise ValueError("source inventory schema_version is invalid")
    source_set_digest = inventory.get("source_set_digest", "")
    if not HEX_DIGEST.fullmatch(source_set_digest):
        raise ValueError("source inventory requires a valid source_set_digest")
    inventoried_published = {
        (source.get("source_repo"), source.get("source_path"))
        for source in inventory.get("sources", [])
        if source.get("disposition") == "published"
    }

    verified: list[str] = []
    for document in documents:
        document_id = document.get("id", "<unknown>")
        for field in (
            "id",
            "title",
            "status",
            "revision",
            "owner",
            "last_reviewed",
            "review_due",
            "canonical_path",
            "revision_path",
            "source_repo",
            "source_path",
        ):
            if not document.get(field) or document.get(field) == "unknown":
                raise ValueError(f"{document_id}: release metadata field {field} is required")
        source_revision = document.get("source_revision", "")
        source_digest = document.get("source_digest", "")
        if not CLEAN_GIT_REVISION.fullmatch(source_revision):
            raise ValueError(
                f"{document_id}: production source_revision must be a clean 40-hex Git commit; "
                f"got {source_revision!r}"
            )
        if not HEX_DIGEST.fullmatch(source_digest):
            raise ValueError(f"{document_id}: invalid source_digest {source_digest!r}")
        source_key = (document["source_repo"], document["source_path"])
        if source_key not in inventoried_published:
            raise ValueError(f"{document_id}: source is not published in source inventory")

        canonical = build / _safe_relative(document["canonical_path"])
        revision = build / _safe_relative(document["revision_path"])
        if not canonical.is_file() or not revision.is_file():
            raise ValueError(f"{document_id}: canonical or immutable revision page is missing")

        current_revision, current_digest = _page_provenance(canonical)
        immutable_revision, immutable_digest = _page_provenance(revision)
        if (current_revision, current_digest) != (source_revision, source_digest):
            raise ValueError(f"{document_id}: canonical page provenance differs from manifest")
        if immutable_digest != source_digest:
            raise ValueError(f"{document_id}: immutable revision digest differs from manifest")
        if not CLEAN_GIT_REVISION.fullmatch(immutable_revision):
            raise ValueError(
                f"{document_id}: immutable revision page was not built from a clean Git commit"
            )
        verified.append(document_id)

    return {
        "schema_version": "policy-nexus-release/v1",
        "publication_manifest_digest": hashlib.sha256(manifest_bytes).hexdigest(),
        "source_inventory_digest": hashlib.sha256(inventory_bytes).hexdigest(),
        "source_set_digest": source_set_digest,
        "generated_as_of": manifest["generated_as_of"],
        "documents": verified,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("build", nargs="?", type=Path, default=Path("build"))
    args = parser.parse_args()
    try:
        evidence = verify(args.build)
    except (KeyError, OSError, ValueError, json.JSONDecodeError) as exc:
        parser.error(str(exc))
    print(json.dumps(evidence, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())