#!/usr/bin/env python3
"""Offline evaluator for bulla.attestation-boundary/0.1-experimental."""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import importlib.metadata
import importlib.util
import itertools
import json
from pathlib import Path
import re
import sys
from typing import Any, Mapping
import xml.etree.ElementTree as ET

sys.dont_write_bytecode = True


ROOT = Path(__file__).resolve().parent
VENDOR_LOCK = ROOT / "vendor-lock.json"
CONTEXT = ROOT / "context.json"
REPORT = ROOT / "generated/report.json"
PROFILE = "bulla.attestation-boundary/0.1-experimental"
REPORT_PROFILE = "bulla.attestation-boundary-report/0.1-experimental"
EXPECTED_SOURCE_LOCK_ROOT = "sha256:80dc5f6daa5a7445c18001825412281fab80cd8fbff17c44b18ae1fecbec1bb2"
EXPECTED_RUNTIME_DEPENDENCIES = {
    "PyNaCl": "1.6.2",
    "cbor2": "5.6.5",
    "cryptography": "43.0.1",
    "ecdsa": "0.18.0",
    "nvidia-ml-py": "12.550.52",
}
SUBCLAIMS = (
    "device_identity_and_state",
    "measured_workload",
    "model_commitment",
    "request_and_nonce_commitment",
    "response_commitment",
    "execution_relation",
)
STATES = {"ESTABLISHED", "NOT_ESTABLISHED", "UNCHECKABLE"}
RESULT_FIELDS = {
    "state", "reason_codes", "verification_depth", "accepted_roots",
    "accepted_policy_sha256", "evidence_sha256",
}


class AttestationBoundaryError(ValueError):
    pass


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


def digest(raw: bytes) -> str:
    return "sha256:" + hashlib.sha256(raw).hexdigest()


def strict_json(path: Path) -> Any:
    def pairs(items: list[tuple[str, Any]]) -> dict[str, Any]:
        result: dict[str, Any] = {}
        for key, value in items:
            if key in result:
                raise AttestationBoundaryError(f"duplicate JSON member {key} in {path}")
            result[key] = value
        return result

    try:
        return json.loads(
            path.read_text(encoding="utf-8"),
            object_pairs_hook=pairs,
            parse_float=lambda _: (_ for _ in ()).throw(AttestationBoundaryError("floats are forbidden")),
            parse_constant=lambda _: (_ for _ in ()).throw(AttestationBoundaryError("non-finite numbers are forbidden")),
        )
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise AttestationBoundaryError(f"invalid JSON at {path}: {exc}") from exc


def exact(value: Mapping[str, Any], fields: set[str], label: str) -> None:
    if set(value) != fields:
        raise AttestationBoundaryError(f"{label} fields changed")


def string_list(value: Any, label: str, *, nonempty: bool = False) -> None:
    if not isinstance(value, list) or (nonempty and not value):
        raise AttestationBoundaryError(f"{label} must be a{' nonempty' if nonempty else ''} string array")
    if any(not isinstance(item, str) or not item for item in value):
        raise AttestationBoundaryError(f"{label} contains an invalid string")


def sha256_value(value: Any, label: str) -> None:
    if not isinstance(value, str) or re.fullmatch(r"sha256:[0-9a-f]{64}", value) is None:
        raise AttestationBoundaryError(f"{label} is not a canonical SHA-256 digest")


def nonempty_string(value: Any, label: str) -> None:
    if not isinstance(value, str) or not value:
        raise AttestationBoundaryError(f"{label} must be a nonempty string")


def lowercase_hex(value: Any, label: str, *, length: int | None = None) -> None:
    pattern = rf"[0-9a-f]{{{length}}}" if length is not None else r"[0-9a-f]+"
    if not isinstance(value, str) or re.fullmatch(pattern, value) is None:
        raise AttestationBoundaryError(f"{label} must be canonical lowercase hexadecimal")


def safe_integer(value: Any, label: str) -> None:
    if type(value) is not int or not 0 <= value <= 9_007_199_254_740_991:
        raise AttestationBoundaryError(f"{label} must be a nonnegative safe integer")


