#!/usr/bin/env python3
"""Clean-room verifier for the CAIN-42 Evolution 14 capability-governance proof bundle.

IMPORTS NO CAIN-42 CODE. Standard library plus `cryptography` (Ed25519). It re-derives, from the published JSON
alone: file hashes (MANIFEST), the canonical serialization, every capability identity / manifest / dependency
digest, publisher artifact signatures, the software attestation, passports, grants (scope within issuer
authority), leases (bounded TTL, bound to capability / grant / sandbox / credential), the credential binding, the
decision binding and the E13 decision it binds, the E8 token and canonical action it binds, revocation records, the
supply-chain graph root, composition findings, the replay record, the C1-C30 matrix, the >=42-scenario bench, both
proof signatures and the signing-key disclosure.

It proves the bundle is intact and self-consistent. It does NOT prove the capability governance is sound, that a
tool is safe, or that anything outside the enforcement boundary is controlled.

Usage:  python3 verify_e14.py <bundle-directory>
"""
from __future__ import annotations

import base64
import hashlib
import json
import sys
from pathlib import Path

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric import ed25519

D = {"id": "CAIN42/E14-CAPABILITY-ID/v1", "manifest": "CAIN42/E14-CAPABILITY-MANIFEST/v1",
     "depset": "CAIN42/E14-DEPENDENCY-SET/v1", "publisher": "CAIN42/E14-PUBLISHER-ARTIFACT-SIGNATURE/v1",
     "passport": "CAIN42/E14-CAPABILITY-PASSPORT/v1", "grant": "CAIN42/E14-CAPABILITY-GRANT/v1",
     "lease": "CAIN42/E14-CAPABILITY-LEASE/v1", "attest": "CAIN42/E14-CAPABILITY-ATTESTATION/v1",
     "sandbox": "CAIN42/E14-SANDBOX-PROFILE/v1", "credential": "CAIN42/E14-CREDENTIAL-BINDING/v1",
     "binding": "CAIN42/E14-CAPABILITY-DECISION-BINDING/v1", "supply": "CAIN42/E14-SUPPLY-CHAIN-GRAPH/v1",
     "record": "CAIN42/E14-CAPABILITY-AUTHORIZATION-RECORD/v1", "decision": "CAIN42/E13-GOVERNED-DECISION/v1",
     "gat": "CAIN42/E8-GOVERNANCE-AUTHORIZATION-TOKEN/v1", "action": "CAIN42/E8-CANONICAL-ACTION/v1",
     "proof": "CAIN42/E14-PROOF/v1", "master": "CAIN42/E14-MASTER/v1"}
MAX_LEASE_TTL = 120.0
REQUIRED_FILES = ("MANIFEST.json", "SCHEMAS.json", "CAIN42_EVOLUTION14_PROOF.json",
                  "CAIN42_EVOLUTION14_MASTER_PROOF.json", "CAPABILITY_EXAMPLES.json", "CAPABILITY_PASSPORTS.json",
                  "GRANT_EXAMPLES.json", "LEASE_EXAMPLES.json", "DEPENDENCY_GRAPH.json", "ATTACK_MANIFEST.json",
                  "TEST_VECTORS.json", "PERFORMANCE.json", "LIMITATIONS.json", "LIMITATIONS.md", "verify_e14.py.txt",
                  "REPRODUCE.txt", "index.html")
REQUIRED_SCHEMAS = ("GovernedCapability", "CapabilityPassport", "CapabilityGrant", "CapabilityLease",
                    "CapabilitySandboxProfile", "CapabilityCredentialBinding", "CapabilityDecisionBinding",
                    "CapabilityConsequenceVector", "ArtifactProvenance", "DependencySpec", "ToolManifest")
MANIFEST_KEYS = ("schema", "capability_id", "capability_type", "name", "version", "publisher_principal",
                 "artifact_digest", "implementation_digest", "semantic_version", "owner_principal", "provenance",
                 "dependency_set", "dependency_digest", "capability_scope", "input_schema", "output_schema",
                 "required_permissions", "data_access_scope", "network_scope", "filesystem_scope",
                 "credential_requirements", "execution_environment", "side_effect_class", "reversibility",
                 "consequence_class", "risk_class", "trust_requirements", "authority_requirements",
                 "policy_requirements", "supported_protocols", "supported_agent_types", "expiration", "extensions")
