govoplan/tools/release/govoplan_release/full_catalog.py

"""Build a private full-profile candidate from exact verified registry bytes."""

from __future__ import annotations

import base64
from datetime import UTC, datetime, timedelta
from functools import lru_cache
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import runpy
import stat
import tarfile
from typing import Any
from urllib.parse import quote

from .artifact_identity import inspect_python_wheel, selected_artifact_identity_issues
from .candidate_artifact import (
    ensure_private_candidate_root, harden_private_candidate_tree,
    validate_release_channel,
)
from .catalog import canonical_hash
from .module_directory import write_module_directory
from .selective_catalog import (
    authenticate_base_catalog_signatures, authenticate_base_keyring,
    configured_signer_public_keys, enforce_selected_version_alignment,
    next_sequence, parse_signing_key, read_bounded_json_source, signature,
    validate_catalog_object,
)
from .source_provenance import (
    registered_source_origin_issues, selected_source_provenance,
    source_tag_provenance_issues,
)
from .version_alignment import candidate_catalog_version_issues
from .workspace import META_ROOT, resolve_workspace_root, website_root


MAX_ARTIFACT_BYTES = 512 * 1024 * 1024


@lru_cache(maxsize=3)
def _tool(name: str) -> dict[str, Any]:
    # These are fixed, operator-controlled release modules, not caller paths.
    if name not in {
        "generate-release-package-set", "generate-release-catalog",
        "resolve-package-artifacts",
    }:
        raise ValueError("unknown registry release tool")
    return runpy.run_path(str(META_ROOT / "tools" / "release" / f"{name}.py"))


