Rendezvous/eng/check_production_readiness.py

#!/usr/bin/env python3
"""Validate the redacted v1 readiness record and emit the release decision."""

from __future__ import annotations

import json
import pathlib
import re
import sys
from datetime import datetime, timedelta
from typing import Any


LOCAL_GATES = {
    "immutable-release-artifacts",
    "debug-and-release-verification",
    "real-consumer-pilots",
    "candidate-capacity-resilience",
    "production-process-recovery",
    "security-privacy-observability",
}
EXTERNAL_GATES = {
    "public-package-empty-cache-restore",
    "signed-publication",
    "source-preserving-udp-ingress",
    "same-lan-direct-canary",
    "home-nat-direct-canary",
    "restrictive-cgnat-typed-failure",
    "firewall-blocked-udp-typed-failure",
    "ipv6-direct-canary",
    "public-rate-shaped-capacity",
    "one-hour-candidate-endurance",
    "alert-delivery",
    "cold-standby-rollback-drill",
    "documentation-only-runbook-exercise",
}
STATUSES = {"pass", "pending", "fail"}
FORBIDDEN_KEY_PARTS = {
    "address",
    "credential",
    "endpoint",
    "listingid",
    "password",
    "playerid",
    "secret",
    "token",
    "userid",
}
UUID = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F-]{27,}\b")
IPV4 = re.compile(r"(?<![0-9])(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?![0-9])")
IPV6 = re.compile(
    r"(?i)(?:\b[0-9a-f]{0,4}:[0-9a-f:]*::[0-9a-f:]*\b|\b(?:[0-9a-f]{1,4}:){4,}[0-9a-f:]{1,39}\b)"
)
COMMIT = re.compile(r"[0-9a-f]{40}")
DIGEST = re.compile(r"[0-9a-f]{64}")


class InvalidRecord(ValueError):
    pass


def reject_sensitive(value: Any, path: str = "$") -> None:
    if isinstance(value, dict):
        for key, child in value.items():
            normalized = re.sub(r"[^a-z0-9]", "", key.lower())
            if any(part in normalized for part in FORBIDDEN_KEY_PARTS):
                raise InvalidRecord(f"{path}.{key} uses a forbidden sensitive-data key")
            reject_sensitive(child, f"{path}.{key}")
    elif isinstance(value, list):
        for index, child in enumerate(value):
            reject_sensitive(child, f"{path}[{index}]")
    elif isinstance(value, str):
        if UUID.search(value) or IPV4.search(value) or IPV6.search(value) \
            or "://" in value or "@" in value:
            raise InvalidRecord(f"{path} contains endpoint, identifier, or account-shaped data")


def evidence_path(repository_root: pathlib.Path, value: str, path: str) -> pathlib.Path:
    relative = pathlib.PurePosixPath(value)
    if relative.is_absolute() or ".." in relative.parts or not value:
        raise InvalidRecord(f"{path} must be a repository-relative reference")
    candidate = (repository_root / pathlib.Path(*relative.parts)).resolve()
    if not candidate.is_relative_to(repository_root.resolve()) or not candidate.is_file():
        raise InvalidRecord(f"{path} does not resolve to a repository evidence file")
    return candidate


def load_json(path: pathlib.Path, label: str) -> Any:
    try:
        with path.open("r", encoding="utf-8") as source:
            return json.load(source)
    except (OSError, json.JSONDecodeError) as error:
        raise InvalidRecord(f"{label} is not readable JSON: {error}") from error


def validate_gate_set(
    items: Any,
    expected: set[str],
    path: str,
    repository_root: pathlib.Path,
) -> list[dict[str, str]]:
    if not isinstance(items, list):
        raise InvalidRecord(f"{path} must be an array")
    gates: list[dict[str, str]] = []
    for index, item in enumerate(items):
        if not isinstance(item, dict) or set(item) != {"id", "status", "evidenceRef", "note"}:
            raise InvalidRecord(f"{path}[{index}] has an invalid shape")
        if not all(isinstance(item[key], str) for key in item):
            raise InvalidRecord(f"{path}[{index}] fields must be strings")
        if item["status"] not in STATUSES:
            raise InvalidRecord(f"{path}[{index}] has an invalid status")
        evidence_path(repository_root, item["evidenceRef"], f"{path}[{index}].evidenceRef")
        if len(item["note"]) > 240:
            raise InvalidRecord(f"{path}[{index}].note is too long")
        gates.append(item)
    identifiers = [gate["id"] for gate in gates]
    if len(identifiers) != len(set(identifiers)):
        raise InvalidRecord(f"{path} contains duplicate gate identifiers")
    if set(identifiers) != expected:
        missing = sorted(expected - set(identifiers))
        extra = sorted(set(identifiers) - expected)
        raise InvalidRecord(f"{path} gate mismatch; missing={missing}, extra={extra}")
    return gates


