#!/usr/bin/env python3 """CAIN-42 Evolution 12 -- CLEAN-ROOM TRUST-NETWORK VERIFIER. ZERO CAIN imports. Re-implements the E12 canonical state model and the identity/capability/trust/ interaction rules from the specification, using only the standard library and `cryptography`. It exists so a third party can verify an E12 trust-network artifact — or replay an E12 operation log and recompute the network state root — without running or trusting the CAIN implementation. Usage: python3 trustnet_verifier.py state [--state-root HEX] python3 trustnet_verifier.py identity [--domain D] python3 trustnet_verifier.py capability [--now-ms N] python3 trustnet_verifier.py interaction Exit 0 = verified, 1 = rejected, 2 = usage/IO error. Honest scope: the identity/capability/interaction inputs are the plain canonical objects (base64 Ed25519 signatures), NOT the E11-signed proof chain; this verifier intentionally has no CAIN imports and does not re-implement the entire E11 proof format. For proof-chain verification use the published E11 verifier (verify_e11.py). """ from __future__ import annotations import argparse import base64 import hashlib import json import sys from typing import Any, Dict, List, Mapping, Optional, Sequence try: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey except Exception: # pragma: no cover Ed25519PublicKey = None # type: ignore STATE_SCHEMA = "cain.trustnet.state.v2" OP_SCHEMA = "cain.trustnet.op.v2" OP_SIG_DOMAIN = "cain.trustnet.op.sig.v2" ROOT_SECTIONS = ("governance", "epoch", "identities", "capabilities", "trust_edges", "contracts", "interactions", "revocations", "incidents", "chain") IDENTITY_SCHEMA = "cain.trustnet.identity.v1" CAPABILITY_SCHEMA = "cain.trustnet.capability.v1" INTERACTION_SCHEMA = "cain.trustnet.interaction.v1" TRUST_LAYERS = ("CLAIMED", "OBSERVED", "VERIFIED", "PREDICTED", "REVOKED") # ------------------------------------------------------------------ canonical primitives (stdlib only) def canon(d: Any) -> bytes: return json.dumps(d, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") def digest(domain: str, obj: Any) -> str: return hashlib.sha256(canon({"d": domain, "o": obj})).hexdigest() def _leaf(b: bytes) -> bytes: return hashlib.sha256(b"\x00" + b).digest() def _node(a: bytes, b: bytes) -> bytes: return hashlib.sha256(b"\x01" + a + b).digest() def _mth(hs: List[bytes]) -> bytes: if len(hs) == 1: return hs[0] k = 1 while k * 2 < len(hs): k *= 2 return _node(_mth(hs[:k]), _mth(hs[k:])) def merkle_root(items: Sequence[Any]) -> str: leaves = [_leaf(canon(x)) for x in items] return hashlib.sha256(b"").hexdigest() if not leaves else _mth(leaves).hex() def state_root(state: Mapping[str, Any]) -> str: return merkle_root([[k, digest("cain.trustnet.section." + k, state[k])] for k in ROOT_SECTIONS]) def verify_ed(pub_b64: str, payload: Any, sig_b64: str) -> bool: if Ed25519PublicKey is None or not pub_b64 or not sig_b64: return False try: Ed25519PublicKey.from_public_bytes(base64.b64decode(pub_b64)).verify( base64.b64decode(sig_b64), canon(payload)) return True except (InvalidSignature, ValueError, TypeError): return False class Reject(Exception): pass def op_id(op: Mapping[str, Any]) -> str: return digest(OP_SCHEMA + ".id", op["body"]) # ------------------------------------------------------------------ state transition (v2, from spec) # Written from docs/evolution12/STATE_SPEC.md, not from cain/govnetwork/state.py. Every rule is a # table lookup or an explicit predicate so a reader can audit it line by line. def _is_int(v: Any) -> bool: return isinstance(v, int) and not isinstance(v, bool) def _alive(s: Mapping[str, Any], who: str) -> bool: rec = s["identities"].get(who) return bool(rec) and rec["revoked"] is False and who not in s["revocations"] def _quorum_signed(s: Mapping[str, Any], op: Mapping[str, Any]) -> bool: governors = s["governance"]["governors"] want = s["governance"]["threshold"] msg = {"d": OP_SIG_DOMAIN, "o": op["body"]} signed = set() for entry in op.get("sigs") or []: if not isinstance(entry, dict): continue who = entry.get("signer") if who in governors and who not in signed and verify_ed(governors[who], msg, entry.get("sig", "")): signed.add(who) return len(signed) >= want def apply(state: Dict[str, Any], op: Mapping[str, Any]) -> Dict[str, Any]: body = op.get("body") if isinstance(op, Mapping) else None if not isinstance(body, dict) or body.get("type") not in _HANDLERS: raise Reject("UNKNOWN_OP") if body.get("domain") != state["domain"]: raise Reject("WRONG_DOMAIN") oid = op_id(op) if oid in state["seen_ops"]: raise Reject("REPLAYED_OP") now = body.get("ts_ms") if not _is_int(now) or now < state["last_ts"]: raise Reject("NON_MONOTONIC_TIME") if not _is_int(body.get("epoch")) or body["epoch"] != state["epoch"]: raise Reject("STALE_EPOCH") if not _quorum_signed(state, op): raise Reject("UNAUTHORIZED_OP") import copy as _copy snap = _copy.deepcopy(state) try: _HANDLERS[body["type"]](state, body, now, oid) except Reject: state.clear() state.update(snap) raise except (KeyError, TypeError, ValueError, AttributeError): state.clear() state.update(snap) raise Reject("MALFORMED") state["seen_ops"][oid] = len(state["seen_ops"]) + 1 state["chain"] = digest("cain.trustnet.chain.v2", {"prev": state["chain"], "op": oid}) state["last_ts"] = now return state def _h_register_identity(s, b, now, oid): who = b["identity_id"] if who in s["identities"]: raise Reject("IDENTITY_EXISTS") if who in s["revocations"]: raise Reject("IDENTITY_REVOKED") if not isinstance(b["pub"], str) or b["pub"] == "": raise Reject("BAD_KEY") for rec in s["identities"].values(): if rec["pub"] == b["pub"]: raise Reject("DUPLICATE_IDENTITY_KEY") s["identities"][who] = {"identity_id": who, "kind": str(b["kind"]), "pub": b["pub"], "domain": s["domain"], "revoked": False, "at_ms": now, "op": oid} def _h_register_capability(s, b, now, oid): cid = b["cert_id"] if cid in s["capabilities"] or cid in s["revocations"]: raise Reject("CAPABILITY_EXISTS") if b["provider_id"] not in s["identities"]: raise Reject("UNKNOWN_PROVIDER") if not _alive(s, b["provider_id"]): raise Reject("PROVIDER_REVOKED") if not _is_int(b["expires_ms"]) or b["expires_ms"] <= now: raise Reject("CAPABILITY_EXPIRED") s["capabilities"][cid] = {"cert_id": cid, "provider_id": b["provider_id"], "capability": str(b["capability"]), "version": str(b.get("version", "1")), "expires_ms": b["expires_ms"], "revoked": False, "at_ms": now, "op": oid} def _h_trust_transition(s, b, now, oid): src, tgt, layer = b["source"], b["target"], b["layer"] if src not in s["identities"] or tgt not in s["identities"]: raise Reject("UNKNOWN_PARTICIPANT") if not _alive(s, src): raise Reject("PARTICIPANT_REVOKED") if layer != "REVOKED" and not _alive(s, tgt): raise Reject("PARTICIPANT_REVOKED") if layer not in TRUST_LAYERS: raise Reject("BAD_TRUST_LAYER") v = b["value_bp"] if not _is_int(v) or v < -10000 or v > 10000: raise Reject("BAD_TRUST_VALUE") ev = str(b.get("evidence", "")) raising = v > 0 if raising and src == tgt: raise Reject("SELF_VOUCH") if raising and layer in ("CLAIMED", "OBSERVED", "PREDICTED"): raise Reject("UNVERIFIED_EVIDENCE_CANNOT_RAISE_TRUST") if raising and layer == "REVOKED": raise Reject("REVOCATION_CANNOT_RAISE_TRUST") if raising and layer == "VERIFIED": rec = s["interactions"].get(ev) if rec is None or rec["result"] != "COMPLETED": raise Reject("VERIFIED_RAISE_NEEDS_COMPLETED_INTERACTION") if set(rec["parties"]) != {src, tgt}: raise Reject("EVIDENCE_NOT_ABOUT_THESE_PARTIES") for e in s["trust_edges"].values(): if (e["source"], e["target"], e["evidence"]) == (src, tgt, ev) and e["value_bp"] > 0: raise Reject("EVIDENCE_REUSED") eid = digest("cain.trustnet.edge.v2", b) if eid in s["trust_edges"]: raise Reject("TRUST_EDGE_REPLAYED") s["trust_edges"][eid] = {"edge_id": eid, "source": src, "target": tgt, "kind": str(b.get("kind", "TRUST")), "layer": layer, "value_bp": v, "evidence": ev, "at_ms": now, "op": oid} if layer == "REVOKED" and tgt not in s["revocations"]: s["revocations"][tgt] = {"kind": "IDENTITY", "reason": "TRUST_REVOKED", "at_ms": now, "op": oid} s["identities"][tgt]["revoked"] = True def _h_register_contract(s, b, now, oid): cid = b["contract_id"] if cid in s["contracts"] or cid in s["revocations"]: raise Reject("CONTRACT_EXISTS") if b.get("grants_permission"): raise Reject("CONTRACT_MUST_NOT_GRANT_PERMISSION") a, r = b["initiator"], b["receiver"] if a not in s["identities"] or r not in s["identities"]: raise Reject("UNKNOWN_PARTY") if a == r: raise Reject("SELF_CONTRACT") if not (_alive(s, a) and _alive(s, r)): raise Reject("PARTY_REVOKED") if b["cert_id"] not in s["capabilities"]: raise Reject("UNKNOWN_CAPABILITY") cert = s["capabilities"][b["cert_id"]] if cert["revoked"] or cert["cert_id"] in s["revocations"]: raise Reject("CAPABILITY_REVOKED") if cert["provider_id"] != r or cert["capability"] != b["capability"]: raise Reject("CAPABILITY_MISMATCH") until = b["valid_until_ms"] if not _is_int(until) or until <= now: raise Reject("CONTRACT_EXPIRED") if until > cert["expires_ms"]: raise Reject("CONTRACT_OUTLIVES_CAPABILITY") s["contracts"][cid] = {"contract_id": cid, "initiator": a, "receiver": r, "cert_id": cert["cert_id"], "capability": cert["capability"], "valid_until_ms": until, "at_ms": now, "op": oid} def _h_record_interaction(s, b, now, oid): iid = b["interaction_id"] if iid in s["interactions"]: raise Reject("INTERACTION_EXISTS") if b["contract_id"] not in s["contracts"]: raise Reject("UNKNOWN_CONTRACT") c = s["contracts"][b["contract_id"]] if c["contract_id"] in s["revocations"]: raise Reject("CONTRACT_REVOKED") if now > c["valid_until_ms"]: raise Reject("CONTRACT_EXPIRED") if not (_alive(s, c["initiator"]) and _alive(s, c["receiver"])): raise Reject("PARTY_REVOKED") cert = s["capabilities"][c["cert_id"]] if cert["revoked"] or now > cert["expires_ms"]: raise Reject("CAPABILITY_NOT_LIVE") if b["result"] not in ("COMPLETED", "FAILED", "DENIED"): raise Reject("BAD_RESULT") pd = b["proof_digest"] if not (isinstance(pd, str) and len(pd) == 64 and all(ch in "0123456789abcdef" for ch in pd)): raise Reject("PROOF_DIGEST_REQUIRED") s["interactions"][iid] = {"interaction_id": iid, "contract_id": c["contract_id"], "parties": sorted([c["initiator"], c["receiver"]]), "result": b["result"], "proof_digest": pd, "at_ms": now, "op": oid} _REVOKE_TABLE = {"IDENTITY": "identities", "CAPABILITY": "capabilities", "CONTRACT": "contracts", "DOMAIN": None} def _h_revoke(s, b, now, oid): kind, subj = b["kind"], b["subject"] if kind not in _REVOKE_TABLE: raise Reject("BAD_REVOCATION_KIND") table = _REVOKE_TABLE[kind] if table and subj not in s[table]: raise Reject("UNKNOWN_SUBJECT") if subj in s["revocations"]: raise Reject("ALREADY_REVOKED") s["revocations"][subj] = {"kind": kind, "reason": str(b.get("reason", "")), "at_ms": now, "op": oid} if kind == "IDENTITY": s["identities"][subj]["revoked"] = True if kind == "CAPABILITY": s["capabilities"][subj]["revoked"] = True def _h_record_incident(s, b, now, oid): if b["incident_id"] in s["incidents"]: raise Reject("INCIDENT_EXISTS") if b["affected"] not in s["identities"]: raise Reject("UNKNOWN_AFFECTED") if b["severity"] not in ("LOW", "MEDIUM", "HIGH", "CRITICAL"): raise Reject("BAD_SEVERITY") er = b["evidence_root"] if not (isinstance(er, str) and len(er) == 64 and all(ch in "0123456789abcdef" for ch in er)): raise Reject("INCIDENT_EVIDENCE_REQUIRED") s["incidents"][b["incident_id"]] = {"incident_id": b["incident_id"], "affected": b["affected"], "severity": b["severity"], "evidence_root": er, "at_ms": now, "op": oid} def _h_advance_epoch(s, b, now, oid): if not _is_int(b["new_epoch"]) or b["new_epoch"] != s["epoch"] + 1: raise Reject("EPOCH_MUST_ADVANCE_BY_ONE") s["epoch"] += 1 _HANDLERS = {"REGISTER_IDENTITY": _h_register_identity, "REGISTER_CAPABILITY": _h_register_capability, "TRUST_TRANSITION": _h_trust_transition, "REGISTER_CONTRACT": _h_register_contract, "RECORD_INTERACTION": _h_record_interaction, "REVOKE": _h_revoke, "RECORD_INCIDENT": _h_record_incident, "ADVANCE_EPOCH": _h_advance_epoch} # ------------------------------------------------------------------ artifact checks def check_identity(wrapper: Mapping[str, Any], *, expected_domain: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = wrapper.get("body") or {} if body.get("schema") != IDENTITY_SCHEMA: problems.append("BAD_IDENTITY_SCHEMA") pub = body.get("pub") if not pub or not verify_ed(pub, {"d": IDENTITY_SCHEMA, "o": body}, wrapper.get("sig", "")): problems.append("IDENTITY_SIGNATURE_INVALID") if body.get("revoked"): problems.append("IDENTITY_REVOKED") if expected_domain is not None and body.get("domain") != expected_domain: problems.append("IDENTITY_DOMAIN_MISMATCH") return {"ok": not problems, "problems": problems} def check_capability(cert: Mapping[str, Any], *, now_ms: Optional[int] = None, trusted_provider_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = cert.get("body") or {} if body.get("schema") != CAPABILITY_SCHEMA: problems.append("BAD_CAPABILITY_SCHEMA") pub = trusted_provider_pub or body.get("provider_pub") if not pub or not verify_ed(pub, {"d": CAPABILITY_SCHEMA, "o": body}, cert.get("sig", "")): problems.append("CAPABILITY_SIGNATURE_INVALID") if body.get("grants_permission"): problems.append("CAPABILITY_MUST_NOT_GRANT_PERMISSION") if body.get("revoked"): problems.append("CAPABILITY_REVOKED") if now_ms is not None and not (int(body.get("issued_ms", 0)) <= now_ms <= int(body.get("expires_ms", 0))): problems.append("CAPABILITY_TEMPORALLY_INVALID") if not body.get("evidence_root"): problems.append("CAPABILITY_EVIDENCE_MISSING") return {"ok": not problems, "problems": problems} def check_interaction(interaction: Mapping[str, Any]) -> Dict[str, Any]: """Structurally verify an E12 interaction: contract bounds, capability/provider binding, and that no field claims permission. The embedded E11 governance proof is checked by the E11 verifier.""" problems: List[str] = [] body = interaction.get("body") or {} if body.get("schema") != INTERACTION_SCHEMA: problems.append("BAD_INTERACTION_SCHEMA") for f in ("interaction_id", "contract", "capability_certificate", "sender_identity"): if not body.get(f): problems.append(f"INTERACTION_MISSING:{f}") contract = body.get("contract") or {} cap = (body.get("capability_certificate") or {}).get("body") or {} if contract.get("grants_permission") or cap.get("grants_permission"): problems.append("INTERACTION_MUST_NOT_GRANT_PERMISSION") if contract.get("capability") and cap.get("capability") != contract.get("capability"): problems.append("CONTRACT_CAPABILITY_MISMATCH") if int(contract.get("valid_until_ms", 0)) <= int(contract.get("valid_from_ms", 0)): problems.append("CONTRACT_BAD_VALIDITY") if body.get("authorized") and not body.get("governance_proof"): problems.append("AUTHORIZED_WITHOUT_GOVERNANCE_PROOF") return {"ok": not problems, "problems": problems} def verify_state(genesis_state: Mapping[str, Any], log: Sequence[Mapping[str, Any]], *, expected_root: Optional[str] = None) -> Dict[str, Any]: import copy as _copy if genesis_state.get("schema") != STATE_SCHEMA: return {"ok": False, "problems": ["BAD_STATE_SCHEMA"]} s = _copy.deepcopy(dict(genesis_state)) for i, op in enumerate(log): try: apply(s, op) except Reject as e: return {"ok": False, "problems": [f"OP_{i}_REJECTED:{e}"], "root": state_root(s)} root = state_root(s) problems: List[str] = [] if expected_root is not None and root != expected_root: problems.append("STATE_ROOT_MISMATCH") return {"ok": not problems, "problems": problems, "root": root, "ops": len(log)} # ------------------------------------------------------------------ CLI def main(argv: Optional[Sequence[str]] = None) -> int: ap = argparse.ArgumentParser(description="Clean-room CAIN E12 trust-network verifier") ap.add_argument("kind", choices=["state", "identity", "capability", "interaction"]) ap.add_argument("path") ap.add_argument("--oplog") ap.add_argument("--state-root") ap.add_argument("--domain") ap.add_argument("--now-ms", type=int) args = ap.parse_args(argv) obj = json.load(open(args.path, encoding="utf-8")) if args.kind == "state": if not args.oplog: print("state verification needs --oplog", file=sys.stderr) return 2 log = json.load(open(args.oplog, encoding="utf-8")) res = verify_state(obj, log, expected_root=args.state_root) elif args.kind == "identity": res = check_identity(obj, expected_domain=args.domain) elif args.kind == "capability": res = check_capability(obj, now_ms=args.now_ms) else: res = check_interaction(obj) if res["ok"]: print(f"CAIN E12 TRUST-NETWORK ARTIFACT VERIFIED ({args.kind})") return 0 print("CAIN E12 TRUST-NETWORK ARTIFACT REJECTED:") for p in res["problems"]: print(" -", p) return 1 # ------------------------------------------------------------------ E12 bundle runner def _bundle_sha(p): return hashlib.sha256(p.read_bytes()).hexdigest() def run_bundle(bundle_dir): import pathlib b = pathlib.Path(bundle_dir) problems, checks = [], 0 import copy as _c gen = json.loads((b / "genesis.json").read_text()) log = json.loads((b / "oplog.json").read_text()) want = json.loads((b / "expected_root.json").read_text())["state_root"] checks += 1 r = verify_state(gen, log, expected_root=want) if not r["ok"]: problems.append("oplog does not replay to the published root: " + str(r["problems"])) neg = json.loads((b / "negative_ops.json").read_text()) base = _c.deepcopy(gen) for o in log[:neg["base_log_len"]]: apply(base, o) for name, o in sorted(neg["ops"].items()): checks += 1 s = _c.deepcopy(base) try: apply(s, o) problems.append("negative op accepted: " + name) except Reject as e: if str(e) != neg["expected"][name]: problems.append(f"{name}: rejected as {e}, published {neg['expected'][name]}") for rel, want in sorted(json.loads((b / "bundle_hashes.json").read_text()).items()): checks += 1 p = b / rel if not p.exists() or _bundle_sha(p) != want: problems.append("hash mismatch: " + rel) return checks - len(problems), checks, problems if __name__ == "__main__": _p, _c, _pr = run_bundle(sys.argv[1] if len(sys.argv) > 1 else ".") print(json.dumps({"result": "INTACT" if not _pr else "BROKEN", "passed": _p, "checks": _c, "problems": _pr[:20]})) raise SystemExit(0 if not _pr else 1)