#!/usr/bin/env python3 """CAIN-42 L5 Authority Fabric — Engine B independent verifier (Prompt 2, Part 41). A SECOND, independent clean-room verifier for the L5 authority bundle. The runtime must not be the only entity able to verify its own authorization claims, and CAIN already values verifier diversity (Engine A production / Engine B clean-room / Engine C verifier-of-verifiers). This is a distinct implementation from scripts/cain42_l5/verify_authority_bundle.py, so a bug in one is not shared. It imports NOTHING from CAIN/CAIN-45/CLAWX (stdlib + `cryptography` only) and mirrors the shared canonical scheme: canon = RFC-8785-style JSON; digest(D, fields) = sha256(canon({"domain":D,**fields})); signature = Ed25519 over the ASCII bytes of that hex digest. python3 scripts/cain42_l5/verify_authority_bundle_engine_b.py BUNDLE.json [--json] Verdict: VALID (0) / INVALID (1) / INCOMPLETE (2). INCOMPLETE never counts as VALID. """ from __future__ import annotations import base64 import hashlib import json import sys import time from typing import Any, Dict, List, Optional, Tuple from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey D_IDENTITY = "CAIN42/L5-AGENT-IDENTITY/v1" D_GRANT = "CAIN42/L5-AUTHORITY-GRANT/v1" D_DELEGATION = "CAIN42/L5-DELEGATION-GRANT/v1" D_DECISION = "CAIN42/L5-AUTHORIZATION-DECISION/v1" D_REVOCATION = "CAIN42/L5-REVOCATION/v1" SCHEMA = "cain42.l5.authority-bundle.v1" R = { "IDENTITY_UNVERIFIED": "signature, canonicalization or key-binding failed", "IDENTITY_EXPIRED": "identity past expires_at", "IDENTITY_REVOKED": "identity revoked", "GRANT_SIGNATURE_INVALID": "grant signature invalid", "GRANT_EXPIRED": "grant past time_expiry", "GRANT_NOT_YET_VALID": "grant before time_start", "GRANT_REVOKED": "grant revoked", "GRANT_TRUST_BELOW_FLOOR": "current trust below the grant's trust_floor", "GRANT_POLICY_MISMATCH": "grant policy/scope is not the active policy", "DELEGATION_SIGNATURE_INVALID": "delegation signature invalid", "DELEGATION_SCOPE_EXCEEDED": "delegated scope is not within the parent scope", "DELEGATION_DEPTH_EXCEEDED": "delegation chain deeper than permitted", "DELEGATION_EXPIRY_EXCEEDS_PARENT": "delegation outlives its parent", "DELEGATION_CYCLE": "cyclic delegation", "DELEGATION_REVOKED": "delegation revoked", "DECISION_SIGNATURE_INVALID": "decision signature invalid", "DECISION_ACTION_MISMATCH": "decision binds a different action than the one presented", "DECISION_RESOURCE_MISMATCH": "decision binds a different resource", "DECISION_POLICY_MISMATCH": "decision binds a different policy version", "DECISION_GRANT_MISSING": "decision references an absent grant/delegation", "DECISION_OVER_AUTHORIZED": "decision ALLOWs an action the intersection does not permit", "DECISION_REASON_MISSING": "decision carries no machine-readable reason codes", "DECISION_EXPIRED": "decision past its own expiry", "DECISION_REPLAY": "single-use decision nonce already consumed", "SELF_AUTHORIZATION": "an issuer granted authority to itself", "MALFORMED": "object is missing required fields", } def canon(o: Any) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def h(o: Any) -> str: return hashlib.sha256(canon(o)).hexdigest() def digest(domain: str, fields: Dict[str, Any]) -> str: if "domain" in fields: raise ValueError("fields must not carry their own domain") return h({"domain": domain, **fields}) def b64d(s: str) -> bytes: return base64.b64decode(s) def verify_sig(pub_b64: str, sig_b64: str, domain: str, fields: Dict[str, Any]) -> bool: try: Ed25519PublicKey.from_public_bytes(b64d(pub_b64)).verify(b64d(sig_b64), digest(domain, fields).encode()) return True except (InvalidSignature, ValueError, TypeError, KeyError): return False # --------------------------------------------------------------------------- scope algebra def _as_set(v: Any) -> frozenset: if v is None: return frozenset() if isinstance(v, str): return frozenset({v}) return frozenset(v) def _cap_scope(scope: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: caps = scope.get("capabilities") if isinstance(scope, dict) else None if caps is None: if isinstance(scope, dict) and scope.get("capability"): caps = {scope["capability"]: {k: v for k, v in scope.items() if k != "capability"}} else: caps = {} out: Dict[str, Dict[str, Any]] = {} for name, body in caps.items(): body = body or {} out[name] = {"resources": _as_set(body.get("resources")), "operations": _as_set(body.get("operations")), "constraints": {k: v for k, v in body.items() if k not in ("resources", "operations")}} return out def _constraint_within(child: Any, parent: Any) -> bool: if parent is None: return True if child is None: return False if isinstance(parent, (int, float)) and isinstance(child, (int, float)): return child <= parent if isinstance(parent, str) and isinstance(child, str): return child == parent if isinstance(parent, (list, tuple, set)): return set(child or []) <= set(parent) return child == parent def scope_within(child: Dict[str, Any], parent: Dict[str, Any]) -> Tuple[bool, Optional[str]]: cs, ps = _cap_scope(child), _cap_scope(parent) for cap, cbody in cs.items(): if cap not in ps: return False, f"capability {cap} not held by parent" pbody = ps[cap] if not cbody["resources"] <= pbody["resources"]: return False, f"resource scope of {cap} exceeds parent" if pbody["operations"] and not cbody["operations"] <= pbody["operations"]: return False, f"operation scope of {cap} exceeds parent" for k, v in cbody["constraints"].items(): if not _constraint_within(v, pbody["constraints"].get(k)): return False, f"constraint {k}={v!r} on {cap} is weaker than parent {pbody['constraints'].get(k)!r}" return True, None def intersect(scopes: List[Dict[str, Any]]) -> Dict[str, Any]: acc: Dict[str, Dict[str, Any]] = {} for scope in scopes: for cap, body in _cap_scope(scope).items(): if cap not in acc: acc[cap] = {"resources": set(body["resources"]), "operations": set(body["operations"]), "constraints": dict(body["constraints"])} continue cur = acc[cap] cur["resources"] &= set(body["resources"]) cur["operations"] &= set(body["operations"]) for k, v in body["constraints"].items(): if k not in cur["constraints"]: cur["constraints"][k] = v else: pv = cur["constraints"][k] if isinstance(pv, (int, float)) and isinstance(v, (int, float)): cur["constraints"][k] = min(pv, v) elif isinstance(pv, (list, tuple, set)) and isinstance(v, (list, tuple, set)): cur["constraints"][k] = sorted(set(pv) & set(v)) elif pv != v: cur["constraints"][k] = None return {"capabilities": {c: {"resources": sorted(b["resources"]), "operations": sorted(b["operations"]), **b["constraints"]} for c, b in acc.items()}} def action_permitted(effective: Dict[str, Any], action: Dict[str, Any]) -> Tuple[bool, str]: caps = _cap_scope(effective) cap = action.get("capability") if cap not in caps: return False, "capability not in effective authority" body = caps[cap] if body["resources"] and action.get("resource") not in body["resources"]: return False, "resource out of effective scope" if body["operations"] and action.get("operation") not in body["operations"]: return False, "operation out of effective scope" if action.get("environment") is not None and body["constraints"].get("environment") is not None: if action["environment"] != body["constraints"]["environment"]: return False, "environment out of effective scope" if action.get("tenant") is not None and body["constraints"].get("tenant") is not None: if action["tenant"] != body["constraints"]["tenant"]: return False, "cross-tenant access" return True, "within effective authority" class Findings: def __init__(self) -> None: self.items: List[Dict[str, Any]] = [] def add(self, reason: str, obj: Optional[str] = None, detail: str = "") -> None: self.items.append({"reason": reason, "object": obj, "detail": detail}) def _by_id(objects: List[Dict[str, Any]], kind: str) -> Dict[str, Dict[str, Any]]: out: Dict[str, Dict[str, Any]] = {} for o in objects: if o.get("type") != kind: continue f = o.get("fields") or {} gid = f.get("grant_id") or f.get("delegation_id") or f.get("decision_id") or f.get("agent_id") if gid: out[gid] = o return out def verify_bundle(bundle: Dict[str, Any]) -> Dict[str, Any]: now = float(bundle.get("now") or time.time()) objs = bundle.get("objects") or [] trust = bundle.get("trust") or {} findings = Findings() if bundle.get("schema") != SCHEMA: findings.add("MALFORMED", None, f"schema must be {SCHEMA}") for o in objs: t, dom, f = o.get("type"), o.get("domain"), o.get("fields") if not isinstance(f, dict) or not dom: findings.add("MALFORMED", t, "missing domain/fields") continue if not verify_sig(o.get("signer_public_key_b64", ""), o.get("signature_b64", ""), dom, f): findings.add(f"{t}_SIGNATURE_INVALID" if t != "AGENT_IDENTITY" else "IDENTITY_UNVERIFIED", f.get("agent_id") or f.get("grant_id") or f.get("delegation_id") or f.get("decision_id")) revoked = set() for rv in bundle.get("revocations") or []: if not verify_sig(rv.get("signer_public_key_b64", ""), rv.get("signature_b64", ""), rv.get("domain", D_REVOCATION), rv.get("fields", {})): findings.add("MALFORMED", rv.get("fields", {}).get("subject_id"), "revocation signature invalid") continue revoked.add(rv.get("fields", {}).get("subject_id")) identities = _by_id(objs, "AGENT_IDENTITY") for aid, o in identities.items(): f = o["fields"] if f.get("public_key") and o.get("signer_public_key_b64") != f.get("public_key"): findings.add("IDENTITY_UNVERIFIED", aid, "signer key != declared public_key (key substitution)") if f.get("status") not in (None, "ACTIVE"): findings.add("IDENTITY_UNVERIFIED", aid, f"status={f.get('status')}") if f.get("expires_at") and now >= float(f["expires_at"]): findings.add("IDENTITY_EXPIRED", aid) if aid in revoked: findings.add("IDENTITY_REVOKED", aid) grants = _by_id(objs, "AUTHORITY_GRANT") active_policy = (bundle.get("policy") or {}).get("policy_version") policy_scope = bundle.get("policy_scope") for gid, o in grants.items(): f = o["fields"] if f.get("issuer_id") and f.get("issuer_id") == f.get("agent_id"): findings.add("SELF_AUTHORIZATION", gid, "a grant's issuer equals its subject") if policy_scope is not None: within, why = scope_within({"capabilities": f.get("capabilities") or {}}, {"capabilities": policy_scope}) if not within: findings.add("GRANT_POLICY_MISMATCH", gid, f"grant exceeds policy scope: {why}") if f.get("time_start") and now < float(f["time_start"]): findings.add("GRANT_NOT_YET_VALID", gid) if f.get("time_expiry") and now >= float(f["time_expiry"]): findings.add("GRANT_EXPIRED", gid) if gid in revoked or f.get("revocation_reference") in revoked: findings.add("GRANT_REVOKED", gid) floor = f.get("trust_floor") subj = f.get("agent_id") or f.get("principal_id") if floor is not None and subj in trust and trust[subj] < floor: findings.add("GRANT_TRUST_BELOW_FLOOR", gid, f"trust {trust[subj]} < floor {floor}") if active_policy is not None and f.get("policy_version") not in (None, active_policy): findings.add("GRANT_POLICY_MISMATCH", gid, f"grant {f.get('policy_version')} != active {active_policy}") delegations = _by_id(objs, "DELEGATION_GRANT") for did, o in delegations.items(): f = o["fields"] parent_id = f.get("parent_grant") or f.get("parent_grant_id") parent = grants.get(parent_id) or delegations.get(parent_id) if parent is None: findings.add("DECISION_GRANT_MISSING", did, f"parent {parent_id} absent") continue within, why = scope_within({"capabilities": f.get("delegated_capabilities") or f.get("scope") or {}}, {"capabilities": parent["fields"].get("capabilities") or parent["fields"].get("scope") or {}}) if not within: findings.add("DELEGATION_SCOPE_EXCEEDED", did, why or "") max_depth = f.get("maximum_depth") depth = f.get("delegation_depth", 0) if max_depth is not None and int(depth) > int(max_depth): findings.add("DELEGATION_DEPTH_EXCEEDED", did, f"depth {depth} > {max_depth}") pe = parent["fields"].get("time_expiry") if f.get("expires_at") and pe and float(f["expires_at"]) > float(pe): findings.add("DELEGATION_EXPIRY_EXCEEDS_PARENT", did) if did in revoked: findings.add("DELEGATION_REVOKED", did) for did in delegations: seen, cur = set(), did while cur: if cur in seen: findings.add("DELEGATION_CYCLE", did) break seen.add(cur) o = delegations.get(cur) cur = (o["fields"].get("parent_grant") or o["fields"].get("parent_grant_id")) if o else None decisions = _by_id(objs, "AUTHORIZATION_DECISION") checked_decisions = 0 consumed = set(bundle.get("consumed_nonces") or []) for dec_id, o in decisions.items(): f = o["fields"] checked_decisions += 1 if f.get("expiry") and now >= float(f["expiry"]): findings.add("DECISION_EXPIRED", dec_id) if f.get("nonce") and f["nonce"] in consumed: findings.add("DECISION_REPLAY", dec_id) if not f.get("reason_codes"): findings.add("DECISION_REASON_MISSING", dec_id) presented = bundle.get("presented_action") if presented is not None: if f.get("action_hash") != h(presented): findings.add("DECISION_ACTION_MISMATCH", dec_id) if f.get("resource_hash") and presented.get("resource") is not None: if f["resource_hash"] != h(presented.get("resource")): findings.add("DECISION_RESOURCE_MISMATCH", dec_id) if active_policy is not None and f.get("policy_version") not in (None, active_policy): findings.add("DECISION_POLICY_MISMATCH", dec_id) if str(f.get("decision", "")).upper() == "ALLOW" and presented is not None: refs = (f.get("effective_authority") or {}).get("sources") or [] scopes: List[Dict[str, Any]] = [] for rid in refs: src = grants.get(rid) or delegations.get(rid) if src is None: findings.add("DECISION_GRANT_MISSING", dec_id, f"source {rid} absent") continue scopes.append({"capabilities": src["fields"].get("capabilities") or src["fields"].get("delegated_capabilities") or {}}) if scopes: ok, why = action_permitted(intersect(scopes), presented) if not ok: findings.add("DECISION_OVER_AUTHORIZED", dec_id, why) status = "INVALID" if findings.items else ("INCOMPLETE" if checked_decisions == 0 else "VALID") return {"status": status, "findings": findings.items, "objects": len(objs), "identities": len(identities), "grants": len(grants), "delegations": len(delegations), "decisions": checked_decisions, "revocations": len(revoked)} # --------------------------------------------------------------------------- runtime bundle format # A second, independent verification path for the flat bundle emitted by the runtime # (schema "cain42.l5.authority_bundle.v1"): identity/grant/delegation/decision objects with the # signature and digest inline. The field lists are the artifact's own schema, not verification logic; # the checks below are this verifier's own semantics (its own intersection and scope algebra). RUNTIME_SCHEMA = "cain42.l5.authority_bundle.v1" R_IDENTITY = ("identity_version", "agent_id", "agent_type", "model_family", "model_identifier", "model_version", "runtime_identifier", "runtime_version", "owner_principal", "organization", "created_at", "expires_at", "status", "capabilities", "public_key", "key_algorithm", "attestation_reference", "policy_binding", "trust_reference", "delegation_root", "issuer", "issued_at") R_GRANT = ("grant_id", "principal_id", "agent_id", "issuer_id", "parent_grant_id", "scope", "capabilities", "resources", "operations", "constraints", "risk_limit", "budget_limit", "delegation_limit", "time_start", "time_expiry", "max_duration", "renewal_policy", "policy_version", "trust_floor", "environment_constraints", "revocation_reference", "single_use", "status") R_DELEGATION = ("delegation_id", "delegator", "delegate", "parent_grant_id", "delegated_capabilities", "delegated_resources", "constraints", "depth", "maximum_depth", "time_start", "time_expiry", "policy_version") R_DECISION = ("decision_id", "action_id", "agent_id", "requested_capability", "resource", "operation", "effective_authority", "policy_version", "trust_state", "delegation_chain", "risk_state", "decision", "reason_codes", "constraints", "timestamp", "expiry", "action_hash", "parameters_hash", "resource_hash", "context_hash", "policy_hash", "evidence_reference") def _runtime_body(obj: Dict[str, Any], names) -> Dict[str, Any]: out = {} for n in names: v = (obj or {}).get(n) out[n] = sorted(v) if isinstance(v, list) else v return out def _pattern_within(child: str, parent: str) -> bool: return child == parent or (str(parent).endswith("*") and str(child).startswith(str(parent)[:-1])) def _set_within_any(children, parents) -> bool: """Every child resource must be covered by some parent pattern (the runtime's '*' semantics).""" return all(any(_pattern_within(c, p) for p in parents) for c in children) def verify_runtime_bundle(b: Dict[str, Any]) -> Dict[str, Any]: """Independent verification of the runtime's flat authority bundle. Same primitives, this verifier's own semantics; a bug in the runtime's own verifier is not shared here.""" f = Findings() def chk(name: str, ok: bool, detail: str = "") -> None: if not ok: f.add(name, None, detail) now = float(b.get("now") or time.time()) issuer = b.get("issuer_public_key", "") ident_o, grant_o, deleg_o, dec_o = (b.get("identity") or {}), (b.get("grant") or {}), \ (b.get("delegation") or {}), (b.get("decision") or {}) ident = _runtime_body(ident_o, R_IDENTITY) grant = _runtime_body(grant_o, R_GRANT) deleg = _runtime_body(deleg_o, R_DELEGATION) dec = _runtime_body(dec_o, R_DECISION) # signatures and digests chk("IDENTITY_SIGNATURE_INVALID", verify_sig(issuer, ident_o.get("identity_signature", ""), D_IDENTITY, ident)) chk("IDENTITY_DIGEST_MISMATCH", digest(D_IDENTITY, ident) == ident_o.get("digest")) chk("IDENTITY_UNVERIFIED", ident.get("public_key") == b.get("agent_public_key"), "agent public key binding") chk("GRANT_SIGNATURE_INVALID", verify_sig(issuer, grant_o.get("signature", ""), D_GRANT, grant)) chk("GRANT_DIGEST_MISMATCH", digest(D_GRANT, grant) == grant_o.get("digest")) chk("DELEGATION_SIGNATURE_INVALID", verify_sig(issuer, deleg_o.get("signature", ""), D_DELEGATION, deleg)) chk("DELEGATION_DIGEST_MISMATCH", digest(D_DELEGATION, deleg) == deleg_o.get("digest")) chk("DECISION_SIGNATURE_INVALID", verify_sig(issuer, dec_o.get("signature", ""), D_DECISION, dec)) chk("DECISION_DIGEST_MISMATCH", digest(D_DECISION, dec) == dec_o.get("digest")) # temporal (this verifier's own rule: a bundle carrying an explicit now must not be expired) if grant.get("time_expiry") and now >= float(grant["time_expiry"]): f.add("GRANT_EXPIRED", grant.get("grant_id")) if deleg.get("time_expiry") and now >= float(deleg["time_expiry"]): f.add("DELEGATION_EXPIRED", deleg.get("delegation_id")) if dec.get("expiry") and now >= float(dec["expiry"]): f.add("DECISION_EXPIRED", dec.get("decision_id")) # delegation cannot escalate: map the flat lists to this verifier's nested scope algebra gcaps = set(grant.get("capabilities") or []) gres = list(grant.get("resources") or []) dcaps = set(deleg.get("delegated_capabilities") or []) dres = list(deleg.get("delegated_resources") or []) if not dcaps <= gcaps: f.add("DELEGATION_SCOPE_EXCEEDED", deleg.get("delegation_id"), "capability beyond parent") if not _set_within_any(dres, gres): f.add("DELEGATION_SCOPE_EXCEEDED", deleg.get("delegation_id"), "resource beyond parent") if int(deleg.get("depth", 1)) > int(deleg.get("maximum_depth", 0)): f.add("DELEGATION_DEPTH_EXCEEDED", deleg.get("delegation_id")) # decision must be action-bound and within the independently computed intersection if not dec.get("action_hash"): f.add("DECISION_ACTION_MISMATCH", dec.get("decision_id"), "no action hash") if str(dec.get("decision", "")).upper() == "ALLOW": if dec.get("requested_capability") not in gcaps: f.add("DECISION_OVER_AUTHORIZED", dec.get("decision_id"), "capability not in grant") eff = dec.get("effective_authority") or {} expected_caps = sorted(gcaps & dcaps) if sorted(eff.get("capabilities") or []) != expected_caps: f.add("DECISION_OVER_AUTHORIZED", dec.get("decision_id"), f"effective caps {sorted(eff.get('capabilities') or [])} != intersection {expected_caps}") expected_res = sorted([r for r in dres if _set_within_any([r], gres)]) if sorted(eff.get("resources") or []) != expected_res: f.add("DECISION_OVER_AUTHORIZED", dec.get("decision_id"), f"effective resources {sorted(eff.get('resources') or [])} != narrowest {expected_res}") # a revoked grant's decision must be a denial rg = b.get("revoked_grant") if rg: chk("GRANT_SIGNATURE_INVALID", verify_sig(issuer, rg.get("signature", ""), D_GRANT, _runtime_body(rg, R_GRANT)), "revoked grant") rd = _runtime_body(b.get("revoked_decision") or {}, R_DECISION) if b.get("revoked_decision"): if str(rd.get("decision", "")).upper() == "ALLOW": f.add("DECISION_OVER_AUTHORIZED", rd.get("decision_id"), "revoked decision is ALLOW") if not (rd.get("reason_codes") or []): f.add("DECISION_REASON_MISSING", rd.get("decision_id")) status = "INVALID" if f.items else "VALID" return {"status": status, "findings": f.items, "schema": b.get("schema"), "kind": "runtime-authority-bundle"} def main(argv: List[str]) -> int: if len(argv) < 2: print(__doc__) return 2 try: bundle = json.load(open(argv[1], encoding="utf-8")) except Exception as exc: # pragma: no cover print(json.dumps({"status": "INVALID", "findings": [{"reason": "MALFORMED", "detail": str(exc)}]})) return 1 if str(bundle.get("schema", "")).startswith("cain42.l5.authority_bundle"): res = verify_runtime_bundle(bundle) else: res = verify_bundle(bundle) if "--json" in argv: print(json.dumps(res, indent=1)) else: counts = {k: res[k] for k in ("objects", "identities", "grants", "delegations", "decisions", "revocations") if k in res} extra = " ".join(f"{k}={v}" for k, v in counts.items()) print(f"authority bundle: {res['status']} {extra}".rstrip()) for x in res["findings"]: print(f" {x['reason']:34s} {x.get('object') or ''} {x.get('detail') or ''}") return {"VALID": 0, "INVALID": 1}.get(res["status"], 2) if __name__ == "__main__": raise SystemExit(main(sys.argv))