def validate_local_evidence(
    record: dict[str, Any],
    gates: list[dict[str, str]],
    repository_root: pathlib.Path,
) -> None:
    if any(gate["status"] != "pass" for gate in gates):
        return
    commit = record["evaluatedCommit"]
    release_path = evidence_path(
        repository_root,
        "docs/evidence/releases/v1.0.0-local-candidate.json",
        "local release evidence",
    )
    release = load_json(release_path, "local release evidence")
    if not isinstance(release, dict) or release.get("schemaVersion") != 1 \
        or release.get("kind") != "rendezvous-local-release-candidate" \
        or release.get("sourceCommit") != commit \
        or release.get("treeState") != "clean" \
        or release.get("result") != "pass":
        raise InvalidRecord("local release evidence is not a passing clean build of evaluatedCommit")
    verification = release.get("verification")
    if not isinstance(verification, dict):
        raise InvalidRecord("local release evidence has no verification object")
    exact_passes = {
        "lockedRestore": "pass",
        "format": "pass",
        "byteReproduciblePackages": "pass",
        "byteReproducibleServerArchive": "pass",
        "sbomChecksumsAndProvenance": "pass",
        "candidateConsumerFixtures": "pass",
        "realConsumerRestores": "pass",
    }
    if any(verification.get(key) != value for key, value in exact_passes.items()) \
        or verification.get("reportedVulnerabilities") != 0 \
        or verification.get("releaseBuildWarnings") != 0 \
        or verification.get("releaseBuildErrors") != 0 \
        or verification.get("debugTestsPassed", 0) < 300 \
        or verification.get("debugTestsFailed") != 0 \
        or verification.get("releaseTestsPassed", 0) < 300 \
        or verification.get("releaseTestsFailed") != 0 \
        or verification.get("selectedProductionFaultTestsPassed", 0) < 17:
        raise InvalidRecord("local release evidence does not satisfy every required verification")
    consumers = release.get("consumers")
    if not isinstance(consumers, list) or {
        item.get("name") for item in consumers if isinstance(item, dict)
    } != {"SpaceGame", "Unscouted"} or any(
        not isinstance(item, dict)
        or item.get("candidateRestore") != "pass"
        or item.get("directTrafficPilot") != "pass"
        for item in consumers
    ):
        raise InvalidRecord("local release evidence does not prove both required consumers")

    capacity_path = evidence_path(
        repository_root,
        "docs/evidence/capacity/v2/candidate-2cpu.json",
        "candidate capacity evidence",
    )
    capacity = load_json(capacity_path, "candidate capacity evidence")
    runtime = capacity.get("runtime") if isinstance(capacity, dict) else None
    state = capacity.get("state") if isinstance(capacity, dict) else None
    if not isinstance(runtime, dict) or not isinstance(state, dict) \
        or capacity.get("schemaVersion") != 2 \
        or capacity.get("profile") != "candidate" \
        or capacity.get("passed") is not True \
        or capacity.get("failures") != [] \
        or runtime.get("commitSha") != commit \
        or runtime.get("treeState") != "clean" \
        or runtime.get("processorCount") != 2 \
        or state.get("soakDurationSeconds", 0) < 300 \
        or state.get("finalListings") != 0 \
        or state.get("finalAttempts") != 0 \
        or state.get("finalReplayMarkers") != 0 \
        or state.get("restartStartedEmpty") is not True \
        or state.get("overloadWasTyped") is not True \
        or state.get("recoverySucceeded") is not True:
        raise InvalidRecord("candidate capacity evidence does not satisfy the clean evaluated commit")