def check_context(context: Mapping[str, Any]) -> None:
    exact(context, {"profile", "air", "nvidia", "live_h100_collection", "never_infer"}, "attestation context")
    if context["profile"] != "bulla.attestation-boundary-context/0.1-experimental":
        raise AttestationBoundaryError("attestation context profile changed")
    air = context["air"]
    nvidia = context["nvidia"]
    if not isinstance(air, Mapping) or not isinstance(nvidia, Mapping):
        raise AttestationBoundaryError("attestation contexts must be objects")
    exact(air, {
        "accepted_vector", "accepted_public_key_hex", "expected_model_hash_hex", "expected_nonce_hex",
        "expected_platform", "frozen_now_epoch",
    }, "AIR context")
    exact(nvidia, {
        "accepted_device_root_der_sha256", "evaluation_time", "expected_report_format", "expected_driver_version",
        "expected_measurement_count", "expected_sample_nonce_hex", "expected_vbios_version", "recorded_state_semantics",
    }, "NVIDIA context")
    accepted_vector = air["accepted_vector"]
    if (
        not isinstance(accepted_vector, str)
        or re.fullmatch(r"vendor/air/vectors/valid/[a-z0-9.-]+\.json", accepted_vector) is None
        or ".." in Path(accepted_vector).parts
    ):
        raise AttestationBoundaryError("AIR accepted vector path is not canonical")
    lowercase_hex(air["accepted_public_key_hex"], "AIR accepted public key", length=64)
    lowercase_hex(air["expected_model_hash_hex"], "AIR expected model hash", length=64)
    lowercase_hex(air["expected_nonce_hex"], "AIR expected nonce")
    nonempty_string(air["expected_platform"], "AIR expected platform")
    safe_integer(air["frozen_now_epoch"], "AIR frozen evaluation time")
    sha256_value(nvidia["accepted_device_root_der_sha256"], "NVIDIA accepted device root")
    evaluation_time = nvidia["evaluation_time"]
    if not isinstance(evaluation_time, str) or re.fullmatch(
        r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z", evaluation_time
    ) is None:
        raise AttestationBoundaryError("NVIDIA evaluation time must use canonical UTC Z form")
    try:
        dt.datetime.fromisoformat(evaluation_time.replace("Z", "+00:00"))
    except ValueError as exc:
        raise AttestationBoundaryError("NVIDIA evaluation time is invalid") from exc
    nonempty_string(nvidia["expected_driver_version"], "NVIDIA expected driver version")
    safe_integer(nvidia["expected_measurement_count"], "NVIDIA expected measurement count")
    lowercase_hex(nvidia["expected_sample_nonce_hex"], "NVIDIA expected sample nonce", length=64)
    nonempty_string(nvidia["expected_vbios_version"], "NVIDIA expected VBIOS version")
    if nvidia["recorded_state_semantics"] != "SIGNED_REPORT_CONTENTS_NOT_RIM_CONFORMITY":
        raise AttestationBoundaryError("NVIDIA recorded-state boundary changed")
    if nvidia["expected_report_format"] != "HOPPER_ATTESTATION_REPORT":
        raise AttestationBoundaryError("NVIDIA accepted report parser profile changed")
    if context["live_h100_collection"] != "UNCHECKABLE":
        raise AttestationBoundaryError("live H100 collection boundary changed")
    string_list(context["never_infer"], "context never_infer", nonempty=True)


def check_vendor_lock() -> Mapping[str, Any]:
    lock = strict_json(VENDOR_LOCK)
    if not isinstance(lock, Mapping):
        raise AttestationBoundaryError("vendor lock must be an object")
    exact(lock, {"profile", "sources", "runtime_dependencies", "members", "lock_root"}, "vendor lock")
    if lock["profile"] != "bulla.attestation-boundary-source-lock/1":
        raise AttestationBoundaryError("vendor lock profile changed")
    if lock["runtime_dependencies"] != EXPECTED_RUNTIME_DEPENDENCIES:
        raise AttestationBoundaryError("accepted verifier dependency versions changed")
    for distribution, expected_version in EXPECTED_RUNTIME_DEPENDENCIES.items():
        try:
            actual_version = importlib.metadata.version(distribution)
        except importlib.metadata.PackageNotFoundError as exc:
            raise AttestationBoundaryError(f"required verifier dependency is absent: {distribution}") from exc
        if actual_version != expected_version:
            raise AttestationBoundaryError(
                f"verifier dependency changed: {distribution} {actual_version} != {expected_version}"
            )
    body = {key: value for key, value in lock.items() if key != "lock_root"}
    if lock["lock_root"] != digest(canonical(body)):
        raise AttestationBoundaryError("vendor lock root changed")
    if lock["lock_root"] != EXPECTED_SOURCE_LOCK_ROOT:
        raise AttestationBoundaryError("vendor lock does not match the receiver-accepted source closure")
    sources = lock["sources"]
    if sources != {
        "air": {"commit": "e3bacb3076338db4f9644e469410a6ffea11c303", "release": "v1.0.1"},
        "nvtrust": {"commit": "0c5d627313037c1e577d05a232e79394a41b2c21"},
    }:
        raise AttestationBoundaryError("accepted upstream source commits changed")
    members = lock["members"]
    if not isinstance(members, list) or not members:
        raise AttestationBoundaryError("vendor lock members must be a nonempty array")
    declared: list[str] = []
    for index, member in enumerate(members):
        if not isinstance(member, Mapping):
            raise AttestationBoundaryError(f"vendor member {index} must be an object")
        exact(member, {
            "path", "source_repository", "source_commit", "source_path", "license", "byte_length", "sha256"
        }, f"vendor member {index}")
        path = member["path"]
        if not isinstance(path, str) or not path.startswith("vendor/") or ".." in Path(path).parts:
            raise AttestationBoundaryError("vendor path is not canonical")
        if declared and path <= declared[-1]:
            raise AttestationBoundaryError("vendor paths are not strict ASCII order")
        declared.append(path)
        if path.startswith("vendor/air/"):
            expected_source = {
                "source_repository": "https://github.com/cyntrisec/air-v1",
                "source_commit": sources["air"]["commit"],
                "source_path": path.removeprefix("vendor/air/"),
                "license": "CC-BY-4.0",
            }
        elif path.startswith("vendor/nvtrust/"):
            expected_source = {
                "source_repository": "https://github.com/NVIDIA/nvtrust",
                "source_commit": sources["nvtrust"]["commit"],
                "source_path": path.removeprefix("vendor/nvtrust/"),
                "license": "BSD-3-Clause-or-bundled-notices" if "/local_gpu_verifier/" in path else "Apache-2.0",
            }
        else:
            raise AttestationBoundaryError("vendor path has no accepted source jurisdiction")
        if any(member[key] != value for key, value in expected_source.items()):
            raise AttestationBoundaryError(f"vendor source attribution changed: {path}")
        target = ROOT / path
        if target.is_symlink() or not target.is_file():
            raise AttestationBoundaryError(f"vendor member is missing or non-regular: {path}")
        raw = target.read_bytes()
        if type(member["byte_length"]) is not int or member["byte_length"] != len(raw):
            raise AttestationBoundaryError(f"vendor member length changed: {path}")
        if member["sha256"] != digest(raw):
            raise AttestationBoundaryError(f"vendor member digest changed: {path}")
    actual = sorted(
        path.relative_to(ROOT).as_posix()
        for path in (ROOT / "vendor").rglob("*")
        if path.is_file() or path.is_symlink()
    )
    if actual != declared:
        raise AttestationBoundaryError("vendor member closure changed")
    return lock


