#!/usr/bin/env python3 """cain-mission-journal-verify — CLEAN-ROOM verifier for a CAIN Evolution 16 mission journal. ZERO CAIN IMPORTS. Needs Python 3 and the `cryptography` package only. python verify.py --trust [--head | --no-head] [--require-final] It reads the journal a mission runtime wrote and the public keys YOU supply, and recomputes, without the runtime: chain every entry commits to the one before it and is signed by the pinned runtime key genesis the mission is signed by the pinned mission authority, the plan by a pinned plan authority and bound to that mission; the keys you supplied are the ones the journal was started under contract every mission version names the version before it, none is installed twice, each is signed by the mission authority; this verifier classifies the change itself, and a version that widens the mission carries the principal's signature over exactly that widening, under a key other than the authority's admission every dispatched action had, at that moment: a request signed by the plan's agent for that action; an authorization signed by a pinned issuer for that exact action digest, unexpired and never used before; a human approval signed by the pinned approver when the plan demands one; and a delegation chain from the mission principal to the agent in which every hop was signed by its parent, unexpired, unrevoked and covers the action class and resource dispatch the intent was written under the contract version the admission was made under, before expiry, not after cancellation, with the idempotency key this verifier derives, and never twice for an outcome that was still unresolved (unless the action is read-only or an operation you declared idempotent) effect an effect counts as confirmed, absent or compensated only on an observation signed by a pinned observer; a pinned observation that contradicts a confirmed effect reopens it (INCOMPLETE), and is a failure if the mission had already been recorded complete completion the mission was not finalized as complete while any dispatch was unresolved Verdicts (exit code) VERIFIED (0) every required check was performed and passed, and nothing is left open FAILED (1) at least one check failed INCOMPLETE (3) nothing failed, but a required check could not be performed or the record is not closed: unresolved dispatches, no witness-held head (see --head), mission not final with --require-final UNSUPPORTED (4) the journal uses a schema, entry kind or operation this verifier does not know UNAVAILABLE (5) the journal or the trust file cannot be read What VERIFIED does not mean. Signatures show who signed which bytes, not that the bytes are true: an observer that signs "the row is there" when it is not is NOT detected here. The policy decision behind an authorization is taken from the pinned issuer's signature, not re-derived. The keys are only as good as the channel you got them from. The journal was produced by the runtime being verified, and this file was written by the same project. Without --head a deleted tail of the journal cannot be noticed, which is why its absence yields INCOMPLETE unless you pass --no-head. """ from __future__ import annotations import argparse import base64 import fnmatch import hashlib import json import sys from typing import Any, Dict, List, Mapping, Optional, Sequence from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 D_JOURNAL = "CAIN42/EVOLUTION16/MISSION-JOURNAL/v1" D_SUCCESSION = "CAIN42/EVOLUTION16/MISSION-SUCCESSION/v1" D_SUCCESSION_APPROVAL = "CAIN42/EVOLUTION16/MISSION-SUCCESSION-APPROVAL/v1" D_MISSION = "CAIN42/MISSION-CONTROL/MISSION-SPEC/v1" D_PLAN = "CAIN42/MISSION-CONTROL/MISSION-PLAN/v1" D_AUTH = "CAIN42/ACTION-COMMIT/AUTHORIZATION/v1" D_APPROVAL = "CAIN42/ACTION-COMMIT/APPROVAL/v1" D_AGENT_REQ = "CAIN42/EVOLUTION15/AGENT-REQUEST/v1" D_OBSERVATION = "CAIN42/EVOLUTION15/OBSERVATION/v1" D_CANCEL = "CAIN42/EVOLUTION15/PLAN-AMENDMENT-APPROVAL/v1/CANCEL" D_COMP_ACTION = "CAIN42/MISSION-CONTROL/COMP-ACTION/v1" JOURNAL_SCHEMA = "CAIN42/EVOLUTION16/JOURNAL/v1" TRUST_SCHEMA = "CAIN42/EVOLUTION16/TRUST-PUBLIC/v1" GENESIS_PREV = "0" * 64 KNOWN_KINDS = ("GENESIS", "INPUT", "DECISION", "TX_INTENT", "TX_RESULT", "TX_AMBIGUOUS", "TX_REFUSED", "COMP_INTENT", "COMP_RESULT", "RECOVERY") KNOWN_OPS = ("delegate", "revoke_agent", "remember", "attest_environment", "promote", "amend_plan", "policy_changed", "suspend", "resume", "propose", "observe", "observe_fact", "observe_criterion", "report_success", "finalize", "cancel", "update_mission", "reconcile_absent", "confirm_compensation") UNRESOLVED = ("INTENT", "AMBIGUOUS", "REPORTED", "CONTRADICTED", "COMP_INTENT", "COMP_REPORTED", "COMP_AMBIGUOUS", "COMPENSATION_FAILED") EXIT = {"VERIFIED": 0, "FAILED": 1, "INCOMPLETE": 3, "UNSUPPORTED": 4, "UNAVAILABLE": 5} def canon(o: Any) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def digest(domain: str, fields: Mapping[str, Any]) -> str: return hashlib.sha256(canon({"domain": domain, **fields})).hexdigest() def sig_ok(pub: Any, sig: Any, domain: str, fields: Mapping[str, Any]) -> bool: try: ed25519.Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify( base64.b64decode(sig), digest(domain, fields).encode()) return True except (InvalidSignature, ValueError, TypeError): return False def within(pattern: str, parents: Sequence[str]) -> bool: return any(pattern == p or fnmatch.fnmatchcase(pattern, p) for p in parents) SEVERITY = {"LOW": 0, "MEDIUM": 1, "HIGH": 2, "CRITICAL": 3} def widenings(old: Mapping[str, Any], new: Mapping[str, Any]) -> Optional[List[str]]: """This verifier's own reading of what a new mission version widens. None = a different mission.""" if any(old.get(f) != new.get(f) for f in ("mission_id", "tenant", "principal")): return None w: List[str] = [] w += [f"SCOPE_ADDED:{s}" for s in new.get("resource_scope") or [] if not within(s, old.get("resource_scope") or [])] for f in ("allowed_action_classes", "data_classifications", "participating_agents", "policy_versions"): w += [f"{f.upper()}_ADDED:{x}" for x in sorted(set(new.get(f) or []) - set(old.get(f) or []))] for f in ("required_approvals", "required_evidence", "cancellation_conditions", "environmental_assumptions", "dependencies"): w += [f"{f.upper()}_REMOVED:{x}" for x in sorted(set(old.get(f) or []) - set(new.get(f) or []))] ob, nb = old.get("budget") or {}, new.get("budget") or {} w += [f"BUDGET_RAISED:{d}" for d in sorted(set(ob) | set(nb)) if float(nb.get(d, 0) or 0) > float(ob.get(d, 0) or 0)] if float(new.get("expires_at") or 0) > float(old.get("expires_at") or 0): w.append("EXPIRY_EXTENDED") op = {p["outcome_id"]: p for p in old.get("prohibited_outcomes") or []} np_ = {p["outcome_id"]: p for p in new.get("prohibited_outcomes") or []} for oid in sorted(op): if oid not in np_: w.append(f"PROHIBITION_REMOVED:{oid}") elif SEVERITY.get(str(np_[oid].get("severity")).upper(), 0) < SEVERITY.get(str(op[oid].get("severity")).upper(), 0): w.append(f"PROHIBITION_WEAKENED:{oid}") if old.get("objective") != new.get("objective"): w.append("OBJECTIVE_CHANGED") if old.get("success_criteria") != new.get("success_criteria"): w.append("SUCCESS_CRITERIA_CHANGED") if str(old.get("failure_policy")) != str(new.get("failure_policy")) and str(new.get("failure_policy")).upper() != "ABORT": w.append("FAILURE_POLICY_CHANGED") return sorted(w) class Verifier: def __init__(self, trust: Mapping[str, Any]) -> None: self.t = trust self.failures: List[Dict[str, Any]] = [] self.unsupported: List[str] = [] self.performed: Dict[str, int] = {} self.mission: Dict[str, Any] = {} self.mission_digest = "" self.contract: List[Dict[str, Any]] = [] self.plan: Dict[str, Dict[str, Any]] = {} self.edges: Dict[str, Dict[str, Any]] = {} self.revoked: set = set() self.approvals: Dict[str, bool] = {} self.admissions: Dict[str, Dict[str, Any]] = {} self.used_auth: set = set() self.tx: Dict[str, Dict[str, Any]] = {} self.contradicted: List[str] = [] self.cancelled_at: Optional[int] = None self.final: Optional[str] = None self.pending_input: Optional[Dict[str, Any]] = None self.hashes: List[str] = [] # ---- bookkeeping ----------------------------------------------------------------------------------- def check(self, name: str, ok: bool, seq: Any, detail: str = "") -> bool: self.performed[name] = self.performed.get(name, 0) + 1 if not ok: self.failures.append({"check": name, "seq": seq, "detail": detail}) return ok # ---- records --------------------------------------------------------------------------------------- def signed_record(self, rec: Any, domain: str, pinned: Sequence[str]) -> bool: """A {"body","digest","signature","signer_pub"} record: digest recomputed, signer pinned, signature valid.""" if not isinstance(rec, Mapping) or not isinstance(rec.get("body"), Mapping): return False return (rec.get("digest") == digest(domain, rec["body"]) and rec.get("signer_pub") in pinned and sig_ok(rec.get("signer_pub"), rec.get("signature"), domain, rec["body"])) def set_plan(self, rec: Mapping[str, Any]) -> None: self.plan = {a["action_id"]: a for a in rec["body"].get("actions", [])} def observation(self, obs: Any) -> Optional[Dict[str, Any]]: b = dict(obs.get("body") or {}) if isinstance(obs, Mapping) else {} pub = (self.t.get("observer_pubs") or {}).get(str(b.get("observer"))) return b if pub and sig_ok(pub, obs.get("signature"), D_OBSERVATION, b) else None def recovery_class(self, a: Mapping[str, Any]) -> str: r = str(a.get("reversibility", "")).upper() if r == "READ_ONLY" and not a.get("external_side_effect"): return "READ_ONLY" if f"{a.get('target_system')}:{a.get('operation')}" in (self.t.get("idempotent_operations") or []): return "IDEMPOTENT_WRITE" return r if r in ("REVERSIBLE", "COMPENSATABLE") else "IRREVERSIBLE" def chain_live(self, agent: str, at: float, cls: str, resource: str, seen: Optional[set] = None) -> bool: """Is there a delegation chain principal -> agent, live at `at`, every hop covering this class and resource?""" if agent in self.revoked: return False if agent == self.mission.get("principal"): return cls in (self.mission.get("allowed_action_classes") or []) and \ within(resource, self.mission.get("resource_scope") or []) seen = seen or set() if agent in seen: return False seen.add(agent) for e in self.edges.values(): if e["child"] == agent and e["expires_at"] > at and cls in e["capabilities"] and \ within(resource, e["scope"]) and self.chain_live(e["parent"], at, cls, resource, seen): return True return False # ---- entries --------------------------------------------------------------------------------------- def genesis(self, e: Mapping[str, Any]) -> None: b, seq = e["body"], e["seq"] if b.get("schema") != JOURNAL_SCHEMA: self.unsupported.append(f"journal schema {b.get('schema')!r}") return self.check("genesis.trust_matches_supplied_keys", b.get("trust_digest") == hashlib.sha256(canon(dict(self.t))).hexdigest(), seq, "the journal was started under other pinned keys than the trust file supplied") m, p = b.get("mission"), b.get("plan") if self.check("genesis.mission_signed_by_authority", self.signed_record(m, D_MISSION, [self.t.get("mission_authority_pub")]), seq): self.mission, self.mission_digest = dict(m["body"]), m["digest"] self.contract = [{"seq": 0, "digest": m["digest"]}] if self.check("genesis.plan_signed_and_bound", self.signed_record(p, D_PLAN, self.t.get("plan_authority_pubs") or []) and p["body"].get("mission_digest") == self.mission_digest, seq): self.set_plan(p) def decided(self, inp: Mapping[str, Any], dec: Mapping[str, Any]) -> None: op, a, at, seq = inp["body"]["op"], inp["body"]["args"], inp["at"], inp["seq"] d = dec["body"].get("decision") or {} verdict = d.get("decision") self.check("decision.digest", dec["body"].get("decision_digest") == hashlib.sha256(canon(d)).hexdigest(), dec["seq"], "the recorded decision does not hash to the recorded decision digest") if op == "delegate" and verdict == "ALLOW": r = (a.get("req") or {}).get("request") or {} body = r.get("body") or {} parent = str(r.get("agent")) pub = (self.t.get("agent_pubs") or {}).get(parent) if self.check("delegation.signed_by_parent", bool(pub) and r.get("kind") == "DELEGATE" and sig_ok(pub, (a.get("req") or {}).get("signature"), D_AGENT_REQ, r), seq, parent): self.edges[str(body.get("delegation_id"))] = { "parent": parent, "child": str(body.get("child")), "expires_at": float(body.get("expires_at") or 0), "capabilities": list(body.get("capabilities") or []), "scope": list(body.get("task_scope") or [])} elif op == "revoke_agent": self.revoked.add(str(a.get("agent"))) elif op == "amend_plan" and verdict == "ALLOW": p = a.get("plan_rec") if self.check("plan.amendment_signed_and_bound", self.signed_record(p, D_PLAN, self.t.get("plan_authority_pubs") or []) and p["body"].get("mission_digest") == self.mission_digest, seq): self.set_plan(p) elif op == "update_mission" and verdict == "ALLOW": self.succession(a, d, seq) elif op == "cancel" and verdict == "ALLOW": c = a.get("cancellation") or {} b = c.get("body") or {} pub = (self.t.get("approver_pubs") or {}).get(str(b.get("approver"))) self.check("cancel.signed_by_pinned_approver", bool(pub) and b.get("cancel") == self.mission.get("mission_id") and sig_ok(pub, c.get("signature"), D_CANCEL, b), seq) self.cancelled_at = seq elif op == "propose": self.proposal(a, d, at, seq) elif op == "observe" and verdict != "ALLOW": # a refused observation is still evidence if a pinned observer signed it: one that contradicts a confirmed # effect reopens that dispatch, whatever the runtime recorded b = self.observation(a.get("observation")) aid = str((b or {}).get("action_id")) if b is not None and not b.get("matches") and self.tx.get(aid, {}).get("state") == "CONFIRMED": self.tx[aid]["state"] = "CONTRADICTED" self.contradicted.append(aid) elif op == "observe" and verdict == "ALLOW": b = self.observation(a.get("observation")) if self.check("effect.observation_signed_by_pinned_observer", b is not None, seq): aid = str(b.get("action_id")) if aid in self.tx: ok_state = self.tx[aid]["state"] in ("REPORTED", "AMBIGUOUS") self.check("effect.observation_follows_a_dispatch", ok_state, seq, aid) self.tx[aid]["state"] = "CONFIRMED" if b.get("matches") else "CONTRADICTED" else: self.check("effect.observation_follows_a_dispatch", False, seq, f"{aid}: an effect was accepted for an action with no journaled dispatch") elif op == "reconcile_absent" and verdict == "ALLOW": b = self.observation(a.get("observation")) ok = b is not None and not b.get("matches") and b.get("observed") == "ABSENT" if self.check("effect.absence_signed_by_pinned_observer", ok, seq): aid = str(b.get("action_id")) if self.check("effect.absence_settles_an_ambiguous_dispatch", self.tx.get(aid, {}).get("state") == "AMBIGUOUS", seq, aid): self.tx[aid]["state"] = "NOT_APPLIED" self.admissions.pop(aid, None) elif op == "confirm_compensation": b = self.observation(a.get("observation")) if b is None: self.check("compensation.confirmed_only_by_pinned_observer", verdict != "ALLOW", seq) return target = str(b.get("action_id")) aid = target[:-len("#compensation")] if target.endswith("#compensation") else "" t = self.tx.get(aid) if t is None or t["state"] not in ("COMP_REPORTED", "COMP_AMBIGUOUS"): self.check("compensation.confirmed_only_by_pinned_observer", verdict != "ALLOW", seq, aid) return self.check("compensation.confirmed_only_by_pinned_observer", (verdict == "ALLOW") == bool(b.get("matches")), seq, "the recorded outcome of the compensation contradicts the signed observation") t["state"] = "COMPENSATED" if b.get("matches") else "COMPENSATION_FAILED" elif op == "finalize": if verdict == "ALLOW": open_tx = sorted(x for x, t in self.tx.items() if t["state"] in UNRESOLVED) self.check("completion.no_unresolved_dispatch", not open_tx, seq, ",".join(open_tx)) self.final = str(d.get("reason")) def succession(self, a: Mapping[str, Any], d: Mapping[str, Any], seq: int) -> None: m, s, p, ap = a.get("mission_rec"), a.get("succession") or {}, a.get("plan_rec"), a.get("approval") sb = s.get("body") or {} auth_pub = self.t.get("mission_authority_pub") if not self.check("contract.version_signed_by_authority", self.signed_record(m, D_MISSION, [auth_pub]) and sig_ok(auth_pub, s.get("signature"), D_SUCCESSION, sb) and sb.get("new_digest") == m.get("digest"), seq): return head = self.contract[-1] self.check("contract.names_current_head", sb.get("prev_digest") == head["digest"] and sb.get("seq") == head["seq"] + 1, seq, "a version was installed over a base that was not the head") self.check("contract.no_version_installed_twice", m["digest"] not in {c["digest"] for c in self.contract}, seq) w = widenings(self.mission, m["body"]) if not self.check("contract.same_mission", w is not None, seq): return if w: principal = str(self.mission.get("principal")) ppub = (self.t.get("approver_pubs") or {}).get(principal) ab = (ap or {}).get("body") or {} if isinstance(ap, Mapping) else {} ok = (bool(ppub) and ppub != auth_pub and ab.get("approver") == principal and ab.get("mission_id") == self.mission.get("mission_id") and ab.get("seq") == sb.get("seq") and ab.get("prev_digest") == head["digest"] and ab.get("new_digest") == m["digest"] and ab.get("widened") == w and sig_ok(ppub, (ap or {}).get("signature"), D_SUCCESSION_APPROVAL, ab)) self.check("contract.widening_approved_by_principal", ok, seq, "; ".join(w)) else: self.check("contract.widening_approved_by_principal", True, seq) self.mission, self.mission_digest = dict(m["body"]), m["digest"] self.contract.append({"seq": sb.get("seq"), "digest": m["digest"], "widened": w}) if self.check("contract.plan_signed_and_bound", self.signed_record(p, D_PLAN, self.t.get("plan_authority_pubs") or []) and p["body"].get("mission_digest") == self.mission_digest, seq): self.set_plan(p) # delegations that no longer fit the new contract are gone; admissions not yet dispatched are void for e_id in [x.split(":", 1)[0] for x in d.get("voided_delegations") or []]: self.edges.pop(e_id, None) for eid, e in list(self.edges.items()): if not all(within(x, self.mission.get("resource_scope") or []) for x in e["scope"]) or \ not set(e["capabilities"]) <= set(self.mission.get("allowed_action_classes") or []) or \ e["expires_at"] > float(self.mission.get("expires_at") or 0): self.check("contract.delegations_outside_new_version_dropped", False, seq, eid) self.edges.pop(eid) for aid in [x for x in self.admissions if x not in self.tx]: self.admissions.pop(aid) def proposal(self, a: Mapping[str, Any], d: Mapping[str, Any], at: float, seq: int) -> None: req = a.get("req") or {} r = req.get("request") or {} aid = str((r.get("body") or {}).get("action_id")) act = self.plan.get(aid) ap = a.get("approval_rec") if act is not None and isinstance(ap, Mapping): authority = str((ap.get("body") or {}).get("approving_authority")) pinned = (self.t.get("approver_pubs") or {}).get(authority) if pinned and self.signed_record(ap, D_APPROVAL, [pinned]) and \ ap["body"].get("action_digest") == act.get("canonical_action_digest") and \ float(ap["body"].get("expires_at") or 0) > at and \ authority in (self.mission.get("required_approvals") or []): self.approvals[aid] = True if d.get("decision") != "ALLOW": return if not self.check("admission.action_in_plan", act is not None, seq, aid): return agent = str(r.get("agent")) pub = (self.t.get("agent_pubs") or {}).get(agent) self.check("admission.request_signed_by_plan_agent", bool(pub) and agent == act.get("agent_identity") and r.get("kind") == "PROPOSE" and sig_ok(pub, req.get("signature"), D_AGENT_REQ, r), seq, aid) au = a.get("authorization_rec") ab = (au or {}).get("body") or {} if isinstance(au, Mapping) else {} self.check("admission.authorization_signed_for_exact_action", self.signed_record(au, D_AUTH, self.t.get("authorization_issuer_pubs") or []) and ab.get("action_digest") == act.get("canonical_action_digest"), seq, aid) self.check("admission.authorization_unexpired", float(ab.get("expires_at") or 0) > at, seq, aid) self.check("admission.authorization_never_used_before", bool(ab.get("authorization_id")) and ab.get("authorization_id") not in self.used_auth, seq, aid) self.used_auth.add(ab.get("authorization_id")) self.check("admission.mission_unexpired_and_not_cancelled", at < float(self.mission.get("expires_at") or 0) and self.cancelled_at is None, seq, aid) if act.get("requires_approval") or str(act.get("reversibility")).upper() == "IRREVERSIBLE": self.check("admission.human_approval_present", self.approvals.get(aid, False), seq, aid) self.check("admission.delegation_chain_live", self.chain_live(agent, at, str(act.get("action_class")), str(act.get("target_resource"))), seq, aid) self.admissions[aid] = {"authorization_id": ab.get("authorization_id"), "expires_at": float(ab.get("expires_at") or 0), "mission_digest": self.mission_digest, "contract_seq": self.contract[-1]["seq"] if self.contract else None, "agent": agent} def intent(self, e: Mapping[str, Any]) -> None: b, at, seq = e["body"], e["at"], e["seq"] aid = str(b.get("action_id")) act, adm, t = self.plan.get(aid), self.admissions.get(aid), self.tx.get(aid) if not self.check("dispatch.admitted_first", act is not None and adm is not None, seq, f"{aid}: dispatched without a live admission"): self.tx[aid] = {"state": "INTENT", "class": b.get("recovery_class"), "attempt": b.get("attempt")} return cls = self.recovery_class(act) self.check("dispatch.recovery_class", b.get("recovery_class") == cls, seq, f"{aid}: recorded {b.get('recovery_class')}, derived {cls}") self.check("dispatch.bound_to_admitting_contract_version", b.get("mission_digest") == self.mission_digest == adm["mission_digest"] and b.get("authorization_id") == adm["authorization_id"] and b.get("action_digest") == act.get("canonical_action_digest"), seq, aid) self.check("dispatch.before_expiry", at < adm["expires_at"] and at < float(self.mission.get("expires_at") or 0), seq, aid) self.check("dispatch.not_after_cancellation", self.cancelled_at is None, seq, aid) key = hashlib.sha256(canon({"mission_id": self.mission.get("mission_id"), "action_id": aid, "action_digest": act.get("canonical_action_digest"), "authorization_id": adm["authorization_id"]})).hexdigest() self.check("dispatch.idempotency_key", b.get("idempotency_key") == key, seq, aid) self.check("dispatch.authority_live_at_dispatch", self.chain_live(adm["agent"], at, str(act.get("action_class")), str(act.get("target_resource"))), seq, aid) if t is None or t["state"] == "NOT_APPLIED": self.check("dispatch.exactly_once", b.get("attempt") == 1, seq, aid) else: self.check("dispatch.exactly_once", t["state"] == "AMBIGUOUS" and cls in ("READ_ONLY", "IDEMPOTENT_WRITE") and b.get("attempt") == (t.get("attempt") or 0) + 1 and b.get("idempotency_key") == t.get("key"), seq, f"{aid}: dispatched again while its outcome was {t['state']} (class {cls})") self.tx[aid] = {"state": "INTENT", "class": cls, "attempt": b.get("attempt"), "key": b.get("idempotency_key")} def tx_entry(self, e: Mapping[str, Any]) -> None: kind, b, at, seq = e["kind"], e["body"], e["at"], e["seq"] aid = str(b.get("action_id")) t = self.tx.get(aid) if kind == "TX_INTENT": self.intent(e) elif kind in ("TX_RESULT", "TX_AMBIGUOUS"): if self.check("dispatch.result_follows_intent", t is not None and t["state"] == "INTENT", seq, aid): if kind == "TX_AMBIGUOUS": t["state"] = "AMBIGUOUS" elif b.get("status") == "NOT_APPLIED": self.check("dispatch.executor_no_effect_only_for_repeatable", t["class"] in ("READ_ONLY", "IDEMPOTENT_WRITE"), seq, aid) t["state"] = "NOT_APPLIED" self.admissions.pop(aid, None) else: t["state"] = "REPORTED" elif kind == "COMP_INTENT": act = self.plan.get(aid) or {} ok = t is not None and t["class"] in ("REVERSIBLE", "COMPENSATABLE") and \ t["state"] in ("CONFIRMED", "CONTRADICTED", "REPORTED", "AMBIGUOUS", "COMPENSATION_FAILED") self.check("compensation.only_for_compensable_dispatched_action", ok, seq, aid) au = b.get("authorization") ab = (au or {}).get("body") or {} if isinstance(au, Mapping) else {} want = digest(D_COMP_ACTION, {"original": act.get("canonical_action_digest"), "compensation": act.get("compensation_action_id")}) self.check("compensation.own_authorization", self.signed_record(au, D_AUTH, self.t.get("authorization_issuer_pubs") or []) and ab.get("action_digest") == want == b.get("compensation_digest") and float(ab.get("expires_at") or 0) > at and bool(act.get("compensation_action_id")) and ab.get("authorization_id") not in self.used_auth, seq, aid) self.used_auth.add(ab.get("authorization_id")) if t is not None: t["state"] = "COMP_INTENT" elif kind == "COMP_RESULT": if self.check("compensation.result_follows_intent", t is not None and t["state"] == "COMP_INTENT", seq, aid): t["state"] = "COMP_REPORTED" if b.get("status") == "REPORTED" else "COMP_AMBIGUOUS" # ---- run ------------------------------------------------------------------------------------------- def run(self, lines: List[bytes]) -> None: prev = GENESIS_PREV for i, line in enumerate(lines): try: e = json.loads(line) except ValueError: self.check("chain.entry_parses", False, i) return fields = {k: e.get(k) for k in ("seq", "prev", "kind", "at", "body")} ok = self.check("chain.continuity", e.get("seq") == i and e.get("prev") == prev and e.get("hash") == digest(D_JOURNAL, fields), i) ok = self.check("chain.signed_by_pinned_runtime_key", sig_ok(self.t.get("runtime_pub"), e.get("sig"), D_JOURNAL, fields), i) and ok if not ok: return # nothing after a broken link can be relied on prev = e["hash"] self.hashes.append(prev) kind = e["kind"] if kind not in KNOWN_KINDS: self.unsupported.append(f"entry kind {kind!r} at {i}") return if i == 0: if not self.check("genesis.first_entry", kind == "GENESIS", i): return self.genesis(e) if self.unsupported or not self.mission: return elif kind == "INPUT": if e["body"].get("op") not in KNOWN_OPS: self.unsupported.append(f"operation {e['body'].get('op')!r} at {i}") return self.check("decision.every_input_decided_in_order", self.pending_input is None, i) self.pending_input = e elif kind == "DECISION": if self.check("decision.every_input_decided_in_order", self.pending_input is not None and self.pending_input["seq"] == e["body"].get("input_seq"), i): self.decided(self.pending_input, e) self.pending_input = None elif kind in ("RECOVERY", "TX_REFUSED"): continue elif kind == "GENESIS": self.check("genesis.first_entry", False, i, "a second genesis") else: self.tx_entry(e) def verify(journal_path: str, trust_path: str, *, head_path: Optional[str] = None, no_head: bool = False, require_final: bool = False) -> Dict[str, Any]: def result(verdict: str, **kw: Any) -> Dict[str, Any]: return {"schema": "CAIN42/EVOLUTION16/MISSION-VERIFICATION/v1", "verdict": verdict, **kw} try: with open(trust_path, "rb") as f: trust = json.loads(f.read()) with open(journal_path, "rb") as f: raw = f.read() except (OSError, ValueError) as e: return result("UNAVAILABLE", reason=f"{type(e).__name__}: {e}") if not isinstance(trust, Mapping) or not raw.strip(): return result("UNAVAILABLE", reason="the trust file is not an object, or the journal is empty") if trust.get("schema") != TRUST_SCHEMA: return result("UNSUPPORTED", reason=f"trust schema {trust.get('schema')!r}") keep = raw.rfind(b"\n") + 1 torn = keep != len(raw) v = Verifier(trust) v.run(raw[:keep].splitlines()) not_performed: List[str] = [] waived: List[str] = [] incomplete: List[str] = [] if head_path: try: with open(head_path, "rb") as f: h = json.loads(f.read()) hb = h.get("body") or {} ok = sig_ok(trust.get("runtime_pub"), h.get("signature"), D_JOURNAL + "/HEAD", hb) and \ isinstance(hb.get("seq"), int) and hb["seq"] < len(v.hashes) and v.hashes[hb["seq"]] == hb.get("hash") v.check("chain.contains_witnessed_head", ok, hb.get("seq"), "the journal does not contain the entry the witness holds: it was truncated or replaced") except (OSError, ValueError, AttributeError) as e: return result("UNAVAILABLE", reason=f"head file: {type(e).__name__}: {e}") elif no_head: waived.append("chain.contains_witnessed_head") else: not_performed.append("chain.contains_witnessed_head") incomplete.append("no witness-held head was supplied, so a truncated journal could not be noticed") open_tx = {a: t["state"] for a, t in sorted(v.tx.items()) if t["state"] in UNRESOLVED} if open_tx: incomplete.append("unresolved dispatches: " + ", ".join(f"{a}={s}" for a, s in open_tx.items())) if v.pending_input is not None: incomplete.append(f"input {v.pending_input['seq']} has no recorded decision") if torn: incomplete.append("the journal ends in a partial line (an interrupted write)") if require_final and v.final is None: incomplete.append("the mission has not been finalized") if v.contradicted: incomplete.append("pinned observers contradict a confirmed effect: " + ", ".join(sorted(set(v.contradicted)))) if v.final == "COMPLETED_VERIFIED": # the mission was recorded complete; evidence that arrived afterwards must not have undone that v.check("completion.still_supported_by_the_evidence", not open_tx, None, "the mission was finalized complete, and the journal now holds an unresolved or contradicted dispatch: " + ", ".join(open_tx)) verdict = "UNSUPPORTED" if v.unsupported else "FAILED" if v.failures else "INCOMPLETE" if incomplete else "VERIFIED" return result( verdict, failures=v.failures, unsupported=v.unsupported, incomplete=incomplete, checks_performed=dict(sorted(v.performed.items())), checks_not_performed=not_performed, checks_waived=waived, entries=len(v.hashes), contract_versions=v.contract, mission_final=v.final, transactions={a: t["state"] for a, t in sorted(v.tx.items())}, does_not_establish=["that a pinned observer's signed observation is true", "that the authorization issuer's policy decision was right", "that the supplied public keys belong to who you think", "anything about actions that did not go through this runtime"]) def main(argv: Optional[Sequence[str]] = None) -> int: ap = argparse.ArgumentParser(description="Clean-room verifier for a CAIN Evolution 16 mission journal.") ap.add_argument("journal") ap.add_argument("--trust", required=True) g = ap.add_mutually_exclusive_group() g.add_argument("--head", help="the signed journal head an outside witness kept") g.add_argument("--no-head", action="store_true", help="accept that truncation cannot be detected") ap.add_argument("--require-final", action="store_true") a = ap.parse_args(argv) out = verify(a.journal, a.trust, head_path=a.head, no_head=a.no_head, require_final=a.require_final) print(json.dumps(out, indent=1, sort_keys=True)) return EXIT[out["verdict"]] if __name__ == "__main__": sys.exit(main())