skills/.portable-resume/runtime/portable_resume/build_identity.py

"""Deterministic, dependency-free build provenance for release artifacts.

The public package version remains the PEP 440 base version from
``portable_resume.__version__``.  A build identity adds checkout provenance
without making normal runtime behavior depend on Git or a third-party version
plugin.  Source archives without Git metadata stay deterministic and report
unknown commit/dirty state honestly.
"""

from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
import stat
from dataclasses import asdict
from pathlib import Path
from typing import Any, Mapping

from . import __version__

BUILD_IDENTITY_SCHEMA_V1 = "portable-resume/build-identity-v1"
BUILD_IDENTITY_SCHEMA_V2 = "portable-resume/build-identity-v2"
BUILD_IDENTITY_SCHEMA = BUILD_IDENTITY_SCHEMA_V2
LATEST_RELEASE_SCHEMA = "portable-resume/latest-release-v1"

_RELEASE_VERSION = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$")
_DEVELOPMENT_VERSION = re.compile(
    r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)\.dev(0|[1-9]\d*)$"
)
_IDENTITY_VERSION = re.compile(
    r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)"
    r"(?:\.dev(0|[1-9]\d*))?(?:\+[0-9a-z]+(?:\.[0-9a-z]+)*)?$"
)
_HEX_40 = re.compile(r"^[0-9a-f]{40}$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")

_V1_REQUIRED_KEYS = frozenset(
    {
        "schema",
        "version",
        "base_version",
        "release_channel",
        "commit_sha",
        "dirty",
        "registry_sha256",
        "source_sha256",
    }
)
_V2_REQUIRED_KEYS = _V1_REQUIRED_KEYS | {"build_inputs_sha256"}
_OPTIONAL_KEYS = frozenset({"provenance"})
_GENERATED_IDENTITY = Path("resources/build-identity.json")
_LATEST_RELEASE = Path("resources/latest-release.json")
MAX_BUILD_IDENTITY_BYTES = 16 * 1024
MAX_SOURCE_FILE_BYTES = 64 * 1024 * 1024
MAX_SOURCE_TOTAL_BYTES = 512 * 1024 * 1024


def _canonical_json_bytes(value: object) -> bytes:
    return (
        json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
        + "\n"
    ).encode("utf-8")


def _registry_payload() -> dict[str, object]:
    """Return the closed registry representation used by identity hashing."""
    from .registry import DESTINATION_PROFILES, PACKAGE_SURFACES, SOURCE_PROFILES

    return {
        "sources": {
            key: asdict(SOURCE_PROFILES[key]) for key in sorted(SOURCE_PROFILES)
        },
        "destinations": {
            key: asdict(DESTINATION_PROFILES[key])
            for key in sorted(DESTINATION_PROFILES)
        },
        "packages": {
            key: asdict(PACKAGE_SURFACES[key]) for key in sorted(PACKAGE_SURFACES)
        },
    }