def validate_external_attestations(
    record: dict[str, Any],
    gates: list[dict[str, str]],
    repository_root: pathlib.Path,
) -> None:
    for gate in gates:
        if gate["status"] != "pass":
            continue
        path = evidence_path(repository_root, gate["evidenceRef"], f"{gate['id']} evidence")
        attestation = load_json(path, f"{gate['id']} evidence")
        if not isinstance(attestation, dict) or set(attestation) != {
            "schemaVersion",
            "kind",
            "gateId",
            "candidateCommit",
            "result",
            "performedAtUtc",
            "artifactDigest",
            "evidenceLocation",
            "reviewerRole",
        }:
            raise InvalidRecord(f"{gate['id']} requires a complete external-gate attestation")
        reject_sensitive(attestation, f"external evidence {gate['id']}")
        if attestation["schemaVersion"] != 1 \
            or attestation["kind"] != "rendezvous-external-gate-attestation" \
            or attestation["gateId"] != gate["id"] \
            or attestation["candidateCommit"] != record["evaluatedCommit"] \
            or attestation["result"] != "pass" \
            or not isinstance(attestation["artifactDigest"], str) \
            or not DIGEST.fullmatch(attestation["artifactDigest"]) \
            or attestation["evidenceLocation"] not in {
                "protected-operations-record",
                "public-release-record",
            } \
            or attestation["reviewerRole"] not in {
                "release-operator",
                "network-operator",
                "security-operator",
                "independent-operator",
            }:
            raise InvalidRecord(f"{gate['id']} external attestation does not match the candidate gate")
        try:
            performed = datetime.fromisoformat(attestation["performedAtUtc"].replace("Z", "+00:00"))
        except (AttributeError, ValueError) as error:
            raise InvalidRecord(f"{gate['id']} has an invalid performedAtUtc") from error
        if performed.tzinfo is None or performed.utcoffset() != timedelta(0):
            raise InvalidRecord(f"{gate['id']} performedAtUtc must be UTC")


def validate(record: Any, repository_root: pathlib.Path) -> tuple[bool, list[str]]:
    if not isinstance(record, dict) or set(record) != {
        "schemaVersion",
        "kind",
        "evaluatedCommit",
        "decision",
        "localGates",
        "externalGates",
    }:
        raise InvalidRecord("The top-level readiness record shape is invalid")
    if record["schemaVersion"] != 1 or record["kind"] != "rendezvous-production-readiness":
        raise InvalidRecord("The readiness schema identity is invalid")
    if not isinstance(record["evaluatedCommit"], str) or not COMMIT.fullmatch(record["evaluatedCommit"]):
        raise InvalidRecord("evaluatedCommit must be a full lowercase Git commit")
    reject_sensitive(record)
    local_gates = validate_gate_set(
        record["localGates"], LOCAL_GATES, "$.localGates", repository_root
    )
    external_gates = validate_gate_set(
        record["externalGates"], EXTERNAL_GATES, "$.externalGates", repository_root
    )
    validate_local_evidence(record, local_gates, repository_root)
    validate_external_attestations(record, external_gates, repository_root)
    gates = local_gates + external_gates
    blockers = sorted(gate["id"] for gate in gates if gate["status"] != "pass")
    ready = not blockers
    expected_decision = "ready" if ready else "not-ready"
    if record["decision"] != expected_decision:
        raise InvalidRecord(
            f"decision must be {expected_decision!r} for the recorded gate statuses"
        )
    return ready, blockers


def main() -> int:
    if len(sys.argv) != 2:
        print("usage: check_production_readiness.py RECORD", file=sys.stderr)
        return 2
    try:
        with open(sys.argv[1], "r", encoding="utf-8") as source:
            record = json.load(source)
        ready, blockers = validate(record, pathlib.Path(__file__).resolve().parent.parent)
    except (OSError, json.JSONDecodeError, InvalidRecord) as error:
        print(f"INVALID: {error}", file=sys.stderr)
        return 2
    if not ready:
        print(f"NOT READY: {len(blockers)} required gate(s) are not passing.")
        for blocker in blockers:
            print(f"- {blocker}")
        return 3
    print("READY: every required v1 production gate is recorded as passing.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())