#!/usr/bin/env python3 """Clean-room verifier for Evolution 18 execution-truth receipts. Imports nothing from CAIN. Given a receipt and the keys the reader trusts, it re-checks every signature and digest and RE-DERIVES the grade and the verdict from the approver-signed and sensor-signed records inside the receipt. It never copies the enforcer's own conclusion. python3 verify.py RECEIPT.json ANCHORS.json exit 0 only for VERIFIED ANCHORS.json: {"approver_pubs": [...], "enforcer_pubs": [...], "sensor_pubs": [...]} (base64 raw Ed25519) Verdicts, for the claim "exactly the authorized effects occurred inside the observed roots": VERIFIED every declared effect was seen by sensors in other processes at the required level, nothing undeclared was seen, postconditions hold, every signature and causal link checks out FAILED evidence contradicts the claim, or the receipt is not intact, or it claims more than it shows INCOMPLETE nothing contradicts the claim, but required evidence is missing UNSUPPORTED a declared effect has no observation method here (network, process execution), or unknown schema UNAVAILABLE there is no observation to judge (the action was refused before execution, or no sensor ran) What VERIFIED does not mean: that nothing happened outside the observed roots, over the network, in memory or in child processes; that the sensors could not be deceived by the host kernel or a same-user debugger; that the principal named in the receipt is who they say (that node is the enforcer's assertion). """ import base64 import hashlib import json import sys from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 D_ACTION_AUTH = "CAIN42/EVOLUTION17/ACTION-AUTHORIZATION/v1" D_EFFECTS = "CAIN42/EVOLUTION17/EFFECT-SET/v1" D_BINDING = "CAIN42/EVOLUTION18/RESOURCE-BINDING/v1" D_RECEIPT = "CAIN42/EVOLUTION18/EXECUTION-TRUTH-RECEIPT/v2" D_OBS = "CAIN42/EVOLUTION18/SENSOR-OBSERVATION/v2" D_NODE = "CAIN42/EVOLUTION18/CAUSAL-NODE/v2" D_ROOT = "CAIN42/EVOLUTION18/CAUSAL-GRAPH-ROOT/v2" D_RECOVERY = "CAIN42/EVOLUTION18/RECOVERY-STATEMENT/v1" RECEIPT_SCHEMA = "cain42.execution_truth_receipt.v2" GRAPH_SCHEMA = "cain42.causal_evidence_graph.v2" NODE_SCHEMA = "cain42.causal_node.v2" CAUSAL = {"DELEGATES", "PROPOSES", "AUTHORIZES", "ADMITS", "EXECUTES", "OBSERVES", "DERIVES", "COMPENSATES", "ATTESTS"} RANK = ["CONFLICTING", "UNAVAILABLE", "SELF_REPORTED", "INSTRUMENTED", "EXTERNALLY_OBSERVED", "INDEPENDENTLY_CORROBORATED"] AUTH_FIELDS = ("schema", "auth_id", "mission_id", "contract_version", "principal", "agent_id", "runtime", "effect_set_digest", "permitted_resources", "policy_version", "decision", "issued_at", "expires_at") def canon(o): return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def h(o): return hashlib.sha256(canon(o)).hexdigest() def dg(domain, fields): return h({"domain": domain, **fields}) def sig_ok(pub, sig, domain, fields): try: ed25519.Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify( base64.b64decode(sig), dg(domain, fields).encode()) return True except (InvalidSignature, ValueError, TypeError): return False def effect_digest(effects): uniq = {} for e in effects: b = {"effect": e["effect"], "resource": e["resource"], "material": dict(e.get("material") or {})} uniq[(b["effect"], b["resource"], h(b["material"]))] = b body = sorted(uniq.values(), key=lambda b: (b["effect"], b["resource"], h(b["material"]))) return dg(D_EFFECTS, {"schema": "cain42.effect.v1", "effects": body}) def check_graph(graph, enforcer_pubs): """(reasons, nodes by id, order).""" bad, nodes, order = [], {}, [] if not isinstance(graph, dict) or graph.get("schema") != GRAPH_SCHEMA: return ["GRAPH_SCHEMA_UNSUPPORTED"], nodes, order for n in graph.get("nodes") or []: try: body = {"schema": NODE_SCHEMA, "kind": n["kind"], "parents": sorted(({"id": str(p["id"]), "rel": str(p["rel"])} for p in n["parents"]), key=lambda p: (p["id"], p["rel"])), "correlation_id": n["correlation_id"], "seq": int(n["seq"]), "source": dict(n["source"]), "payload": dict(n["payload"]), "clock": dict(n["clock"])} if n.get("schema") != NODE_SCHEMA or dg(D_NODE, body) != n.get("node_id"): bad.append("NODE_DIGEST_MISMATCH") continue basis = n["source"].get("basis") if basis == "SIGNED_BY_SOURCE": if not n.get("sig") or not sig_ok(n["source"].get("pub"), n["sig"], D_NODE, body): bad.append("NODE_SIGNATURE_INVALID") if n["source"].get("role") == "ENFORCER" and n["source"].get("pub") not in enforcer_pubs: bad.append("ENFORCER_NODE_KEY_NOT_ANCHORED") elif basis != "ASSERTED_BY_RECORDER" or n.get("sig"): bad.append("NODE_BASIS_INVALID") if n["node_id"] in nodes: bad.append("DUPLICATE_NODE") if n["correlation_id"] != graph.get("correlation_id"): bad.append("NODE_CORRELATION_MISMATCH") for p in n["parents"]: if p["id"] not in nodes: bad.append("PARENT_MISSING_OR_LATER") # also rules out cycles elif int(n["seq"]) <= int(nodes[p["id"]]["seq"]): bad.append("LOGICAL_CLOCK_NOT_AFTER_PARENT") nodes[n["node_id"]] = n order.append(n["node_id"]) except Exception as e: # noqa: BLE001 bad.append("MALFORMED_NODE:" + type(e).__name__) if dg(D_ROOT, {"schema": GRAPH_SCHEMA, "correlation_id": graph.get("correlation_id"), "nodes": order}) != \ graph.get("root_digest"): bad.append("GRAPH_ROOT_MISMATCH") return bad, nodes, order def causal_path(nodes, ancestor, descendant): stack, seen = [descendant], set() while stack: cur = stack.pop() if cur == ancestor: return True if cur in seen or cur not in nodes: continue seen.add(cur) stack.extend(p["id"] for p in nodes[cur]["parents"] if p["rel"] in CAUSAL) return False def under(resource, root): return resource == root or resource.startswith(root.rstrip("/") + "/") def derive_grade(declared, observations, claims, instrumented): sensors, gaps, keys, pids = [], set(), set(), set() for o in observations: s, sc = o.get("sensor") or {}, o.get("scope") or {} if s.get("boundary") != "SEPARATE_PROCESS": gaps.add("SENSOR_NOT_A_SEPARATE_PROCESS:%s" % s.get("id")) elif s.get("pub") in keys or s.get("pid") in pids: gaps.add("SENSOR_SHARES_KEY_OR_PROCESS_WITH_ANOTHER:%s" % s.get("id")) elif not sc.get("complete"): gaps.add("SENSOR_SCOPE_INCOMPLETE:%s" % s.get("id")) else: sensors.append((s.get("id"), s.get("method"), str(sc.get("root") or ""), set(sc.get("observable") or []), {(c["effect"], c["resource"]) for c in o.get("changes") or []})) keys.add(s.get("pub")) pids.add(s.get("pid")) dkeys = {(d["effect"], d["resource"]) for d in declared} ckeys = {(c["effect"], c["resource"]) for c in claims} mkeys = {(c["effect"], c["resource"]) for c in instrumented} levels, contradictions, absent, unsupported = [], [], [], [] for d in declared: k = (d["effect"], d["resource"]) if d["effect"] not in ("READ", "WRITE", "DELETE") or not str(d["resource"]).startswith("file:"): unsupported.append(k) levels.append("SELF_REPORTED" if k in ckeys else "UNAVAILABLE") continue cover = [s for s in sensors if d["effect"] in s[3] and under(d["resource"], s[2])] saw = [s for s in cover if k in s[4]] if not cover: levels.append("INSTRUMENTED" if k in mkeys else "SELF_REPORTED" if k in ckeys else "UNAVAILABLE") gaps.add("NO_EXTERNAL_SENSOR_COVERS:%s:%s" % k) elif d["effect"] == "READ": if saw: levels.append("EXTERNALLY_OBSERVED") else: levels.append("SELF_REPORTED" if k in ckeys else "UNAVAILABLE") gaps.add("READ_NOT_OBSERVED:%s" % d["resource"]) elif saw and len(saw) < len(cover): levels.append("CONFLICTING") contradictions.append(k) elif saw: levels.append("INDEPENDENTLY_CORROBORATED" if len({s[1] for s in saw}) >= 2 else "EXTERNALLY_OBSERVED") else: levels.append("UNAVAILABLE") absent.append(k) undeclared = {k for s in sensors for k in s[4] if k not in dkeys} | (mkeys - dkeys) if not any({"WRITE", "DELETE"} <= s[3] for s in sensors): gaps.add("NO_COMPLETE_SENSOR_FOR_UNDECLARED_MUTATIONS") if not any("READ" in s[3] for s in sensors): gaps.add("NO_COMPLETE_SENSOR_FOR_UNDECLARED_READS") return {"levels": levels, "overall": min(levels, key=RANK.index) if levels else "UNAVAILABLE", "undeclared": sorted(undeclared), "contradictions": contradictions, "absent": absent, "unsupported": unsupported, "gaps": sorted(gaps), "sensors": len(sensors)} def derive_postconditions(declared, observations, stated): tree = {} for o in observations: s = o.get("sensor") or {} if s.get("method") == "TREE_DIFF" and s.get("boundary") == "SEPARATE_PROCESS" and (o.get("scope") or {}).get("complete"): for c in o.get("changes") or []: tree[(c["effect"], c["resource"])] = c by_enforcer = {(p["effect"], p["resource"]): p for p in stated or [] if p.get("basis") == "ENFORCER_INSTRUMENTATION"} out = [] for d in declared: k = (d["effect"], d["resource"]) content = (d.get("material") or {}).get("content") if d["effect"] == "WRITE" and isinstance(content, str): want = hashlib.sha256(content.encode()).hexdigest() if k in tree: out.append("HOLDS" if tree[k].get("sha256") == want else "VIOLATED") elif k in by_enforcer: out.append("HOLDS" if by_enforcer[k].get("observed_sha256") == want else "VIOLATED") else: out.append("NOT_CHECKED") elif d["effect"] == "DELETE": out.append("HOLDS" if k in tree else by_enforcer[k]["status"] if k in by_enforcer else "NOT_CHECKED") elif d["effect"] == "READ": out.append("NOT_APPLICABLE") else: out.append("NOT_CHECKED") return out def verify_receipt(rx, anchors, seen=None, required_level="EXTERNALLY_OBSERVED"): """seen: a dict the caller keeps between calls ({"receipts": set, "auth_ids": set}) to catch replays.""" fail, notes = [], [] def done(verdict, reasons, **extra): return dict({"verdict": verdict, "reasons": sorted(set(reasons)), "notes": notes}, **extra) if not isinstance(rx, dict) or rx.get("schema") != RECEIPT_SCHEMA: return done("UNSUPPORTED", ["RECEIPT_SCHEMA_UNSUPPORTED"]) enforcer_pubs = list(anchors.get("enforcer_pubs") or []) approver_pubs = list(anchors.get("approver_pubs") or []) sensor_pubs = anchors.get("sensor_pubs") core = {k: v for k, v in rx.items() if k not in ("signature", "graph")} if rx.get("enforcer_pub") not in enforcer_pubs: fail.append("ENFORCER_KEY_NOT_ANCHORED") elif not sig_ok(rx["enforcer_pub"], rx.get("signature"), D_RECEIPT, core): fail.append("RECEIPT_SIGNATURE_INVALID") gbad, nodes, order = check_graph(rx.get("graph"), enforcer_pubs) fail += gbad if (rx.get("graph") or {}).get("root_digest") != rx.get("graph_root"): fail.append("RECEIPT_NOT_BOUND_TO_THIS_GRAPH") kinds = {} for i in order: kinds.setdefault(nodes[i]["kind"], []).append(nodes[i]) rnode = (kinds.get("RECEIPT") or [None])[-1] if rnode is None or rnode["payload"].get("receipt_core_digest") != h({k: v for k, v in core.items() if k != "graph_root"}): fail.append("RECEIPT_NODE_DOES_NOT_MATCH_RECEIPT") if seen is not None: if rx.get("receipt_id") in seen.setdefault("receipts", set()): fail.append("RECEIPT_REPLAYED") seen["receipts"].add(rx.get("receipt_id")) if fail: return done("FAILED", fail) status = rx.get("status") if status == "BLOCKED": notes.append("REFUSED_BEFORE_EXECUTION:%s" % rx.get("block_reason")) return done("UNAVAILABLE", ["NOT_EXECUTED:NON_EXECUTION_WAS_NOT_INDEPENDENTLY_OBSERVED"]) if status == "AMBIGUOUS_OUTCOME": return done("INCOMPLETE", ["CRASH_BETWEEN_ADMISSION_AND_EVIDENCE"]) # ---- authorization: real, anchored, for exactly these effects, live at admission auth, binding = rx.get("authorization") or {}, rx.get("binding") or {} bbody = binding.get("body") or {} declared = rx.get("declared_effects") or [] if auth.get("approver_pub") not in approver_pubs: fail.append("APPROVER_KEY_NOT_ANCHORED") elif not sig_ok(auth["approver_pub"], auth.get("signature"), D_ACTION_AUTH, {k: auth.get(k) for k in AUTH_FIELDS}): fail.append("AUTHORIZATION_SIGNATURE_INVALID") if auth.get("decision") != "ALLOW": fail.append("AUTHORIZATION_IS_NOT_ALLOW") try: if effect_digest(declared) != auth.get("effect_set_digest") or rx.get("effect_set_digest") != auth.get("effect_set_digest"): fail.append("EFFECTS_DIFFER_FROM_AUTHORIZED") except Exception: # noqa: BLE001 fail.append("MALFORMED_DECLARED_EFFECTS") if auth.get("policy_version") != rx.get("policy_version"): fail.append("POLICY_VERSION_MISMATCH") if auth.get("mission_id") != rx.get("mission_id") or auth.get("auth_id") != rx.get("auth_id"): fail.append("RECEIPT_BOUND_TO_ANOTHER_MISSION_OR_AUTHORIZATION") if binding.get("signer_pub") != auth.get("approver_pub") or binding.get("digest") != dg(D_BINDING, bbody) or \ not sig_ok(binding.get("signer_pub"), binding.get("signature"), D_BINDING, bbody): fail.append("RESOURCE_BINDING_INVALID") elif bbody.get("auth_id") != auth.get("auth_id") or bbody.get("effect_set_digest") != auth.get("effect_set_digest"): fail.append("RESOURCE_BINDING_IS_FOR_ANOTHER_AUTHORIZATION") if seen is not None: if auth.get("auth_id") in seen.setdefault("auth_ids", set()): fail.append("AUTHORIZATION_USED_FOR_MORE_THAN_ONE_EXECUTION") seen["auth_ids"].add(auth.get("auth_id")) # ---- causal chain: authorization -> admission -> execution -> each observation ; execution -> receipt a_nodes = [n for n in kinds.get("AUTHORIZATION", []) if n["payload"].get("authorization") == auth and n["payload"].get("binding") == binding] e_nodes = [n for n in kinds.get("ENFORCEMENT", []) if n["payload"].get("decision") == "ADMIT" and n["payload"].get("auth_id") == auth.get("auth_id")] x_nodes = [n for n in kinds.get("EXECUTION", []) if n["payload"].get("task_id") == rx.get("task_id")] if len(a_nodes) != 1 or len(e_nodes) != 1 or len(x_nodes) != 1: fail.append("CAUSAL_CHAIN_NODES_MISSING_OR_DUPLICATED") else: an, en, xn = a_nodes[0], e_nodes[0], x_nodes[0] if not (causal_path(nodes, an["node_id"], en["node_id"]) and causal_path(nodes, en["node_id"], xn["node_id"]) and causal_path(nodes, xn["node_id"], rnode["node_id"])): fail.append("CAUSAL_CHAIN_BROKEN") if float(en["clock"]["t"]) > float(auth.get("expires_at", 0)) + float(en["clock"].get("uncertainty_s", 0)): fail.append("ADMITTED_AFTER_AUTHORIZATION_EXPIRED") authorized = {r["resource"]: r for r in bbody.get("resources") or []} for p in en["payload"].get("pinned") or []: a = authorized.get(p.get("resource")) if a is None or any(a.get(f) != p.get(f) for f in ("parent", "exists", "dev", "ino", "sha256", "size")): fail.append("EXECUTED_OBJECT_IS_NOT_THE_AUTHORIZED_OBJECT") if xn["payload"].get("status") != status or xn["payload"].get("executor_claims") != rx.get("executor_claims"): fail.append("EXECUTION_NODE_DOES_NOT_MATCH_RECEIPT") obs_in_graph = [n for n in kinds.get("OBSERVATION", [])] if sorted(h(n["payload"].get("signed")) for n in obs_in_graph) != sorted(h(o) for o in rx.get("observations") or []): fail.append("OBSERVATIONS_DIFFER_BETWEEN_RECEIPT_AND_GRAPH") for n in obs_in_graph: if not causal_path(nodes, xn["node_id"], n["node_id"]): fail.append("OBSERVATION_NOT_CAUSALLY_LINKED_TO_THIS_EXECUTION") # ---- observations: signed by distinct keys that are not the enforcer's or approver's, for THIS run bind = {"auth_id": rx.get("auth_id"), "task_id": rx.get("task_id"), "effect_set_digest": rx.get("effect_set_digest")} obs, nonces = [], set() for o in rx.get("observations") or []: try: body = o["observation"] pub = body["sensor"]["pub"] if not sig_ok(pub, o["sig"], D_OBS, body): fail.append("OBSERVATION_SIGNATURE_INVALID") elif pub in enforcer_pubs or pub in approver_pubs: fail.append("SENSOR_KEY_IS_AN_ENFORCER_OR_APPROVER_KEY") elif body.get("bind") != bind: fail.append("OBSERVATION_BOUND_TO_ANOTHER_ACTION") elif body.get("nonce") in nonces: fail.append("OBSERVATION_NONCE_REUSED") elif sensor_pubs is not None and pub not in sensor_pubs: fail.append("SENSOR_KEY_NOT_ANCHORED") else: nonces.add(body.get("nonce")) obs.append(body) except Exception as e: # noqa: BLE001 fail.append("MALFORMED_OBSERVATION:" + type(e).__name__) if fail: return done("FAILED", fail) claims = rx.get("executor_claims") or [] occurred = bool(claims) or status in ("COMPLETED", "PARTIAL") eff = declared if occurred else [] g = derive_grade(eff, obs, claims, rx.get("instrumented") or []) post = derive_postconditions(eff, obs, rx.get("postconditions")) derived = {"overall_level": g["overall"], "levels": g["levels"], "gaps": g["gaps"], "undeclared": g["undeclared"], "postconditions": post, "claim": "AUTHORIZED_EFFECTS_OCCURRED" if occurred else "NO_EFFECT_OCCURRED"} if rx.get("authority_at_completion") not in ("LIVE", "NOT_APPLICABLE"): notes.append("AUTHORITY_%s_BY_COMPLETION:the effect was committed while authority was live" % rx.get("authority_at_completion")) if RANK.index(rx.get("trust_level", "UNAVAILABLE")) > RANK.index(g["overall"]): fail.append("TRUST_LEVEL_OVERCLAIMED") reasons = [] if g["undeclared"]: reasons.append("UNDECLARED_EFFECT_OBSERVED") if g["contradictions"]: reasons.append("SENSORS_CONTRADICT_EACH_OTHER") if g["absent"]: reasons.append("DECLARED_EFFECT_NOT_OBSERVED_BY_ANY_COVERING_SENSOR") if "VIOLATED" in post: reasons.append("POSTCONDITION_VIOLATED") if status == "PARTIAL": reasons.append("PARTIAL_EXECUTION") if reasons: verdict = "FAILED" elif g["unsupported"]: verdict, reasons = "UNSUPPORTED", ["NO_OBSERVATION_METHOD_FOR_SOME_DECLARED_EFFECT"] elif not g["sensors"] and all(x == "UNAVAILABLE" for x in g["levels"]): verdict, reasons = "UNAVAILABLE", ["NO_OBSERVATION_AT_ALL"] else: if any(RANK.index(x) < RANK.index(required_level) for x in g["levels"]): reasons.append("EFFECT_BELOW_REQUIRED_LEVEL:%s" % required_level) if g["gaps"]: reasons.append("EVIDENCE_GAPS_PRESENT") if "NOT_CHECKED" in post: reasons.append("POSTCONDITION_NOT_CHECKED") if rx.get("sensor_failures"): reasons.append("A_SENSOR_FAILED") verdict = "INCOMPLETE" if reasons else "VERIFIED" stated = (rx.get("verdict") or {}).get("verdict") if stated == "VERIFIED" and verdict != "VERIFIED": fail.append("RECEIPT_CLAIMS_VERIFIED_WITHOUT_SUPPORT") if verdict == "VERIFIED" and sensor_pubs is None: # the reader's own limit, not a fault of the receipt: it cannot tell these sensors from ones the enforcer made up verdict, reasons = "INCOMPLETE", ["SENSOR_KEYS_NOT_ANCHORED_BY_THE_READER"] if fail: return done("FAILED", fail + reasons, derived=derived) return done(verdict, reasons, derived=derived, enforcer_stated=stated) def verify_recovery(statement, original, compensations, anchors, seen=None): """Re-derive the recovery state. FULLY_COMPENSATED needs every changed object restored by a VERIFIED run.""" body = (statement or {}).get("body") or {} if statement.get("signer_pub") not in (anchors.get("enforcer_pubs") or []) or \ statement.get("digest") != dg(D_RECOVERY, body) or \ not sig_ok(statement.get("signer_pub"), statement.get("signature"), D_RECOVERY, body): return {"verdict": "FAILED", "reasons": ["RECOVERY_STATEMENT_NOT_INTACT"]} if body.get("original_receipt") != original.get("receipt_id"): return {"verdict": "FAILED", "reasons": ["RECOVERY_STATEMENT_IS_FOR_ANOTHER_RECEIPT"]} good = [c for c in compensations if verify_receipt(c, anchors, seen)["verdict"] == "VERIFIED"] before = {r["resource"]: r for r in ((original.get("binding") or {}).get("body") or {}).get("resources") or []} changed = {c["resource"] for o in original.get("observations") or [] for c in o["observation"]["changes"] if c["effect"] in ("WRITE", "DELETE")} | {c["resource"] for c in original.get("instrumented") or []} restored = set() for c in good: for o in c.get("observations") or []: ob = o["observation"] if ob["sensor"]["method"] != "TREE_DIFF": continue for ch in ob["changes"]: b = before.get(ch["resource"]) if b is None: continue if (ch["effect"] == "WRITE" and b["exists"] and ch.get("sha256") == b["sha256"]) or \ (ch["effect"] == "DELETE" and not b["exists"]): restored.add(ch["resource"]) state = "NOTHING_TO_COMPENSATE" if not changed else "FULLY_COMPENSATED" if changed <= restored else \ "PARTIALLY_COMPENSATED" if changed & restored else "NOT_COMPENSATED" if body.get("state") != state: return {"verdict": "FAILED", "reasons": ["RECOVERY_STATE_OVERSTATED_OR_WRONG"], "derived_state": state} return {"verdict": "VERIFIED", "reasons": [], "derived_state": state} def main(argv): if len(argv) != 3: print(__doc__) return 64 with open(argv[1], "r", encoding="utf-8") as f: rx = json.load(f) with open(argv[2], "r", encoding="utf-8") as f: anchors = json.load(f) out = verify_receipt(rx, anchors) print(json.dumps(out, indent=2, sort_keys=True)) return {"VERIFIED": 0, "FAILED": 2}.get(out["verdict"], 3) if __name__ == "__main__": sys.exit(main(sys.argv))