def result(
    state: str,
    reason_codes: list[str],
    verification_depth: list[str],
    accepted_roots: list[str],
    accepted_policy_sha256: str,
    evidence_sha256: list[str],
) -> dict[str, Any]:
    if state not in STATES or not reason_codes:
        raise AttestationBoundaryError("subclaim result is not closed")
    return {
        "state": state,
        "reason_codes": reason_codes,
        "verification_depth": verification_depth,
        "accepted_roots": accepted_roots,
        "accepted_policy_sha256": accepted_policy_sha256,
        "evidence_sha256": evidence_sha256,
    }


def load_air_module():
    path = ROOT / "vendor/air/scripts/interop_test.py"
    spec = importlib.util.spec_from_file_location("attestation_boundary_air", path)
    if spec is None or spec.loader is None:
        raise AttestationBoundaryError("cannot load the pinned AIR verifier")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def evaluate_air(context: Mapping[str, Any], policy_digest: str) -> tuple[dict[str, Any], dict[str, Any]]:
    air = load_air_module()
    vector_root = ROOT / "vendor/air/vectors"
    vector_results: list[dict[str, Any]] = []
    for path in sorted((vector_root / "valid").glob("*.json")) + sorted((vector_root / "invalid").glob("*.json")):
        vector = strict_json(path)
        verified = air.verify_vector(dict(vector), context["frozen_now_epoch"])
        if "expected_failure" in vector:
            failure = verified.failure
            expected = vector["expected_failure"]
            passed = (
                not verified.ok
                and failure is not None
                and failure.layer == expected["layer"]
                and failure.check == expected["check"]
                and failure.code == expected["code"]
            )
        else:
            passed = verified.ok
        if not passed:
            raise AttestationBoundaryError(f"pinned AIR vector outcome changed: {path.name}")
        vector_results.append({"name": vector["name"], "expected_outcome_reproduced": True})

    accepted_path = ROOT / context["accepted_vector"]
    accepted = strict_json(accepted_path)
    if accepted["public_key_hex"] != context["accepted_public_key_hex"]:
        raise AttestationBoundaryError("AIR accepted key changed")
    policy_vector = dict(accepted)
    policy_vector["verify_policy"] = {
        "expected_model_hash_hex": context["expected_model_hash_hex"],
        "expected_nonce_hex": context["expected_nonce_hex"],
        "expected_platform": context["expected_platform"],
        "require_nonce": True,
    }
    verification = air.verify_vector(policy_vector, context["frozen_now_epoch"])
    if not verification.ok:
        raise AttestationBoundaryError("AIR accepted vector failed the receiver policy")
    protected, _, payload, signature = air.parse_receipt_cose(bytes.fromhex(accepted["receipt_hex"]))
    claims = air.decode_and_validate_claims(payload)
    air.verify_sig_structure(protected, payload, signature, bytes.fromhex(context["accepted_public_key_hex"]))
    evidence = [digest(accepted_path.read_bytes())]
    root = digest(bytes.fromhex(context["accepted_public_key_hex"]))
    signed_depth = ["COSE_SIGN1", "ED25519", "AIR_V1_CLAIMS", "RECEIVER_PINNED_POLICY"]
    signed = lambda reasons: result("ESTABLISHED", reasons, signed_depth, [root], policy_digest, evidence)
    not_execution = lambda reasons: result("NOT_ESTABLISHED", reasons, signed_depth, [root], policy_digest, evidence)
    subclaims = {
        "device_identity_and_state": not_execution([
            "SIGNED_PLATFORM_AND_MEASUREMENT_CLAIMS_BOUND",
            "SIGNER_KEY_NOT_BOUND_TO_DEVICE_IDENTITY",
        ]),
        "measured_workload": not_execution(["PLATFORM_MEASUREMENTS_DO_NOT_BIND_WORKLOAD_EXECUTION"]),
        "model_commitment": signed(["SIGNED_MODEL_COMMITMENT_BOUND"]),
        "request_and_nonce_commitment": signed(["SIGNED_REQUEST_AND_NONCE_COMMITMENTS_BOUND"]),
        "response_commitment": signed(["SIGNED_RESPONSE_COMMITMENT_BOUND"]),
        "execution_relation": not_execution(["NO_VERIFIED_INFERENCE_RELATION"]),
    }
    details = {
        "accepted_vector": accepted_path.relative_to(ROOT).as_posix(),
        "accepted_key_sha256": root,
        "model_hash_hex": bytes(claims[air.AIR_MODEL_HASH]).hex(),
        "request_hash_hex": bytes(claims[air.AIR_REQUEST_HASH]).hex(),
        "response_hash_hex": bytes(claims[air.AIR_RESPONSE_HASH]).hex(),
        "nonce_hex": bytes(claims[air.EAT_NONCE]).hex(),
        "platform_claim": claims[air.AIR_ENCLAVE_MEASUREMENTS]["measurement_type"],
        "vectors": vector_results,
    }
    return subclaims, details


