DataOps-platform/scripts/generate_p3_wp14_verification_ledger.py

"""Run bounded local P3-WP14 verification and write a machine-readable ledger.

The ledger intentionally stores digests of captured stdout/stderr, not their
contents, so it carries no service configuration or credentials.  It neither
starts Compose nor contacts an enterprise system; the config validator performs
only a local Compose render.
"""

from __future__ import annotations

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

from p3_wp14_secure_io import atomic_write_bytes, exclusive_lock, read_bytes_once, read_json_once


ROOT = Path(__file__).resolve().parents[1]
COMMAND_HOME = ROOT / "work/p3_wp14_ledger_command_home"
PYTHON = ROOT / ".venv/bin/python"
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"
MANIFEST = ROOT / "docs/phase3/P3_WP14_RELEASE_MANIFEST.json"
STATE = ROOT / "docs/validation/P3_WP14_VERIFICATION_RUN_STATE.json"
LOCK = ROOT / "work/p3_wp14_ledger_lock_scope/lock"
TTL_SECONDS = 900
INPUT_PATHS = (
    "docs/phase3/P3_WP14_RELEASE_MANIFEST.json",
    "docs/validation/P3_WP14_LOCAL_CONFIG_A.json",
    "docs/validation/P3_WP14_LOCAL_CONFIG_B.json",
    "docs/validation/P3_WP14_LOCAL_CONFIG_COMPARISON.json",
    "docs/acceptance/P3_WP14_UAT_CASES.json",
    "tests/test_phase3_wp14_acceptance_handover_contract.py",
    "scripts/validate_p3_wp14_local_configs.py",
    "scripts/generate_p3_wp14_release_manifest.py",
    "scripts/generate_p3_wp14_verification_ledger.py",
    "scripts/verify_p3_wp14_fresh_local_engineering.py",
    "scripts/p3_wp14_secure_io.py",
)
TARGETED_TESTS = (
    "tests/test_phase3_wp01_enterprise_acceptance_contract.py",
    "tests/test_phase3_wp02_enterprise_identity_contract.py",
    "tests/test_phase3_wp03_enterprise_connectors.py",
    "tests/test_phase3_wp04_delivery_contract.py",
    "tests/test_phase3_wp04_production_runtime.py",
    "tests/core/orchestration/test_production_observability.py",
    "tests/test_phase3_wp06_subscription_migration_contract.py",
    "tests/test_phase3_wp07_enterprise_delivery_contract.py",
    "tests/acceptance/test_phase3_wp08_third_domain_replication.py",
    "tests/agent/test_wp09_model_gateway.py",
    "tests/test_wp10_tenant_api.py",
    "tests/test_wp11_bi_ai_catalog_api.py",
    "tests/test_wp12_metering_showback_api.py",
    "tests/test_wp13_plugin_platform_api.py",
    "tests/test_permission_matrix.py",
)


def _sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def _sha256_path(path: Path) -> str:
    return _sha256_bytes(read_bytes_once(path))

def ledger_is_fresh(payload: dict, now: datetime | None = None) -> bool:
    current = now or datetime.now(timezone.utc)
    return datetime.fromisoformat(payload["expires_at"].replace("Z", "+00:00")) > current


def _write_json(path: Path, value: object) -> None:
    atomic_write_bytes(path, json.dumps(value, ensure_ascii=False, indent=2).encode("utf-8") + b"\n")


def _sanitized(value: str) -> str:
    return re.sub(r"(?i)(password|token|secret|api[_-]?key)\s*[:=]\s*\S+", r"\1=<REDACTED>", value)


def _entry_path(command: list[str], dependencies: list[Path]) -> Path:
    """The entry is the first repository file explicitly passed to the command."""
    for part in command[1:]:
        path = ROOT / part
        if path.is_file():
            return path
    return dependencies[0]