def authenticate_full_rebuild_base(
    *, web_root: Path, channel: str, signer_public_keys: dict[str, str],
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Authenticate the fixed website pair without reusing legacy entry data.

    Only this full rebuild may migrate a legacy catalog without a keyring hash.
    In that case *every* active website key must exactly equal a configured
    signer; injecting an additional website key cannot extend trust. Selective
    candidates retain their stricter existing hash-pinned-base requirement.
    """

    root = web_root / "public" / "catalogs" / "v1"
    catalog = read_bounded_json_source(root / "channels" / f"{channel}.json", label="published base catalog")
    keyring = read_bounded_json_source(root / "keyring.json", label="published website keyring")
    if not isinstance(catalog, dict) or not isinstance(keyring, dict):
        raise ValueError("published website catalog/keyring must be objects")
    trusted_keys = authenticate_base_keyring(keyring)
    release = catalog.get("release")
    pinned = release.get("keyring_sha256") if isinstance(release, dict) else None
    if pinned is None:
        if trusted_keys != signer_public_keys or len(keyring["keys"]) != len(trusted_keys):
            raise ValueError("legacy full rebuild requires exactly the configured known website signers")
    elif pinned != canonical_hash(keyring):
        raise ValueError("published base catalog does not pin its exact website keyring")
    if any(trusted_keys.get(key) != value for key, value in signer_public_keys.items()):
        raise ValueError("full rebuild cannot introduce or replace a website signer")
    authenticate_base_catalog_signatures(
        catalog, base_trusted_keys=trusted_keys, configured_signers=signer_public_keys,
    )
    validation = validate_catalog_object(
        catalog, approved_channel=channel, signer_public_keys=signer_public_keys,
    )
    if validation.get("valid") is not True:
        raise ValueError(f"published base catalog failed validation: {validation.get('error')}")
    return catalog, keyring


def build_full_registry_candidate(
    *, package_set_path: Path, package_lock_path: Path,
    wheelhouse: Path, webui_packages: Path, output_dir: Path,
    selected_repositories: tuple[str, ...], signing_keys: tuple[str, ...],
    workspace_root: Path | str | None = None, channel: str = "stable",
    source_remote: str = "origin", public_base_url: str = "https://govoplan.add-ideas.de",
    expires_days: int = 90, sequence: int | None = None,
) -> dict[str, object]:
    channel = validate_release_channel(channel)
    if not isinstance(expires_days, int) or isinstance(expires_days, bool) or not 1 <= expires_days <= 365:
        raise ValueError("catalog expiry must be between 1 and 365 days")
    workspace = resolve_workspace_root(workspace_root)
    ensure_private_candidate_root(workspace)
    output = output_dir.expanduser().absolute()
    ensure_private_candidate_root(output.parent, create=True)
    if output.exists() or output.is_symlink():
        raise ValueError("full candidate output must not already exist")
    parsed_keys = tuple(parse_signing_key(value) for value in signing_keys)
    if not parsed_keys:
        raise ValueError("full candidate needs a configured signing key")
    signer_keys = configured_signer_public_keys(parsed_keys)
    base, keyring = authenticate_full_rebuild_base(
        web_root=website_root(workspace), channel=channel, signer_public_keys=signer_keys,
    )
    package_set = _hashed_json(package_set_path, "package_set_sha256")
    lock = _hashed_json(package_lock_path, "lock_sha256")
    generator = _tool("generate-release-catalog")
    version = package_set.get("release_version")
    if package_set.get("profile") != "full" or not isinstance(version, str):
        raise ValueError("full candidate requires the complete full-profile package set")
    expected = _tool("generate-release-package-set")["generate_package_set"](
        core_version=version, requirements=META_ROOT / "requirements-release.txt",
        workspace=workspace, profile="full",
        meta_package=META_ROOT / "packages/govoplan-meta/pyproject.toml",
    )
    if package_set != expected:
        raise ValueError("package set differs from exact Meta full pins or immutable tag metadata")
    generator["_validate_release_inputs"](package_set, lock, core_version=version)
    if lock.get("registries") != package_set.get("registries"):
        raise ValueError("artifact lock uses different package registries")
    versions = {row["repository"]: row["version"] for row in package_set["python"]}
    origin_failures = registered_source_origin_issues(
        repo_versions=versions, workspace=workspace, remote=source_remote,
    )
    if origin_failures:
        raise ValueError("Registered source origin gate failed: " + "; ".join(item.describe() for item in origin_failures))
    selected = set(selected_repositories)
    if not selected or len(selected) != len(selected_repositories) or not selected <= versions.keys():
        raise ValueError("selected repositories must be unique members of the full package set")
    selected_versions = {repo: versions[repo] for repo in sorted(selected)}
    enforce_selected_version_alignment(repo_versions=selected_versions, workspace=workspace)
    failures = source_tag_provenance_issues(
        repo_versions=versions, workspace=workspace, remote=source_remote,
        require_head_repos=selected,
    )
    if failures:
        raise ValueError("Full source provenance gate failed: " + "; ".join(item.describe() for item in failures))
    provenance = selected_source_provenance(repo_versions=versions, workspace=workspace)
    wheel_identities = verify_registry_artifacts(
        package_set=package_set, lock=lock, wheelhouse=wheelhouse,
        webui_packages=webui_packages,
    )
    generated_at = datetime.now(tz=UTC)
    resolved_sequence = sequence if sequence is not None else next_sequence(base, generated_at=generated_at)
    if isinstance(resolved_sequence, bool) or not isinstance(resolved_sequence, int) or resolved_sequence <= int(base.get("sequence") or 0):
        raise ValueError("full candidate sequence must advance the authenticated published channel")
    # Fresh tagged manifests, including unchanged tagged ancestors; no legacy
    # entry, registry hash, source URL or dependency contract is carried over.
    candidate = generator["_catalog_payload"](
        package_set=package_set, package_lock=lock, channel=channel,
        sequence=resolved_sequence, generated_at=generated_at,
        expires_at=generated_at + timedelta(days=expires_days), workspace=workspace,
        public_base_url=public_base_url.rstrip("/"),
    )
    for entry in [candidate["core_release"], *candidate["modules"]]:
        repo = entry["python_package"]
        entry["source"] = {
            "repository": repo, "tag": f"v{versions[repo]}",
            "commit": provenance[repo]["commit_sha"],
            "tag_object_sha": provenance[repo]["tag_object_sha"],
            "repository_url": f"https://git.add-ideas.de/GovOPlaN/{repo}",
            "revision_url": f"https://git.add-ideas.de/GovOPlaN/{repo}/commit/{provenance[repo]['commit_sha']}",
        }
    candidate["release"].update({
        "selected_units": [
            {"repo": repo, "version": versions[repo], "tag": f"v{versions[repo]}", **provenance[repo]}
            for repo in sorted(selected)
        ],
        "keyring_sha256": canonical_hash(keyring),
        "artifacts": wheel_identities,
        "base_catalog_sha256": canonical_hash(base),
    })
    # Recheck the exact objects used by synthesis, including unchanged source
    # ancestors, before signing. No late tag movement can change this candidate.
    origin_failures = registered_source_origin_issues(
        repo_versions=versions, workspace=workspace, remote=source_remote,
    )
    if origin_failures:
        raise ValueError("Registered source origin changed during synthesis: " + "; ".join(item.describe() for item in origin_failures))
    failures = source_tag_provenance_issues(
        repo_versions=versions, workspace=workspace, remote=source_remote,
        require_head_repos=selected,
        expected_commits={repo: row["commit_sha"] for repo, row in provenance.items()},
        expected_tag_objects={repo: row["tag_object_sha"] for repo, row in provenance.items()},
    )
    if failures:
        raise ValueError("Full source provenance changed during synthesis: " + "; ".join(item.describe() for item in failures))
    failures = candidate_catalog_version_issues(candidate)
    identity_failures = selected_artifact_identity_issues(candidate)
    if failures or identity_failures:
        raise ValueError("full candidate identity validation failed: " + "; ".join(
            [item.message for item in failures] + list(identity_failures)
        ))
    candidate["signatures"] = [signature(candidate, key_id=key, private_key=value) for key, value in parsed_keys]
    validation = validate_catalog_object(candidate, approved_channel=channel, signer_public_keys=signer_keys)
    if validation.get("valid") is not True:
        raise ValueError(f"signed full candidate failed validation: {validation.get('error')}")
    # Exclusive output creation preserves earlier reviewed candidates.
    output.mkdir(mode=0o700)
    (output / "channels").mkdir(mode=0o700)
    catalog_path = output / "channels" / f"{channel}.json"
    _write_private_json(catalog_path, candidate)
    _write_private_json(output / "keyring.json", keyring)
    write_module_directory(
        catalog_payload=candidate, keyring_payload=keyring, output_root=output,
        channel=channel, public_base_url=public_base_url,
    )
    result = {
        "status": "ready", "candidate_dir": str(output), "catalog_path": str(catalog_path),
        "channel": channel, "sequence": resolved_sequence,
        "package_count": len(package_set["python"]), "webui_count": len(package_set["webui"]),
        "selected_count": len(selected), "candidate_catalog_hash": canonical_hash(candidate),
        "candidate_keyring_hash": canonical_hash(keyring), "validation_valid": True,
    }
    _write_private_json(output / "summary.json", result)
    harden_private_candidate_tree(output)
    return result


def _hashed_json(path: Path, field: str) -> dict[str, Any]:
    payload = read_bounded_json_source(path, label=field)
    if not isinstance(payload, dict):
        raise ValueError(f"{field} input must be an object")
    unsigned = dict(payload)
    expected = unsigned.pop(field, None)
    # Registry tools use their established ASCII-escaped canonical form.
    encoded = json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode()
    if expected != hashlib.sha256(encoded).hexdigest():
        raise ValueError(f"{field} does not match its contents")
    return payload


def verify_registry_artifacts(
    *, package_set: dict[str, Any], lock: dict[str, Any],
    wheelhouse: Path, webui_packages: Path,
) -> list[dict[str, object]]:
    """Compare exact registry bytes, metadata and source bindings without installs."""

    identities = []
    for group, root, suffix in (("python", wheelhouse, ".whl"), ("webui", webui_packages, ".tgz")):
        ensure_private_candidate_root(root)
        expected = {row["name"]: row for row in package_set[group]}
        rows = lock[group]
        if not isinstance(rows, list) or len(rows) != len(expected):
            raise ValueError(f"artifact lock has duplicate/missing {group} rows")
        filenames: set[str] = set()
        seen: set[str] = set()
        for row in rows:
            selected = expected.get(row.get("name")) if isinstance(row, dict) else None
            if selected is None or row["name"] in seen:
                raise ValueError(f"artifact lock has unexpected/duplicate {group} identities")
            seen.add(row["name"])
            for key in ("name", "version", "repository", "tag", "commit"):
                if row.get(key) != selected[key]:
                    raise ValueError("artifact lock differs from selected source identity")
            filename = row.get("filename")
            if not isinstance(filename, str) or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,254}", filename) is None or not filename.endswith(suffix) or filename in filenames:
                raise ValueError("artifact filename is invalid or duplicated")
            filenames.add(filename)
            path = root / filename
            sha256, sha512, size = _hash_artifact(path)
            if (sha256, size) != (row.get("sha256"), row.get("size")):
                raise ValueError(f"registry artifact bytes differ from locked identity: {filename}")
            if group == "python":
                expected_url = _tool("resolve-package-artifacts")["_python_artifact_url"](
                    package_set["registries"]["python"], package=selected, filename=filename,
                )
                if row.get("url") != expected_url:
                    raise ValueError("Python artifact URL differs from the selected registry")
                identity = inspect_python_wheel(path)
                if (identity.package_name, identity.package_version, identity.archive_sha256, identity.archive_size) != (row["name"], row["version"], sha256, size):
                    raise ValueError("wheel metadata or bytes differ from the registry lock")
                identities.append(identity.catalog_payload())
            else:
                expected_url = (
                    package_set["registries"]["npm"].rstrip("/") + "/"
                    + quote(row["name"], safe="") + "/-/"
                    + quote(row["version"], safe="") + "/"
                    + quote(row["name"].split("/", 1)[1] + "-" + row["version"] + ".tgz", safe="")
                )
                if row.get("url") != expected_url:
                    raise ValueError("WebUI artifact URL differs from the selected registry")
                if row.get("integrity") != "sha512-" + base64.b64encode(sha512).decode("ascii"):
                    raise ValueError("WebUI registry integrity differs from downloaded bytes")
                _inspect_npm_metadata(
                    path, name=row["name"], version=row["version"],
                    expected_identity=(sha256, sha512, size),
                )
        actual = set()
        for index, path in enumerate(root.iterdir()):
            if index >= 1000:
                raise ValueError("registry artifact directory exceeds its inspection bound")
            actual.add(path.name)
        if actual != filenames:
            raise ValueError(f"registry directory contains unexpected or missing {group} files")
    return sorted(identities, key=lambda row: str(row["package_name"]))


def _hash_artifact(path: Path) -> tuple[str, bytes, int]:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
    try:
        initial = os.fstat(descriptor)
        if not stat.S_ISREG(initial.st_mode) or not 0 < initial.st_size <= MAX_ARTIFACT_BYTES:
            raise ValueError("registry artifact must be a bounded regular file")
        digest, sri, total = hashlib.sha256(), hashlib.sha512(), 0
        while chunk := os.read(descriptor, 1024 * 1024):
            total += len(chunk)
            if total > MAX_ARTIFACT_BYTES:
                raise ValueError("registry artifact exceeds its byte bound")
            digest.update(chunk)
            sri.update(chunk)
        final = os.fstat(descriptor)
        if (initial.st_ino, initial.st_size, initial.st_mtime_ns, initial.st_ctime_ns) != (final.st_ino, total, final.st_mtime_ns, final.st_ctime_ns):
            raise ValueError("registry artifact changed while being inspected")
        return digest.hexdigest(), sri.digest(), total
    finally:
        os.close(descriptor)


def _inspect_npm_metadata(
    path: Path, *, name: str, version: str,
    expected_identity: tuple[str, bytes, int] | None = None,
) -> None:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))
    try:
        initial = os.fstat(descriptor)
        if not stat.S_ISREG(initial.st_mode) or not 0 < initial.st_size <= MAX_ARTIFACT_BYTES:
            raise ValueError("WebUI archive must be a bounded regular file")
        digest, sri, total = hashlib.sha256(), hashlib.sha512(), 0
        while chunk := os.read(descriptor, 1024 * 1024):
            total += len(chunk)
            if total > MAX_ARTIFACT_BYTES:
                raise ValueError("WebUI archive exceeds its byte bound")
            digest.update(chunk)
            sri.update(chunk)
        if expected_identity is not None and (digest.hexdigest(), sri.digest(), total) != expected_identity:
            raise ValueError("WebUI archive changed before metadata inspection")
        os.lseek(descriptor, 0, os.SEEK_SET)
        with os.fdopen(os.dup(descriptor), "rb") as stream:
            _inspect_npm_stream(stream, name=name, version=version)
        final = os.fstat(descriptor)
        if (initial.st_ino, initial.st_size, initial.st_mtime_ns, initial.st_ctime_ns) != (final.st_ino, total, final.st_mtime_ns, final.st_ctime_ns):
            raise ValueError("WebUI archive changed during metadata inspection")
    finally:
        os.close(descriptor)


def _inspect_npm_stream(stream: Any, *, name: str, version: str) -> None:
    found = False
    total = 0
    with tarfile.open(fileobj=stream, mode="r|gz") as archive:
        for index, member in enumerate(archive):
            total += member.size
            parts = PurePosixPath(member.name).parts
            if index >= 10000 or total > 1024 * 1024 * 1024 or member.size > 64 * 1024 * 1024:
                raise ValueError("WebUI archive exceeds its inspection bound")
            if not parts or parts[0] != "package" or ".." in parts or not (member.isfile() or member.isdir()):
                raise ValueError("WebUI archive contains an unsafe member")
            if member.name == "package/package.json":
                if found or not member.isfile() or member.size > 1024 * 1024:
                    raise ValueError("WebUI archive has duplicate or oversized metadata")
                stream = archive.extractfile(member)
                if stream is None:
                    raise ValueError("WebUI archive metadata cannot be read")
                metadata = json.loads(stream.read(1024 * 1024 + 1))
                if not isinstance(metadata, dict) or (metadata.get("name"), metadata.get("version")) != (name, version):
                    raise ValueError("WebUI metadata differs from the registry lock")
                found = True
    if not found:
        raise ValueError("WebUI archive metadata is missing")


def _write_private_json(path: Path, payload: object) -> None:
    encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode()
    if len(encoded) > 16 * 1024 * 1024:
        raise ValueError("candidate JSON exceeds its byte bound")
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600)
    with os.fdopen(descriptor, "wb") as handle:
        handle.write(encoded)