def verify_certificate_signature(certificate: Any, issuer: Any) -> None:
    from cryptography.hazmat.primitives.asymmetric import ec
    try:
        issuer.public_key().verify(
            certificate.signature,
            certificate.tbs_certificate_bytes,
            ec.ECDSA(certificate.signature_hash_algorithm),
        )
    except Exception as exc:
        raise AttestationBoundaryError("NVIDIA certificate signature verification failed") from exc


def rim_measurements(path: Path, runtime: list[str]) -> dict[str, Any]:
    try:
        root = ET.fromstring(path.read_bytes())
    except ET.ParseError as exc:
        raise AttestationBoundaryError(f"pinned RIM XML is invalid: {path.name}") from exc
    metadata = next((element for element in root.iter() if element.tag.endswith("Meta")), None)
    if metadata is None:
        raise AttestationBoundaryError(f"pinned RIM has no metadata: {path.name}")
    active: list[int] = []
    mismatched: list[int] = []
    namespace = "{http://www.w3.org/2001/04/xmlenc#sha384}"
    for element in root.iter():
        if element.attrib.get("type") != "Measurement" or element.attrib.get("active") != "True":
            continue
        index = int(element.attrib["index"])
        alternatives = int(element.attrib["alternatives"])
        accepted = [element.attrib[f"{namespace}Hash{offset}"] for offset in range(alternatives)]
        active.append(index)
        if index >= len(runtime) or runtime[index] not in accepted:
            mismatched.append(index)
    return {
        "sha256": digest(path.read_bytes()),
        "declared_version": metadata.attrib.get("colloquialVersion"),
        "active_measurements": active,
        "mismatched_measurements": mismatched,
    }