def _minimal_command_env() -> dict[str, str]:
    """Run bounded local commands without inherited credential/config paths."""
    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(command_id: str, command: list[str], case_ids: list[str]) -> tuple[dict, str]:
    completed = subprocess.run(
        command,
        cwd=ROOT,
        env=_minimal_command_env(),
        text=True,
        capture_output=True,
        check=False,
        timeout=300,
    )
    output = completed.stdout + completed.stderr
    sanitized_output = _sanitized(output)
    match = re.search(r"(\d+) passed", output)
    dependencies = [ROOT / part for part in command[1:] if (ROOT / part).is_file()]
    if command_id == "p3_targeted_regression": dependencies.extend(ROOT / target for target in TARGETED_TESTS)
    dependencies.append(ROOT / "scripts/p3_wp14_secure_io.py")
    dependencies = sorted(set(dependencies))
    dependency_digests = {path.relative_to(ROOT).as_posix(): _sha256_path(path) for path in dependencies}
    run = {
        "command_id": command_id,
        "command": command,
        "case_ids": case_ids,
        "run_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "status": "PASS" if completed.returncode == 0 else "FAIL",
        "exit_code": completed.returncode,
        "passed_count": int(match.group(1)) if match else 1 if completed.returncode == 0 else 0,
        "timeout_seconds": 300,
        "cwd": ".",
        "command_sha256": _sha256_bytes(json.dumps(command, separators=(",", ":")).encode("utf-8")),
        "entry_path": _entry_path(command, dependencies).relative_to(ROOT).as_posix(),
        "script_sha256": _sha256_path(_entry_path(command, dependencies)),
        "dependency_sha256": dependency_digests,
        "dependency_aggregate_sha256": _sha256_bytes(json.dumps(dependency_digests, sort_keys=True, separators=(",", ":")).encode("utf-8")),
        "stdout_sha256": _sha256_bytes(_sanitized(completed.stdout).encode("utf-8")),
        "stderr_sha256": _sha256_bytes(_sanitized(completed.stderr).encode("utf-8")),
    }
    return run, f"{command_id} exit={completed.returncode} passed={run['passed_count']} output_sha256={_sha256_bytes(sanitized_output.encode('utf-8'))}"


def main() -> int:
    with exclusive_lock(LOCK) as nonce:
        now = datetime.now(timezone.utc).replace(microsecond=0)
        _write_json(STATE, {
            "schema_version": "1.0", "status": "RUNNING", "run_nonce": nonce,
            "started_at": now.isoformat().replace("+00:00", "Z"),
            "expires_at": (now + timedelta(seconds=TTL_SECONDS)).isoformat().replace("+00:00", "Z"),
        })
        try:
            return _main_locked(nonce, now)
        except (subprocess.TimeoutExpired, Exception) as error:
            _write_json(STATE, {"schema_version": "1.0", "status": "FAILED", "run_nonce": nonce, "error_type": type(error).__name__, "invalidates_prior_ledger_and_trace": True})
            return 1


