40-validation/evidence_lifecycle.py

#!/usr/bin/env python3
"""Shared evidence generation, provenance sealing and freshness validation."""
from __future__ import annotations
import hashlib, json
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable


def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def repository_roots(container: Path) -> dict[str, Path]:
    """Resolve repository roots from immediate children or a single root."""
    candidates = [container]
    candidates.extend(p for p in container.iterdir() if p.is_dir())
    roots: dict[str, Path] = {}
    for candidate in candidates:
        manifest = candidate / "manifest.json"
        if not manifest.is_file():
            continue
        data = json.loads(manifest.read_text(encoding="utf-8"))
        rid = data.get("artifactName") or data.get("repository", {}).get("id")
        if not isinstance(rid, str) or not rid:
            continue
        if rid in roots and roots[rid] != candidate:
            raise ValueError(f"duplicate repository identity: {rid}")
        roots[rid] = candidate
    return roots


def resolve_reference(roots: dict[str, Path], reference: str) -> Path:
    if ":" not in reference:
        raise ValueError(f"invalid repository reference: {reference}")
    rid, relative = reference.split(":", 1)
    if rid not in roots:
        raise ValueError(f"unknown repository: {rid}")
    path = (roots[rid] / relative).resolve()
    root = roots[rid].resolve()
    if path != root and root not in path.parents:
        raise ValueError(f"unsafe repository reference: {reference}")
    if not path.is_file():
        raise ValueError(f"missing referenced file: {reference}")
    return path


def build_provenance(
    roots_container: Path,
    validator_reference: str,
    artifact_references: Iterable[str],
) -> dict:
    roots = repository_roots(roots_container)
    validator_path = resolve_reference(roots, validator_reference)
    repositories = {}
    for rid, root in sorted(roots.items()):
        manifest = root / "manifest.json"
        data = json.loads(manifest.read_text(encoding="utf-8"))
        repositories[rid] = {
            "version": data.get("version") or data.get("repository", {}).get("version"),
            "manifestSha256": sha256(manifest),
        }
    artifacts = [
        {"reference": ref, "sha256": sha256(resolve_reference(roots, ref))}
        for ref in artifact_references
    ]
    return {
        "repositories": repositories,
        "validator": {"reference": validator_reference, "sha256": sha256(validator_path)},
        "artifacts": artifacts,
    }


def seal_evidence(
    payload: dict,
    roots_container: Path,
    validator_reference: str,
    artifact_references: Iterable[str],
    *,
    work_item: str,
    chain: str,
) -> dict:
    sealed = dict(payload)
    sealed["generatedAt"] = datetime.now(timezone.utc).isoformat()
    sealed["workItem"] = work_item
    sealed["chain"] = chain
    sealed["freshness"] = "fresh"
    sealed["provenance"] = build_provenance(
        roots_container, validator_reference, artifact_references
    )
    return sealed


def write_evidence(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temp = path.with_name(path.name + ".new")
    temp.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    temp.replace(path)


def validate_evidence(roots_container: Path, evidence_path: Path) -> list[str]:
    errors: list[str] = []
    roots = repository_roots(roots_container)
    data = json.loads(evidence_path.read_text(encoding="utf-8"))
    provenance = data.get("provenance")
    if not isinstance(provenance, dict):
        return ["missing provenance object"]
    repositories = provenance.get("repositories")
    if not isinstance(repositories, dict):
        errors.append("missing provenance.repositories")
    else:
        for rid, metadata in repositories.items():
            if rid not in roots:
                errors.append(f"unknown repository {rid}")
                continue
            expected = metadata.get("manifestSha256") if isinstance(metadata, dict) else None
            if not isinstance(expected, str):
                errors.append(f"{rid}: missing manifestSha256")
            elif sha256(roots[rid] / "manifest.json") != expected:
                errors.append(f"{rid}: stale manifest revision")
    validator = provenance.get("validator")
    if not isinstance(validator, dict):
        errors.append("missing provenance.validator")
    else:
        reference, expected = validator.get("reference"), validator.get("sha256")
        try:
            target = resolve_reference(roots, reference)
        except Exception as exc:
            errors.append(str(exc))
        else:
            if not isinstance(expected, str) or sha256(target) != expected:
                errors.append("stale validator revision")
    for item in provenance.get("artifacts", []):
        if not isinstance(item, dict):
            errors.append("invalid artifact provenance")
            continue
        reference, expected = item.get("reference"), item.get("sha256")
        try:
            target = resolve_reference(roots, reference)
        except Exception as exc:
            errors.append(str(exc))
        else:
            if not isinstance(expected, str) or sha256(target) != expected:
                errors.append(f"stale artifact {reference}")
    return errors


def seal_failed_evidence(
    payload: dict,
    roots_container: Path,
    validator_reference: str,
    artifact_references: Iterable[str] = (),
    *,
    work_item: str,
    chain: str,
    failure_phase: str,
    rollback: dict | None = None,
) -> dict:
    """Seal failed execution evidence with traceable failure and rollback outcome."""
    failed = dict(payload)
    failed.setdefault("evidenceId", "execution-failure")
    failed["result"] = "failed"
    failed.setdefault("exitCode", 1)
    failed["failurePhase"] = failure_phase
    failed["rollback"] = rollback or {"attempted": False, "status": "NOT_REQUIRED"}
    return seal_evidence(
        failed, roots_container, validator_reference, artifact_references,
        work_item=work_item, chain=chain,
    )