IDENTITY_KEYS = ("capability_type", "name", "version", "publisher_principal", "artifact_digest",
                 "implementation_digest")
# the dangerous-combination table, restated independently (not imported)
DANGEROUS = {("READ_SECRET", "NETWORK_ACCESS"): "DENY", ("WRITE_DATABASE", "FINANCIAL_API"): "HUMAN_REVIEW",
             ("CODE_EXECUTION", "CREDENTIAL_ACCESS"): "DENY", ("BROWSER_CONTROL", "PAYMENT"): "HUMAN_REVIEW",
             ("DEPLOYMENT", "PRODUCTION_ACCESS"): "HUMAN_REVIEW", ("SUBAGENT_CREATION", "DELEGATION"): "HUMAN_REVIEW",
             ("MEMORY_WRITE", "POLICY_WRITE"): "DENY", ("IDENTITY_MUTATION", "AUTHORITY_GRANT"): "DENY",
             ("CREDENTIAL_ACCESS", "AUTHORITY_GRANT"): "DENY", ("CREDENTIAL_ACCESS", "NETWORK_ACCESS"): "DENY"}
FORBIDDEN = ("private_key", "private key", "-----begin", "secret_key", "password")


def canon(o) -> bytes:
    return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()


def h(o) -> str:
    return hashlib.sha256(canon(o)).hexdigest()


def digest(domain: str, fields: dict) -> str:
    return h({"domain": domain, **fields})


def sig_ok(pub_b64: str, sig_b64: str, domain: str, fields: dict) -> bool:
    try:
        ed25519.Ed25519PublicKey.from_public_bytes(base64.b64decode(pub_b64)).verify(
            base64.b64decode(sig_b64), digest(domain, fields).encode())
        return True
    except (InvalidSignature, ValueError, TypeError):
        return False


def strip(d: dict, *keys: str) -> dict:
    return {k: v for k, v in d.items() if k not in keys}


class Checker:
    def __init__(self) -> None:
        self.checks: list = []
        self.problems: list = []

    def check(self, name: str, ok: bool, detail: str = "") -> None:
        self.checks.append({"check": name, "ok": bool(ok)})
        if not ok:
            self.problems.append(f"{name}: {detail}" if detail else name)