def _main_locked(nonce: str, now: datetime) -> int:
    commands = (
        (
            "p3_targeted_regression",
            [str(PYTHON), "-m", "pytest", "-q", *TARGETED_TESTS],
            [f"P3-WP14-UAT-{number:02d}" for number in range(1, 14)] + ["P3-WP14-UAT-23", "P3-WP14-UAT-25", "P3-WP14-UAT-27"],
        ),
        (
            "local_config_compose_render",
            [str(PYTHON), "scripts/validate_p3_wp14_local_configs.py"],
            ["P3-WP14-UAT-30"],
        ),
        (
            "release_manifest_generation",
            [str(PYTHON), "scripts/generate_p3_wp14_release_manifest.py"],
            ["P3-WP14-UAT-29"],
        ),
        (
            "wp14_264_audit_contract",
            [str(PYTHON), "-m", "pytest", "-q", "tests/test_phase3_wp14_acceptance_handover_contract.py", "-k", "264_item_closure"],
            ["P3-WP14-UAT-26"],
        ),
        (
            "wp14_browser_boundary_contract",
            [str(PYTHON), "-m", "pytest", "-q", "tests/test_phase3_wp14_acceptance_handover_contract.py", "-k", "report_ledger_signoff_runbook"],
            ["P3-WP14-UAT-28"],
        ),
    )
    runs: list[dict] = []
    log_sections: list[str] = []
    for command_id, command, case_ids in commands:
        run, output = _run(command_id, command, case_ids)
        runs.append(run)
        log_sections.append(output)
        if run["exit_code"] != 0:
            atomic_write_bytes(LOG, ("\n".join(log_sections) + "\n").encode("utf-8"))
            _write_json(STATE, {"schema_version": "1.0", "status": "FAILED", "run_nonce": nonce, "failed_command_id": command_id})
            return run["exit_code"]
    atomic_write_bytes(LOG, ("\n".join(log_sections) + "\n").encode("utf-8"))
    controlled = {path: _sha256_path(ROOT / path) for path in INPUT_PATHS}
    payload = {
        "schema_version": "1.0",
        "work_package": "P3-WP14",
        "cache_class": "UNSIGNED_REPRODUCIBILITY_CACHE",
        "pass_local_authority": "FRESH_VERIFIER_CURRENT_PROCESS_ONLY",
        "external_signature_anchor": {"status": "TBD_EXTERNAL", "gate": "BLOCKED_EXTERNAL"},
        "generated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
        "expires_at": (now + timedelta(seconds=TTL_SECONDS)).isoformat().replace("+00:00", "Z"),
        "run_nonce": nonce,
        "security_envelope": {
            "locked": True,
            "atomic_write": True,
            "fd_single_read": True,
            "minimal_environment": True,
            "stdout_stderr_persisted": "SANITIZED_DIGEST_ONLY",
        },
        "controlled_input_paths": list(INPUT_PATHS),
        "controlled_input_sha256": _sha256_bytes(
            json.dumps(controlled, sort_keys=True, separators=(",", ":")).encode("utf-8")
        ),
        "log_path": "docs/validation/P3_WP14_LOCAL_VERIFICATION.log",
        "log_sha256": _sha256_path(LOG),
        "runs": runs,
        "boundary": "This file and its log/trace are unsigned reproducibility caches and cannot independently grant PASS_LOCAL. Only the fresh verifier may report a current-process local result. All commands are local engineering checks; Compose is rendered only; no deployment, enterprise endpoint, or enterprise UAT is performed.",
    }
    _write_json(LEDGER, payload)
    trace = read_json_once(TRACE)
    assert isinstance(trace, dict)
    trace["release_manifest_path"] = "docs/phase3/P3_WP14_RELEASE_MANIFEST.json"
    trace["release_manifest_sha256"] = _sha256_path(MANIFEST)
    trace["verification_ledger_sha256"] = _sha256_path(LEDGER)
    trace["verification_log_sha256"] = _sha256_path(LOG)
    trace["local_verification_cache"] = {
        "classification": "UNSIGNED_REPRODUCIBILITY_CACHE",
        "pass_local_authority": "FRESH_VERIFIER_CURRENT_PROCESS_ONLY",
        "external_signature_status": "TBD_EXTERNAL",
        "enterprise_gate": "BLOCKED_EXTERNAL",
    }
    trace["second_isolated_local_environment"] = {
        "formal_same_signed_version_status": "BLOCKED_EXTERNAL",
        "formal_same_signed_version_reason": "manifest is unsigned; signature and enterprise second-environment authority are TBD_EXTERNAL",
        "local_portability_rehearsal": {
            "result": "PASS_LOCAL",
            "manifest_sha256": trace["release_manifest_sha256"],
            "manifest_signed": False,
            "boundary": "Two closed, non-secret local configuration renders are bound to this unsigned manifest digest. This is portability rehearsal only, not the required same signed-version second isolated environment.",
        },
    }
    _write_json(TRACE, trace)
    _write_json(STATE, {
        "schema_version": "1.0", "status": "COMPLETE", "run_nonce": nonce,
        "generated_at": payload["generated_at"], "expires_at": payload["expires_at"],
    })
    return 0


if __name__ == "__main__":
    sys.exit(main())