def evaluate_nvidia(context: Mapping[str, Any], policy_digest: str) -> tuple[dict[str, Any], dict[str, Any]]:
    from cryptography import x509
    from cryptography.hazmat.primitives import serialization

    local = ROOT / "vendor/nvtrust/guest_tools/gpu_verifiers/local_gpu_verifier"
    source = local / "src"
    sys.path.insert(0, str(source))
    try:
        from verifier.attestation import AttestationReport
        from verifier.config import HopperSettings
        from verifier.utils import format_vbios_version
    finally:
        sys.path.pop(0)

    sample = local / "src/verifier/samples"
    certs = local / "src/verifier/certs"
    report_path = sample / "hopperAttestationReport.txt"
    chain_path = sample / "hopperCertChain.txt"
    root_path = certs / "verifier_device_root.pem"
    report_raw = bytes.fromhex(report_path.read_text(encoding="ascii"))
    report = AttestationReport(report_raw, HopperSettings())
    blocks = re.findall(
        b"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----",
        chain_path.read_bytes(),
        re.DOTALL,
    )
    chain = [x509.load_pem_x509_certificate(block) for block in blocks]
    accepted_root = x509.load_pem_x509_certificate(root_path.read_bytes())
    if len(chain) != 5 or chain[-1].public_bytes(serialization.Encoding.DER) != accepted_root.public_bytes(serialization.Encoding.DER):
        raise AttestationBoundaryError("NVIDIA device chain does not terminate at the accepted root")
    root_digest = digest(accepted_root.public_bytes(serialization.Encoding.DER))
    if root_digest != context["accepted_device_root_der_sha256"]:
        raise AttestationBoundaryError("NVIDIA accepted device root changed")
    evaluation_time = dt.datetime.fromisoformat(context["evaluation_time"].replace("Z", "+00:00"))
    for index, certificate in enumerate(chain):
        issuer = chain[index + 1] if index + 1 < len(chain) else certificate
        if certificate.issuer != issuer.subject:
            raise AttestationBoundaryError("NVIDIA certificate issuer chain changed")
        if not (certificate.not_valid_before_utc <= evaluation_time <= certificate.not_valid_after_utc):
            raise AttestationBoundaryError("NVIDIA device certificate is outside the accepted evaluation time")
        verify_certificate_signature(certificate, issuer)
    if not report.verify_signature(chain[0], HopperSettings.signature_length, hashlib.sha384):
        raise AttestationBoundaryError("NVIDIA sample report signature failed")
    request_nonce = report.get_request_message().get_nonce().hex()
    if request_nonce != context["expected_sample_nonce_hex"]:
        raise AttestationBoundaryError("NVIDIA sample nonce changed")
    opaque = report.get_response_message().get_opaque_data()
    driver = opaque.get_data("OPAQUE_FIELD_ID_DRIVER_VERSION").decode().rstrip("\0")
    vbios = format_vbios_version(opaque.get_data("OPAQUE_FIELD_ID_VBIOS_VERSION"))
    if driver != context["expected_driver_version"] or vbios != context["expected_vbios_version"]:
        raise AttestationBoundaryError("NVIDIA recorded driver or VBIOS version changed")
    runtime = report.get_measurements()
    if len(runtime) != context["expected_measurement_count"]:
        raise AttestationBoundaryError("NVIDIA measurement count changed")
    driver_rim = rim_measurements(sample / "Driver_RIM_test_no_gpu.swidtag", runtime)
    vbios_rim = rim_measurements(sample / "1010_0200_882_96005E0001_test_no_gpu.swidtag", runtime)
    evidence = [
        digest(report_path.read_bytes()),
        digest(chain_path.read_bytes()),
        driver_rim["sha256"],
        vbios_rim["sha256"],
    ]
    signed_depth = ["PINNED_DEVICE_CERTIFICATE_CHAIN", "SPDM_REPORT_SIGNATURE", "RECORDED_SAMPLE_NONCE", "MEASUREMENT_RECORD_PARSE"]
    subclaims = {
        "device_identity_and_state": result(
            "ESTABLISHED",
            ["PINNED_DEVICE_CHAIN_VALID", "RECORDED_REPORT_SIGNATURE_VALID", "SIGNED_RECORDED_STATE_BOUND"],
            signed_depth,
            [root_digest],
            policy_digest,
            evidence,
        ),
        "measured_workload": result(
            "NOT_ESTABLISHED",
            ["RIM_POLICY_MISMATCH", "FIRMWARE_MEASUREMENTS_DO_NOT_BIND_WORKLOAD_EXECUTION"],
            signed_depth,
            [root_digest],
            policy_digest,
            evidence,
        ),
        "model_commitment": result("NOT_ESTABLISHED", ["MODEL_COMMITMENT_ABSENT"], signed_depth, [root_digest], policy_digest, evidence),
        "request_and_nonce_commitment": result("NOT_ESTABLISHED", ["REQUEST_COMMITMENT_ABSENT", "SAMPLE_NONCE_NOT_LIVE_CHALLENGE"], signed_depth, [root_digest], policy_digest, evidence),
        "response_commitment": result("NOT_ESTABLISHED", ["RESPONSE_COMMITMENT_ABSENT"], signed_depth, [root_digest], policy_digest, evidence),
        "execution_relation": result("NOT_ESTABLISHED", ["MODEL_REQUEST_RESPONSE_RELATION_ABSENT"], signed_depth, [root_digest], policy_digest, evidence),
    }
    details = {
        "report_format": context["expected_report_format"],
        "report_format_jurisdiction": "PINNED_PARSER_PROFILE_NOT_SIGNED_REPORT_FIELD",
        "device_root_der_sha256": root_digest,
        "report_sha256": digest(report_path.read_bytes()),
        "certificate_chain_sha256": digest(chain_path.read_bytes()),
        "sample_nonce_hex": request_nonce,
        "driver_version": driver,
        "vbios_version": vbios,
        "measurement_count": len(runtime),
        "measurements_sha256": digest(canonical(runtime)),
        "driver_rim": driver_rim,
        "vbios_rim": vbios_rim,
        "rim_signature_evaluation": "NOT_PERFORMED",
    }
    return subclaims, details


def minimal_sets(sets: list[frozenset[str]]) -> list[list[str]]:
    unique = sorted(set(sets), key=lambda item: (len(item), tuple(sorted(item))))
    minimal = [item for item in unique if not any(other < item for other in unique)]
    return [sorted(item) for item in minimal]