def main() -> int:
    if len(sys.argv) < 2:
        print("usage: python3 verify_e14.py <bundle-directory>")
        return 2
    d = Path(sys.argv[1])
    C = Checker()
    load = lambda n: json.loads((d / n).read_text())

    # 1 evidence manifest: every file present and hash-exact
    manifest = load("MANIFEST.json")
    bad = [n for n, want in manifest["files"].items() if not (d / n).exists()
           or hashlib.sha256((d / n).read_bytes()).hexdigest() != want]
    C.check("manifest_file_hashes", not bad, ",".join(bad))
    C.check("required_files_present", all((d / f).exists() for f in REQUIRED_FILES) and
            all(f in manifest["files"] for f in REQUIRED_FILES if f != "MANIFEST.json"))

    proof, master = load("CAIN42_EVOLUTION14_PROOF.json"), load("CAIN42_EVOLUTION14_MASTER_PROOF.json")
    ex, pps = load("CAPABILITY_EXAMPLES.json"), load("CAPABILITY_PASSPORTS.json")
    grants, leases = load("GRANT_EXAMPLES.json"), load("LEASE_EXAMPLES.json")
    graph, vectors = load("DEPENDENCY_GRAPH.json"), load("TEST_VECTORS.json")

    # 2 schemas
    C.check("schemas_published", set(REQUIRED_SCHEMAS) <= set(load("SCHEMAS.json")))
    # 3 canonical serialization: independent test vectors
    tv = vectors["canonical"]
    C.check("canonical_serialization_vector", canon(tv["input"]).decode() == tv["canonical_json"]
            and h(tv["input"]) == tv["sha256"])
    C.check("identity_vector", "cap:" + digest(D["id"], vectors["identity"]["identity_body"]) ==
            vectors["identity"]["capability_id"])
    C.check("dependency_vector", digest(D["depset"], {"dependencies": vectors["dependency_set"]["dependencies"]})
            == vectors["dependency_set"]["digest"])
    C.check("sandbox_vector", digest(D["sandbox"], vectors["sandbox"]["body"]) == vectors["sandbox"]["digest"])

    caps = {c["capability_id"]: c for c in ex["capabilities"]}
    powers = set(ex["powers"])
    # 4-8 capability manifests
    id_ok = man_ok = dep_ok = pub_ok = schema_ok = att_ok = True
    for c in ex["capabilities"]:
        m = c["manifest"]
        schema_ok &= set(MANIFEST_KEYS) <= set(m) and m["required_permissions"] == sorted(m["required_permissions"]) \
            and set(m["required_permissions"]) <= powers
        id_ok &= "cap:" + digest(D["id"], {k: m[k] for k in IDENTITY_KEYS}) == c["capability_id"] == m["capability_id"]
        man_ok &= digest(D["manifest"], m) == c["manifest_digest"]
        dep_ok &= digest(D["depset"], {"dependencies": m["dependency_set"]}) == m["dependency_digest"]
        prov = c["provenance"]
        pub_ok &= prov["artifact_digest"] == m["artifact_digest"] and sig_ok(
            prov["publisher_key_b64"], prov["publisher_signature_b64"], D["publisher"],
            strip(prov, "publisher_key_b64", "publisher_signature_b64")) and \
            strip(prov, "publisher_key_b64", "publisher_signature_b64") == m["provenance"] and \
            ex["trusted_publishers"].get(prov["publisher_principal"]) == prov["publisher_key_b64"]
        a = c["attestation"]
        att_ok &= a["level"] == "SOFTWARE_CONFIGURATION" and a["hardware_attestation"] == "UNKNOWN" and \
            a["manifest_digest"] == c["manifest_digest"] and a["artifact_digest"] == m["artifact_digest"] and \
            a["dependency_digest"] == m["dependency_digest"] and sig_ok(
                a["issuer_public_key_b64"], a["signature_b64"], D["attest"],
                strip(a, "status", "issuer_public_key_b64", "signature_b64"))
    C.check("capability_manifest_schema", schema_ok)
    C.check("capability_identity_binds_artifact", id_ok)
    C.check("capability_manifest_digest", man_ok)
    C.check("dependency_binding", dep_ok)
    C.check("publisher_artifact_signature", pub_ok)
    C.check("software_attestation_not_hardware", att_ok)

    fabric_key = ex["fabric_public_key_b64"]
    # 9 passports
    ok = True
    for p in pps["passports"]:
        body = strip(p, "digest", "issuer_public_key_b64", "signature_b64", "authority")
        c = caps.get(p["capability_id"])
        ok &= c is not None and p["issuer_public_key_b64"] == fabric_key and sig_ok(
            fabric_key, p["signature_b64"], D["passport"], body) and digest(D["passport"], body) == p["digest"] \
            and p["manifest_digest"] == c["manifest_digest"] and p["artifact_digest"] == c["manifest"]["artifact_digest"]
    C.check("passport_signatures_and_bindings", ok and pps["passports"])

    # 10-11 grants
    g_ok = scope_ok = True
    for g in grants["grants"]:
        body = strip(g["grant"], "digest", "issuer_public_key_b64", "signature_b64", "authority")
        g_ok &= sig_ok(fabric_key, g["grant"]["signature_b64"], D["grant"], body) and \
            digest(D["grant"], body) == g["grant"]["digest"] and \
            caps[body["capability_id"]]["manifest_digest"] == body["capability_manifest_digest"] and \
            body["expires_at"] > body["issued_at"]
        auth = sorted(g["issuer_authority"])
        scope_ok &= all(s in auth for s in body["scope"]) and h(auth) == body["authority_digest"] and \
            all(s[0] in caps[body["capability_id"]]["manifest"]["required_permissions"] for s in body["scope"])
    C.check("grant_signatures", g_ok and grants["grants"])
    C.check("grant_within_issuer_authority", scope_ok)
    C.check("refused_grants_recorded", all("reason" in r and r.get("granted") is False for r in grants["refused"])
            and len(grants["refused"]) >= 2)

    # 12-14 leases, sandbox, credential
    grant_by_id = {g["grant"]["grant_id"]: g["grant"] for g in grants["grants"]}
    sb = ex["sandbox"]
    sb_digest = digest(D["sandbox"], strip(sb, "digest", "confinement", "authority"))
    cred_ok, cred_digests = True, set()
    for cred in leases["credential_bindings"]:
        cred_body = strip(cred, "digest", "issuer_public_key_b64", "signature_b64", "authority", "grants_authority")
        cred_ok &= sig_ok(fabric_key, cred["signature_b64"], D["credential"], cred_body) and \
            digest(D["credential"], cred_body) == cred["digest"] and cred["grants_authority"] is False and \
            cred["credential_ref"].startswith("credref:") and cred["expires_at"] - cred["issued_at"] <= MAX_LEASE_TTL
        cred_digests.add(cred["digest"])
    C.check("credential_bindings", cred_ok and leases["credential_bindings"]
            and leases["credential_binding"]["digest"] in cred_digests)
    l_ok = ttl_ok = bind_ok = True
    for l in leases["leases"]:
        body = strip(l, "digest", "issuer_public_key_b64", "signature_b64", "authority")
        l_ok &= sig_ok(fabric_key, l["signature_b64"], D["lease"], body) and digest(D["lease"], body) == l["digest"]
        ttl_ok &= 0 < l["expiration"] - l["issued_at"] <= MAX_LEASE_TTL
        g = grant_by_id.get(l["grant_id"])
        bind_ok &= g is not None and g["capability_id"] == l["capability_id"] and \
            g["capability_manifest_digest"] == l["capability_manifest_digest"] and g["scope"] == l["scope"] and \
            g["subject_agent"] == l["agent_id"] and g["authority_digest"] == l["authority_digest"] and \
            l["sandbox_profile_digest"] == sb_digest and \
            l["credential_binding_digest"] in cred_digests | {""}
    C.check("lease_signatures", l_ok and leases["leases"])
    C.check("lease_bounded_ttl", ttl_ok)
    C.check("lease_bindings", bind_ok)
    C.check("sandbox_default_deny", sb_digest == sb["digest"] and sb["default"] == "DENY")

    # 15 revocations
    rv = leases["revocation"]
    revoked = rv["registry"]["revoked"]
    C.check("revocation_propagates", f"capability:{rv['subject']}" in revoked and
            all(f"lease:{l}" in revoked for l in rv["propagated"].get("lease", [])) and
            all(f"token:{t}" in revoked for t in rv["propagated"].get("token", [])) and
            rv["propagated"].get("lease") and rv["propagated"].get("token") and
            "LEASE_REVOKED" in rv["lease_validation_after"])

    # 16-18 decision binding, E13 decision, E8 token + canonical action
    b, dec_, gat, act = leases["binding"], leases["decision"], leases["gat"], leases["action"]
    b_body = strip(b, "digest", "issuer_public_key_b64", "signature_b64", "authority")
    dec_body = strip(dec_, "digest", "issuer", "issuer_public_key_b64", "signature_b64", "authority")
    act_body = strip(act, "action_hash", "authorization_id", "authorization_expiry", "nonce", "authority")
    committed = next(l for l in leases["leases"] if l["lease_id"] == leases["committed_lease_id"])
    C.check("decision_signature_e13", sig_ok(dec_["issuer_public_key_b64"], dec_["signature_b64"], D["decision"],
                                             dec_body) and digest(D["decision"], dec_body) == dec_["digest"]
            and dec_["decision_state"] in ("ALLOWED", "CONDITIONALLY_ALLOWED"))
    C.check("capability_decision_binding", sig_ok(b["issuer_public_key_b64"], b["signature_b64"], D["binding"], b_body)
            and digest(D["binding"], b_body) == b["digest"] and b["decision_digest"] == dec_["digest"]
            and b["lease_digest"] == committed["digest"] and b["capability_digest"] == committed["capability_id"]
            and b["capability_manifest_digest"] == committed["capability_manifest_digest"]
            and b["action_hash"] == act["action_hash"])
    C.check("canonical_action_hash", digest(D["action"], act_body) == act["action_hash"])
    gat_body = strip(gat, "issuer", "signature_b64")
    C.check("capability_action_binding_e8_token", sig_ok(gat["issuer"], gat["signature_b64"], D["gat"], gat_body)
            and gat["action_hash"] == act["action_hash"] and gat["capability_lease_digest"] == committed["digest"]
            and gat["capability_manifest_digest"] == committed["capability_manifest_digest"]
            and gat["capability_digest"] == committed["capability_id"] and gat["capability_binding_digest"] == b["digest"]
            and gat["decision_digest"] == dec_["digest"] and gat["expires_at"] - gat["issued_at"] <= 30.0)

    # 19 supply chain graph
    nodes = sorted([n["id"], n["stage"]] for n in graph["graph"]["nodes"])
    C.check("supply_chain_graph_root", digest(D["supply"], {"nodes": nodes, "edges": sorted(graph["graph"]["edges"])})
            == graph["graph"]["root"] and graph["graph"]["vulnerability_intelligence"] == "UNKNOWN")
    C.check("supply_chain_findings_recorded", all(x["admitted"] is False and x["codes"] for x in graph["refused_updates"])
            and len(graph["refused_updates"]) >= 3)

    # 20 composition recomputed from the published power surface with the independent table
    comp = ex["composition_example"]
    surface = set(comp["power_surface"])
    effects = [eff for pair, eff in DANGEROUS.items() if set(pair) <= surface]
    want = "DENY" if "DENY" in effects else ("HUMAN_REVIEW" if effects else "ALLOW")
    C.check("composition_recomputed", comp["effect"] == want == "DENY")

    # 21 replay record
    rec = leases["authorization_record"]
    rec_body = {"schema": rec["schema"], "input_digests": rec["input_digests"], "outputs": rec["outputs"]}
    C.check("replay_record", sig_ok(rec["issuer_public_key_b64"], rec["signature_b64"], D["record"], rec_body)
            and all(h(rec["inputs"][k]) == v for k, v in rec["input_digests"].items())
            and rec["outputs"]["disposition"] == "AUTHORIZED"
            and rec["outputs"]["manifest_digest"] == digest(D["manifest"], strip(
                rec["inputs"]["manifest"], "manifest_digest", "lifecycle_state", "revocation_state", "attestation",
                "evidence_refs", "created_at", "updated_at", "authority") | {"provenance": strip(
                    rec["inputs"]["manifest"]["provenance"], "publisher_key_b64", "publisher_signature_b64")}))

    # 22-23 invariants + bench
    inv = proof["invariants"]
    C.check("invariants_c1_c30", {x["id"] for x in inv["checks"]} == {f"C{i}" for i in range(1, 31)}
            and all(x["holds"] for x in inv["checks"]) and inv["all_hold"] is True)
    bench = load("ATTACK_MANIFEST.json")
    C.check("adversarial_bench", bench["total"] >= 42 and bench["contained"] == bench["total"] and
            len(bench["attacks"]) == bench["total"] and all(a["contained"] for a in bench["attacks"].values())
            and (bench["contained"], bench["total"]) == (proof["attack_bench"]["contained"],
                                                          proof["attack_bench"]["total"]))

    # 24-26 signatures, module digests, disclosure
    pbody = strip(proof, "signature_b64", "signer_public_key_b64")
    C.check("proof_signature", sig_ok(proof["signer_public_key_b64"], proof["signature_b64"], D["proof"], pbody))
    mbody = strip(master, "signature_b64", "signer_public_key_b64")
    C.check("master_signature", sig_ok(master["signer_public_key_b64"], master["signature_b64"], D["master"], mbody)
            and master["signer_public_key_b64"] == proof["signer_public_key_b64"])
    C.check("master_matches_proof", master["modules"] == {k: v["sha256"] for k, v in proof["modules"].items()}
            and master["tests_failed"] == 0 and master["tests_passed"] == proof["test_run"]["passed"] > 0
            and master["attacks_contained"] == bench["contained"])
    C.check("signing_key_disclosed_ephemeral", proof["signing_key"]["class"] == "EPHEMERAL"
            and "not a production trust root" in proof["signing_key"]["note"])
    C.check("honest_unknowns", proof["status"]["HARDWARE_ATTESTATION"] == "UNKNOWN" and
            proof["status"]["VULNERABILITY_INTELLIGENCE"] == "UNKNOWN" and
            proof["status"]["HOSTED_SERVICE"] == "NOT_IMPLEMENTED" and
            proof["status"]["MULTI_HOST_SCALE"] == "UNVERIFIED")
    blob = "".join((d / n).read_text().lower() for n in manifest["files"] if n.endswith(".json"))
    C.check("no_key_material", not any(f in blob for f in FORBIDDEN))

    print(json.dumps({"bundle": str(d), "checks": len(C.checks), "passed": sum(c["ok"] for c in C.checks),
                      "problems": C.problems, "result": "INTACT" if not C.problems else "FAILED",
                      "detail": C.checks}, indent=2))
    return 0 if not C.problems else 1


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