def latest_release(package_root: Path | None = None) -> dict[str, object]:
    """Load and validate the immutable latest-published-release baseline."""
    root = (
        Path(package_root).resolve()
        if package_root is not None
        else Path(__file__).resolve().parent
    )
    try:
        value = json.loads((root / _LATEST_RELEASE).read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise ValueError("latest release baseline is unreadable") from error
    if not isinstance(value, dict):
        raise ValueError("latest release baseline has the wrong shape")
    required = {
        "schema",
        "version",
        "tag",
        "commit_sha",
        "registry_sha256",
        "source_sha256",
        "matrix_dimensions",
        "artifact_classes",
        "published_at",
        "github_release",
        "pypi",
    }
    if set(value) != required or value.get("schema") != LATEST_RELEASE_SCHEMA:
        raise ValueError("latest release baseline fields are invalid")
    version = value.get("version")
    if not isinstance(version, str) or _RELEASE_VERSION.fullmatch(version) is None:
        raise ValueError("latest release version is invalid")
    if value.get("tag") != f"v{version}":
        raise ValueError("latest release tag does not match its version")
    if not isinstance(value.get("commit_sha"), str) or _HEX_40.fullmatch(
        str(value["commit_sha"])
    ) is None:
        raise ValueError("latest release commit is invalid")
    for key in ("registry_sha256", "source_sha256"):
        if not isinstance(value.get(key), str) or _SHA256.fullmatch(
            str(value[key])
        ) is None:
            raise ValueError(f"latest release {key} is invalid")
    dimensions = value.get("matrix_dimensions")
    if not isinstance(dimensions, dict) or set(dimensions) != {
        "sources",
        "destinations",
        "cells",
    }:
        raise ValueError("latest release matrix dimensions are invalid")
    if not all(type(dimensions[key]) is int and dimensions[key] > 0 for key in dimensions):
        raise ValueError("latest release matrix counts are invalid")
    if dimensions["sources"] * dimensions["destinations"] != dimensions["cells"]:
        raise ValueError("latest release matrix is not rectangular")
    classes = value.get("artifact_classes")
    if (
        not isinstance(classes, list)
        or not classes
        or not all(isinstance(item, str) and item for item in classes)
        or classes != sorted(set(classes))
    ):
        raise ValueError("latest release artifact classes are invalid")
    for key in ("published_at", "github_release", "pypi"):
        if not isinstance(value.get(key), str) or not value[key]:
            raise ValueError(f"latest release {key} is invalid")
    return value


def registry_sha256(
    *, registry_payload: Mapping[str, object] | None = None
) -> str:
    """Hash the canonical source/destination/package profile representation."""
    payload: Mapping[str, object]
    if registry_payload is None:
        payload = _registry_payload()
    else:
        payload = registry_payload
    return hashlib.sha256(_canonical_json_bytes(payload)).hexdigest()


def _stat_fingerprint(value: os.stat_result) -> tuple[int, ...]:
    """Stable identity for open-vs-path bind checks.

    On Windows, opening a file can update ctime without content change, so the
    fingerprint omits ``st_ctime_ns`` there (#207 nt qualification).
    """

    base = (
        value.st_dev,
        value.st_ino,
        value.st_mode,
        value.st_size,
        value.st_mtime_ns,
    )
    if os.name == "nt":
        return base
    return base + (value.st_ctime_ns,)


def _read_stable_regular_file(
    path: Path,
    *,
    before: os.stat_result,
    max_bytes: int,
    subject: str,
) -> bytes:
    """Read one bounded inode without following a swapped final symlink."""
    if not stat.S_ISREG(before.st_mode) or before.st_size > max_bytes:
        raise ValueError(f"{subject} is not a bounded regular file")
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(path, flags)
    except OSError as error:
        raise ValueError(f"{subject} is unreadable") from error
    try:
        opened = os.fstat(descriptor)
        if (
            not stat.S_ISREG(opened.st_mode)
            or opened.st_size > max_bytes
            or _stat_fingerprint(opened) != _stat_fingerprint(before)
        ):
            raise ValueError(f"{subject} changed before open")
        chunks: list[bytes] = []
        remaining = max_bytes + 1
        while remaining:
            chunk = os.read(descriptor, min(remaining, 64 * 1024))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        data = b"".join(chunks)
        after = os.fstat(descriptor)
        if len(data) > max_bytes:
            raise ValueError(f"{subject} is oversized")
        if _stat_fingerprint(after) != _stat_fingerprint(opened):
            raise ValueError(f"{subject} changed during read")
    finally:
        os.close(descriptor)
    try:
        final_path = path.lstat()
    except OSError as error:
        raise ValueError(f"{subject} changed after read") from error
    if _stat_fingerprint(final_path) != _stat_fingerprint(opened):
        raise ValueError(f"{subject} changed after read")
    return data


def source_sha256(package_root: Path) -> str:
    """Hash deployment-stable package paths, types, and source bytes."""
    root = package_root.resolve()
    if not root.is_dir():
        raise ValueError("package root is not a directory")
    digest = hashlib.sha256()
    found = False
    total_bytes = 0
    for path in sorted(root.rglob("*"), key=lambda item: item.as_posix()):
        relative = path.relative_to(root)
        if "__pycache__" in relative.parts or relative == _GENERATED_IDENTITY:
            continue
        try:
            metadata = path.lstat()
        except OSError as error:
            raise ValueError("package source is unreadable") from error
        name = relative.as_posix().encode("utf-8")
        if stat.S_ISDIR(metadata.st_mode):
            digest.update(name)
            digest.update(b"\0directory\0")
            continue
        if not stat.S_ISREG(metadata.st_mode):
            raise ValueError("package source contains a non-regular file")
        if path.suffix in {".pyc", ".pyo"}:
            continue
        data = _read_stable_regular_file(
            path,
            before=metadata,
            max_bytes=MAX_SOURCE_FILE_BYTES,
            subject="package source file",
        )
        total_bytes += len(data)
        if total_bytes > MAX_SOURCE_TOTAL_BYTES:
            raise ValueError("package source exceeds the aggregate bound")
        digest.update(name)
        digest.update(b"\0file\0")
        digest.update(data)
        digest.update(b"\0")
        found = True
    if not found:
        raise ValueError("package root contains no source files")
    return digest.hexdigest()


def _validate_base_version(base_version: str) -> str:
    if _RELEASE_VERSION.fullmatch(base_version):
        return "stable"
    if _DEVELOPMENT_VERSION.fullmatch(base_version):
        return "development"
    raise ValueError("base version must be strict X.Y.Z or X.Y.Z.devN")


def build_identity(
    *,
    package_root: Path | None = None,
    base_version: str | None = None,
    commit_sha: str | None = None,
    dirty: bool | None = None,
    exact_tag: bool = False,
    build_inputs_sha256: str | None = None,
) -> dict[str, object]:
    """Return a closed identity from explicit, already-collected build facts.

    This product-path function never invokes Git or any other process.  Normal
    reader/installer runtimes therefore report an honest source-archive
    fallback.  Explicit build and release tooling may collect Git facts outside
    ``src/portable_resume`` and pass them here.
    """
    package = (
        Path(package_root).resolve()
        if package_root is not None
        else Path(__file__).resolve().parent
    )
    base = base_version or __version__
    base_kind = _validate_base_version(base)
    if type(exact_tag) is not bool:
        raise ValueError("exact tag state must be boolean")
    if commit_sha is None:
        if dirty is not None or exact_tag or build_inputs_sha256 is not None:
            raise ValueError("dirty/tag facts require a commit SHA")
    else:
        if _HEX_40.fullmatch(commit_sha) is None:
            raise ValueError("invalid commit SHA")
        if type(dirty) is not bool:
            raise ValueError("Git identity requires a boolean dirty state")
        if (
            not isinstance(build_inputs_sha256, str)
            or _SHA256.fullmatch(build_inputs_sha256) is None
        ):
            raise ValueError("Git identity requires a build inputs digest")

    release = base_kind == "stable" and exact_tag and dirty is False
    release_channel = "release" if release else "development"
    if release:
        version = base
    elif commit_sha is not None:
        version = f"{base}+g{commit_sha[:12]}"
        if dirty:
            version += ".dirty"
    else:
        # A raw source archive cannot prove a commit or cleanliness.  Retain the
        # deterministic base version and expose the unknowns as null fields.
        version = base

    identity: dict[str, object] = {
        "schema": BUILD_IDENTITY_SCHEMA,
        "version": version,
        "base_version": base,
        "release_channel": release_channel,
        "commit_sha": commit_sha,
        "dirty": dirty,
        "build_inputs_sha256": build_inputs_sha256,
        "registry_sha256": registry_sha256(),
        "source_sha256": source_sha256(package),
        "provenance": "git" if commit_sha is not None else "source-archive",
    }
    validate_identity(identity)
    return identity


def validate_identity(identity: Mapping[str, Any]) -> None:
    """Validate the closed identity schema; raise ``ValueError`` on drift."""
    schema = identity.get("schema")
    if schema == BUILD_IDENTITY_SCHEMA_V1:
        required_keys = _V1_REQUIRED_KEYS
    elif schema == BUILD_IDENTITY_SCHEMA_V2:
        required_keys = _V2_REQUIRED_KEYS
    else:
        raise ValueError("unsupported build identity schema")
    keys = frozenset(identity)
    if not required_keys.issubset(keys) or not keys.issubset(
        required_keys | _OPTIONAL_KEYS
    ):
        raise ValueError("build identity fields are incomplete or unknown")
    version = identity.get("version")
    base = identity.get("base_version")
    if not isinstance(version, str) or _IDENTITY_VERSION.fullmatch(version) is None:
        raise ValueError("invalid identity version")
    if not isinstance(base, str):
        raise ValueError("invalid base version")
    _validate_base_version(base)
    if identity.get("release_channel") not in {"release", "development"}:
        raise ValueError("invalid release channel")
    commit = identity.get("commit_sha")
    if commit is not None and (
        not isinstance(commit, str) or _HEX_40.fullmatch(commit) is None
    ):
        raise ValueError("invalid commit SHA")
    dirty = identity.get("dirty")
    if dirty is not None and type(dirty) is not bool:
        raise ValueError("invalid dirty state")
    build_inputs_digest = identity.get("build_inputs_sha256")
    if build_inputs_digest is not None and (
        not isinstance(build_inputs_digest, str)
        or _SHA256.fullmatch(build_inputs_digest) is None
    ):
        raise ValueError("invalid build_inputs_sha256")
    for key in ("registry_sha256", "source_sha256"):
        value = identity.get(key)
        if not isinstance(value, str) or _SHA256.fullmatch(value) is None:
            raise ValueError(f"invalid {key}")
    provenance = identity.get("provenance")
    if provenance is not None and provenance not in {
        "git",
        "source-archive",
        "embedded",
    }:
        raise ValueError("invalid provenance")
    channel = identity.get("release_channel")
    if commit is None:
        if dirty is not None or build_inputs_digest is not None:
            raise ValueError("dirty state requires a commit SHA")
        expected_version = base
        if provenance not in {None, "source-archive", "embedded"}:
            raise ValueError("commitless identity has invalid provenance")
    else:
        if type(dirty) is not bool:
            raise ValueError("Git identity requires a boolean dirty state")
        if schema == BUILD_IDENTITY_SCHEMA_V2 and not isinstance(
            build_inputs_digest, str
        ):
            raise ValueError("Git identity requires a build inputs digest")
        expected_version = f"{base}+g{commit[:12]}"
        if dirty:
            expected_version += ".dirty"
        if provenance not in {None, "git", "embedded"}:
            raise ValueError("commit identity has invalid provenance")
    if channel == "release":
        if (
            _RELEASE_VERSION.fullmatch(base) is None
            or commit is None
            or dirty is not False
        ):
            raise ValueError("release identity must be an exact clean Git commit")
        expected_version = base
        if version != expected_version:
            raise ValueError("release identity must use the exact base version")
    elif version != expected_version:
        raise ValueError("development identity version does not match its build facts")


def validate_current_identity(identity: Mapping[str, Any]) -> None:
    """Require the current schema for newly produced or verified artifacts."""
    validate_identity(identity)
    if identity.get("schema") != BUILD_IDENTITY_SCHEMA:
        raise ValueError("artifact operations require the current identity schema")


def identity_json_bytes(identity: Mapping[str, Any]) -> bytes:
    """Serialize a validated identity as canonical UTF-8 JSON."""
    validate_identity(identity)
    return _canonical_json_bytes(identity)


def assert_identity_matches_package(
    identity: Mapping[str, Any],
    *,
    package_root: Path | None = None,
) -> None:
    """Require identity digests to describe the complete package source tree."""
    validate_identity(identity)
    root = (
        Path(package_root).resolve()
        if package_root is not None
        else Path(__file__).resolve().parent
    )
    if identity.get("base_version") != __version__:
        raise ValueError("build identity base version differs from package")
    if identity.get("registry_sha256") != registry_sha256():
        raise ValueError("build identity registry differs from package")
    if identity.get("source_sha256") != source_sha256(root):
        raise ValueError("build identity source digest differs from package")


def load_identity_bytes(data: bytes) -> dict[str, object]:
    """Load one bounded canonical identity document from bytes."""
    if not isinstance(data, bytes) or len(data) > MAX_BUILD_IDENTITY_BYTES:
        raise ValueError("build identity is oversized")
    try:
        value = json.loads(data.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise ValueError("build identity is unreadable") from error
    if not isinstance(value, dict):
        raise ValueError("build identity has the wrong shape")
    validate_identity(value)
    if data != identity_json_bytes(value):
        raise ValueError("build identity is not canonical")
    return dict(value)


def load_identity_file(
    path: str | Path,
    *,
    expected_sha256: str | None = None,
) -> dict[str, object]:
    """Read a canonical identity through a bounded no-follow regular file."""
    if expected_sha256 is not None and _SHA256.fullmatch(expected_sha256) is None:
        raise ValueError("expected build identity digest is invalid")
    identity_path = Path(path)
    try:
        before = identity_path.lstat()
    except FileNotFoundError:
        raise
    except OSError as error:
        raise ValueError("build identity is unreadable") from error
    data = _read_stable_regular_file(
        identity_path,
        before=before,
        max_bytes=MAX_BUILD_IDENTITY_BYTES,
        subject="build identity",
    )
    if expected_sha256 is not None and not hmac.compare_digest(
        hashlib.sha256(data).hexdigest(),
        expected_sha256,
    ):
        raise ValueError("build identity digest mismatch")
    return load_identity_bytes(data)


def load_embedded_identity(
    package_root: Path | None = None,
) -> dict[str, object] | None:
    """Load the fixed packaged identity, or return ``None`` when absent."""
    root = (
        Path(package_root).resolve()
        if package_root is not None
        else Path(__file__).resolve().parent
    )
    try:
        identity = load_identity_file(root / _GENERATED_IDENTITY)
    except FileNotFoundError:
        return None
    if identity.get("base_version") != __version__:
        raise ValueError("embedded identity base version differs from runtime")
    if identity.get("registry_sha256") != registry_sha256():
        raise ValueError("embedded identity registry differs from runtime")
    return identity


def runtime_identity(package_root: Path | None = None) -> dict[str, object]:
    """Return embedded artifact provenance or an honest source-only fallback.

    Runtime lookup uses only the fixed packaged resource. It never consults an
    environment-selected path and never invokes Git or another process.
    """
    root = (
        Path(package_root).resolve()
        if package_root is not None
        else Path(__file__).resolve().parent
    )
    embedded = load_embedded_identity(root)
    if embedded is not None:
        return embedded
    return build_identity(package_root=root)