def acceptance_cuts(nodes: list[Mapping[str, Any]], edges: list[Mapping[str, str]]) -> list[dict[str, Any]]:
    by_id = {node["id"]: node for node in nodes}
    dependencies: dict[str, list[str]] = {}
    for edge in edges:
        dependencies.setdefault(edge["to"], []).append(edge["from"])

    def paths(node_id: str) -> list[frozenset[str]]:
        node = by_id[node_id]
        if node["kind"] == "CONTROL":
            return [frozenset({node_id})]
        if node["state"] != "ESTABLISHED":
            return []
        child_sets = [paths(child) for child in dependencies.get(node_id, [])]
        if not child_sets or any(not choices for choices in child_sets):
            return []
        return [frozenset().union(*choice) for choice in itertools.product(*child_sets)]

    results: list[dict[str, Any]] = []
    for node in nodes:
        if node["kind"] == "CONCLUSION":
            results.append({"conclusion": node["id"], "state": node["state"], "cuts": minimal_sets(paths(node["id"]))})
    return results


def dependency_graph(air: Mapping[str, Any], nvidia: Mapping[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    controls = {
        "air.accepted_key", "air.accepted_policy", "air.vector_bytes", "air.verifier_source",
        "air.runtime_dependency_versions", "nvidia.certificate_chain", "nvidia.device_root",
        "nvidia.accepted_policy", "nvidia.report_bytes", "nvidia.report_parser_source", "nvidia.runtime_dependency_versions",
    }
    nodes: list[dict[str, Any]] = [{"id": item, "kind": "CONTROL"} for item in sorted(controls)]
    edges: list[dict[str, str]] = []
    dependencies = {
        "air": [
            "air.accepted_key", "air.accepted_policy", "air.vector_bytes", "air.verifier_source",
            "air.runtime_dependency_versions",
        ],
        "nvidia": [
            "nvidia.certificate_chain", "nvidia.device_root", "nvidia.accepted_policy", "nvidia.report_bytes",
            "nvidia.report_parser_source", "nvidia.runtime_dependency_versions",
        ],
    }
    for artifact, subclaims in (("air", air), ("nvidia", nvidia)):
        for name in SUBCLAIMS:
            conclusion = f"{artifact}.{name}"
            nodes.append({"id": conclusion, "kind": "CONCLUSION", "state": subclaims[name]["state"]})
            for control in dependencies[artifact]:
                edges.append({"from": control, "to": conclusion})
    nodes.sort(key=lambda item: item["id"])
    edges.sort(key=lambda item: (item["to"], item["from"]))
    graph = {"logic": "ALL_DECLARED_CONTROLS_REQUIRED", "nodes": nodes, "edges": edges}
    return graph, acceptance_cuts(nodes, edges)


def build_report() -> dict[str, Any]:
    lock = check_vendor_lock()
    context = strict_json(CONTEXT)
    if not isinstance(context, Mapping):
        raise AttestationBoundaryError("attestation context must be an object")
    check_context(context)
    policy_digest = digest(canonical(context))
    air, air_details = evaluate_air(context["air"], policy_digest)
    nvidia, nvidia_details = evaluate_nvidia(context["nvidia"], policy_digest)
    for artifact in (air, nvidia):
        if set(artifact) != set(SUBCLAIMS):
            raise AttestationBoundaryError("attestation subclaim closure changed")
    graph, cuts = dependency_graph(air, nvidia)
    report: dict[str, Any] = {
        "profile": REPORT_PROFILE,
        "source_profile": PROFILE,
        "source_lock_root": lock["lock_root"],
        "accepted_policy_sha256": policy_digest,
        "upstream_commits": {
            "air": lock["sources"]["air"]["commit"],
            "nvtrust": lock["sources"]["nvtrust"]["commit"],
        },
        "artifacts": {
            "air": {"subclaims": air, "details": air_details},
            "nvidia": {"subclaims": nvidia, "details": nvidia_details},
        },
        "combined_execution_relation": result(
            "NOT_ESTABLISHED",
            ["NO_AUTHENTICATED_SHARED_EVENT_BINDING", "NO_VERIFIED_INFERENCE_RELATION"],
            ["SEPARATE_AIR_AND_NVIDIA_VERIFICATION"],
            [],
            policy_digest,
            sorted(set(air["execution_relation"]["evidence_sha256"] + nvidia["execution_relation"]["evidence_sha256"])),
        ),
        "live_h100_collection": result(
            "UNCHECKABLE",
            ["LIVE_H100_EVIDENCE_NOT_SUPPLIED"],
            [],
            [],
            policy_digest,
            [],
        ),
        "control_dependency_graph": graph,
        "acceptance_cuts": cuts,
        "never_infer": context["never_infer"],
    }
    report["report_sha256"] = digest(canonical(report))
    return report


def check_report(report: Mapping[str, Any]) -> None:
    exact(report, {
        "profile", "source_profile", "source_lock_root", "accepted_policy_sha256", "upstream_commits",
        "artifacts", "combined_execution_relation", "live_h100_collection", "control_dependency_graph",
        "acceptance_cuts", "never_infer", "report_sha256",
    }, "attestation boundary report")
    if report["profile"] != REPORT_PROFILE or report["source_profile"] != PROFILE:
        raise AttestationBoundaryError("attestation boundary report profile changed")
    if report["source_lock_root"] != EXPECTED_SOURCE_LOCK_ROOT:
        raise AttestationBoundaryError("attestation boundary source lock changed")
    if report["upstream_commits"] != {
        "air": "e3bacb3076338db4f9644e469410a6ffea11c303",
        "nvtrust": "0c5d627313037c1e577d05a232e79394a41b2c21",
    }:
        raise AttestationBoundaryError("attestation boundary upstream commits changed")
    sha256_value(report["accepted_policy_sha256"], "accepted policy")
    sha256_value(report["report_sha256"], "report")
    body = {key: value for key, value in report.items() if key != "report_sha256"}
    if report["report_sha256"] != digest(canonical(body)):
        raise AttestationBoundaryError("attestation boundary report digest changed")
    artifacts = report["artifacts"]
    if not isinstance(artifacts, Mapping) or set(artifacts) != {"air", "nvidia"}:
        raise AttestationBoundaryError("attestation artifact set changed")
    for artifact_name in ("air", "nvidia"):
        artifact = artifacts[artifact_name]
        if not isinstance(artifact, Mapping):
            raise AttestationBoundaryError(f"{artifact_name} artifact must be an object")
        exact(artifact, {"subclaims", "details"}, f"{artifact_name} artifact")
        if not isinstance(artifact["subclaims"], Mapping) or set(artifact["subclaims"]) != set(SUBCLAIMS):
            raise AttestationBoundaryError(f"{artifact_name} subclaim closure changed")
        for name, claim in artifact["subclaims"].items():
            if not isinstance(claim, Mapping):
                raise AttestationBoundaryError(f"{artifact_name}.{name} must be an object")
            exact(claim, RESULT_FIELDS, f"{artifact_name}.{name}")
            if claim["state"] not in STATES:
                raise AttestationBoundaryError(f"{artifact_name}.{name} state changed")
            string_list(claim["reason_codes"], f"{artifact_name}.{name}.reason_codes", nonempty=True)
            for field in ("verification_depth", "accepted_roots", "evidence_sha256"):
                string_list(claim[field], f"{artifact_name}.{name}.{field}")
            for field in ("accepted_roots", "evidence_sha256"):
                for value in claim[field]:
                    sha256_value(value, f"{artifact_name}.{name}.{field}")
            if claim["accepted_policy_sha256"] != report["accepted_policy_sha256"]:
                raise AttestationBoundaryError(f"{artifact_name}.{name} policy binding changed")
        details = artifact["details"]
        if not isinstance(details, Mapping):
            raise AttestationBoundaryError(f"{artifact_name} details must be an object")
        if artifact_name == "air":
            exact(details, {
                "accepted_vector", "accepted_key_sha256", "model_hash_hex", "request_hash_hex",
                "response_hash_hex", "nonce_hex", "platform_claim", "vectors",
            }, "AIR details")
            if not isinstance(details["vectors"], list) or len(details["vectors"]) != 10:
                raise AttestationBoundaryError("AIR vector result closure changed")
            for vector in details["vectors"]:
                if not isinstance(vector, Mapping):
                    raise AttestationBoundaryError("AIR vector result must be an object")
                exact(vector, {"name", "expected_outcome_reproduced"}, "AIR vector result")
                if not isinstance(vector["name"], str) or vector["expected_outcome_reproduced"] is not True:
                    raise AttestationBoundaryError("AIR vector result changed")
            sha256_value(details["accepted_key_sha256"], "AIR accepted key")
            nonempty_string(details["accepted_vector"], "AIR accepted vector")
            for field in ("model_hash_hex", "request_hash_hex", "response_hash_hex", "nonce_hex"):
                lowercase_hex(details[field], f"AIR {field}")
            nonempty_string(details["platform_claim"], "AIR platform claim")
        else:
            exact(details, {
                "report_format", "report_format_jurisdiction", "device_root_der_sha256", "report_sha256", "certificate_chain_sha256",
                "sample_nonce_hex", "driver_version", "vbios_version", "measurement_count",
                "measurements_sha256", "driver_rim", "vbios_rim", "rim_signature_evaluation",
            }, "NVIDIA details")
            if type(details["measurement_count"]) is not int:
                raise AttestationBoundaryError("NVIDIA measurement count must be an exact integer")
            safe_integer(details["measurement_count"], "NVIDIA measurement count")
            if details["report_format"] != "HOPPER_ATTESTATION_REPORT" or details["report_format_jurisdiction"] != "PINNED_PARSER_PROFILE_NOT_SIGNED_REPORT_FIELD":
                raise AttestationBoundaryError("NVIDIA report-format jurisdiction changed")
            for field in ("device_root_der_sha256", "report_sha256", "certificate_chain_sha256", "measurements_sha256"):
                sha256_value(details[field], f"NVIDIA {field}")
            lowercase_hex(details["sample_nonce_hex"], "NVIDIA sample nonce")
            nonempty_string(details["driver_version"], "NVIDIA driver version")
            nonempty_string(details["vbios_version"], "NVIDIA VBIOS version")
            if details["rim_signature_evaluation"] != "NOT_PERFORMED":
                raise AttestationBoundaryError("NVIDIA RIM signature boundary changed")
            for label in ("driver_rim", "vbios_rim"):
                rim = details[label]
                if not isinstance(rim, Mapping):
                    raise AttestationBoundaryError(f"{label} must be an object")
                exact(rim, {"sha256", "declared_version", "active_measurements", "mismatched_measurements"}, label)
                sha256_value(rim["sha256"], f"{label}.sha256")
                if rim["declared_version"] is not None:
                    nonempty_string(rim["declared_version"], f"{label}.declared_version")
                for field in ("active_measurements", "mismatched_measurements"):
                    values = rim[field]
                    if not isinstance(values, list):
                        raise AttestationBoundaryError(f"{label}.{field} must be an exact integer array")
                    for value in values:
                        safe_integer(value, f"{label}.{field}")
    for label in ("combined_execution_relation", "live_h100_collection"):
        claim = report[label]
        if not isinstance(claim, Mapping):
            raise AttestationBoundaryError(f"{label} must be an object")
        exact(claim, RESULT_FIELDS, label)
        if claim["state"] not in STATES:
            raise AttestationBoundaryError(f"{label} state changed")
        string_list(claim["reason_codes"], f"{label}.reason_codes", nonempty=True)
        for field in ("verification_depth", "accepted_roots", "evidence_sha256"):
            string_list(claim[field], f"{label}.{field}")
        for field in ("accepted_roots", "evidence_sha256"):
            for value in claim[field]:
                sha256_value(value, f"{label}.{field}")
        if claim["accepted_policy_sha256"] != report["accepted_policy_sha256"]:
            raise AttestationBoundaryError(f"{label} policy binding changed")
    graph = report["control_dependency_graph"]
    if not isinstance(graph, Mapping):
        raise AttestationBoundaryError("control dependency graph must be an object")
    exact(graph, {"logic", "nodes", "edges"}, "control dependency graph")
    if graph["logic"] != "ALL_DECLARED_CONTROLS_REQUIRED":
        raise AttestationBoundaryError("control dependency logic changed")
    if not isinstance(graph["nodes"], list) or not isinstance(graph["edges"], list):
        raise AttestationBoundaryError("control dependency members must be arrays")
    for node in graph["nodes"]:
        if not isinstance(node, Mapping) or set(node) not in ({"id", "kind"}, {"id", "kind", "state"}):
            raise AttestationBoundaryError("control dependency node fields changed")
        if node["kind"] not in {"CONTROL", "CONCLUSION"}:
            raise AttestationBoundaryError("control dependency node kind changed")
        if node["kind"] == "CONCLUSION" and node.get("state") not in STATES:
            raise AttestationBoundaryError("control dependency conclusion state changed")
    for edge in graph["edges"]:
        if not isinstance(edge, Mapping):
            raise AttestationBoundaryError("control dependency edge must be an object")
        exact(edge, {"from", "to"}, "control dependency edge")
    if not isinstance(report["acceptance_cuts"], list):
        raise AttestationBoundaryError("acceptance cuts must be an array")
    for cut in report["acceptance_cuts"]:
        if not isinstance(cut, Mapping):
            raise AttestationBoundaryError("acceptance cut must be an object")
        exact(cut, {"conclusion", "state", "cuts"}, "acceptance cut")
        if cut["state"] not in STATES or not isinstance(cut["cuts"], list):
            raise AttestationBoundaryError("acceptance cut state or members changed")
        for controls in cut["cuts"]:
            string_list(controls, "acceptance cut controls")
    string_list(report["never_infer"], "report never_infer", nonempty=True)
    expected_graph, expected_cuts = dependency_graph(
        artifacts["air"]["subclaims"], artifacts["nvidia"]["subclaims"]
    )
    if graph != expected_graph or report["acceptance_cuts"] != expected_cuts:
        raise AttestationBoundaryError("control dependency graph or acceptance cuts changed")
    context = strict_json(CONTEXT)
    if not isinstance(context, Mapping):
        raise AttestationBoundaryError("attestation context must be an object")
    check_context(context)
    if report["accepted_policy_sha256"] != digest(canonical(context)) or report["never_infer"] != context["never_infer"]:
        raise AttestationBoundaryError("report context binding changed")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--report", type=Path, default=REPORT)
    arguments = parser.parse_args(argv)
    try:
        expected = build_report()
        actual = strict_json(arguments.report)
        if actual != expected:
            raise AttestationBoundaryError("generated attestation boundary report drifted")
        check_report(actual)
    except (AttestationBoundaryError, OSError, ImportError) as exc:
        print(f"attestation boundary rejected: {exc}", file=sys.stderr)
        return 1
    print(f"attestation boundary: {actual['report_sha256']} · PASS")
    return 0


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