DataOps-platform/scripts/verify_p3_wp14_fresh_local_engineering.py

"""Issue a local PASS only after a fresh, bounded regeneration in this process.

The persisted ledger, log, and trace are intentionally unsigned reproducibility
caches.  They are useful inputs to this verifier, but never independently
authorize a PASS_LOCAL claim: this entry point always reruns the generator
before reading them again.
"""

from __future__ import annotations

import hashlib
import json
import os
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

from generate_p3_wp14_verification_ledger import INPUT_PATHS, ledger_is_fresh
from p3_wp14_secure_io import read_bytes_once, read_json_once


ROOT = Path(__file__).resolve().parents[1]
COMMAND_HOME = ROOT / "work/p3_wp14_fresh_command_home"
PYTHON = ROOT / ".venv/bin/python"
GENERATOR = ROOT / "scripts/generate_p3_wp14_verification_ledger.py"
GENERATOR_TIMEOUT_SECONDS = 360
MANIFEST = ROOT / "docs/phase3/P3_WP14_RELEASE_MANIFEST.json"
LEDGER = ROOT / "docs/validation/P3_WP14_LOCAL_VERIFICATION_LEDGER.json"
LOG = ROOT / "docs/validation/P3_WP14_LOCAL_VERIFICATION.log"
TRACE = ROOT / "docs/phase3/P3_WP14_RELEASE_TRACEABILITY.json"
CASES = ROOT / "docs/acceptance/P3_WP14_UAT_CASES.json"
STATE = ROOT / "docs/validation/P3_WP14_VERIFICATION_RUN_STATE.json"


def _sha256(path: Path) -> str:
    return hashlib.sha256(read_bytes_once(path)).hexdigest()


def _minimal_command_env() -> dict[str, str]:
    return {
        "PATH": os.environ.get("PATH", ""),
        "HOME": str(COMMAND_HOME),
        "TMPDIR": str(COMMAND_HOME / "tmp"),
        "DOCKER_CONFIG": str(COMMAND_HOME / "docker"),
        "PYTHONDONTWRITEBYTECODE": "1",
        "LANG": "C.UTF-8",
        "LC_ALL": "C.UTF-8",
    }


def _run_generator() -> str:
    completed = subprocess.run(
        [str(PYTHON), str(GENERATOR)],
        cwd=ROOT,
        env=_minimal_command_env(),
        text=True,
        capture_output=True,
        check=False,
        timeout=GENERATOR_TIMEOUT_SECONDS,
    )
    if completed.returncode != 0:
        raise RuntimeError("fresh local ledger generator failed")
    state = read_json_once(STATE)
    if not isinstance(state, dict) or state.get("status") != "COMPLETE" or not isinstance(state.get("run_nonce"), str):
        raise RuntimeError("generator did not produce a complete session state")
    return state["run_nonce"]


def _assert_current_expiry(payload: dict, now: datetime) -> None:
    if not ledger_is_fresh(payload, now):
        raise ValueError("unsigned reproducibility cache is expired")


def _expected_case_ids() -> set[str]:
    cases = read_json_once(CASES)
    if not isinstance(cases, dict) or not isinstance(cases.get("cases"), list):
        raise ValueError("invalid acceptance case matrix")
    return {case["id"] for case in cases["cases"] if case.get("result") == "PASS_LOCAL"}


def _verify_cache(now: datetime, *, expected_nonce: str | None = None) -> None:
    ledger = read_json_once(LEDGER)
    trace = read_json_once(TRACE)
    state = read_json_once(STATE)
    if not isinstance(ledger, dict) or not isinstance(trace, dict) or not isinstance(state, dict):
        raise ValueError("invalid unsigned cache format")
    if state.get("status") != "COMPLETE" or state.get("run_nonce") != ledger.get("run_nonce"):
        raise ValueError("ledger state does not identify a complete session")
    if expected_nonce is not None and ledger.get("run_nonce") != expected_nonce:
        raise ValueError("fresh verifier session nonce mismatch")
    if ledger.get("cache_class") != "UNSIGNED_REPRODUCIBILITY_CACHE":
        raise ValueError("ledger is not explicitly an unsigned cache")
    if ledger.get("pass_local_authority") != "FRESH_VERIFIER_CURRENT_PROCESS_ONLY":
        raise ValueError("ledger must not independently grant PASS_LOCAL")
    anchor = ledger.get("external_signature_anchor")
    if anchor != {"status": "TBD_EXTERNAL", "gate": "BLOCKED_EXTERNAL"}:
        raise ValueError("external signature boundary is not closed")
    _assert_current_expiry(ledger, now)

    if trace.get("release_manifest_sha256") != _sha256(MANIFEST):
        raise ValueError("manifest digest mismatch")
    if trace.get("verification_ledger_sha256") != _sha256(LEDGER):
        raise ValueError("ledger digest mismatch")
    if trace.get("verification_log_sha256") != _sha256(LOG):
        raise ValueError("verification log digest mismatch")
    if trace.get("local_verification_cache") != {
        "classification": "UNSIGNED_REPRODUCIBILITY_CACHE",
        "pass_local_authority": "FRESH_VERIFIER_CURRENT_PROCESS_ONLY",
        "external_signature_status": "TBD_EXTERNAL",
        "enterprise_gate": "BLOCKED_EXTERNAL",
    }:
        raise ValueError("trace does not preserve unsigned-cache boundary")

    controlled = {path: _sha256(ROOT / path) for path in ledger.get("controlled_input_paths", [])}
    controlled_digest = hashlib.sha256(
        json.dumps(controlled, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    if ledger.get("controlled_input_paths") != list(INPUT_PATHS) or ledger.get("controlled_input_sha256") != controlled_digest:
        raise ValueError("controlled input closure drift")
    if ledger.get("log_sha256") != _sha256(LOG):
        raise ValueError("verification log digest mismatch")

    case_ids = {case_id for run in ledger.get("runs", []) for case_id in run.get("case_ids", [])}
    if case_ids != _expected_case_ids():
        raise ValueError("PASS_LOCAL case mapping drift")
    for run in ledger.get("runs", []):
        dependencies = run.get("dependency_sha256")
        if run.get("status") != "PASS" or run.get("exit_code") != 0 or not isinstance(dependencies, dict):
            raise ValueError("bounded local command did not pass")
        current = {path: _sha256(ROOT / path) for path in dependencies}
        aggregate = hashlib.sha256(
            json.dumps(current, sort_keys=True, separators=(",", ":")).encode("utf-8")
        ).hexdigest()
        if dependencies != current or run.get("dependency_aggregate_sha256") != aggregate:
            raise ValueError("command dependency closure drift")


def main(argv: list[str] | None = None) -> int:
    arguments = [] if argv is None else argv
    if set(arguments) - {"--verify-existing"} or len(arguments) > 1:
        print("usage: verify_p3_wp14_fresh_local_engineering.py [--verify-existing]", file=sys.stderr)
        return 2
    try:
        generated_nonce = None
        if not arguments:
            generated_nonce = _run_generator()
        _verify_cache(datetime.now(timezone.utc), expected_nonce=generated_nonce)
    except (subprocess.TimeoutExpired, Exception) as error:
        print(f"P3-WP14 fresh local verification failed: {type(error).__name__}", file=sys.stderr)
        return 1
    if arguments:
        print("P3-WP14 VERIFY_EXISTING_DIAGNOSTIC_ONLY (cannot grant PASS_LOCAL)")
    else:
        print("P3-WP14 PASS_LOCAL_CURRENT_SESSION (unsigned local reproducibility only)")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))