#!/usr/bin/env python3 """cain-launchpad-verify — CLEAN-ROOM verifier for CAIN Evolution 14 Launchpad evidence. ZERO CAIN IMPORTS. Needs Python 3 and the `cryptography` package only. python verify.py deployment --trust [--strict] python verify.py release --policy [--expect-commit ] python verify.py soak [--require-hours 72] (exit 3 = sound so far, but not yet long enough) It recomputes every verdict from the exported records and the public keys YOU supply. It never reads the producer's summary to decide anything; if that summary contradicts what the records prove, that is reported as INCONSISTENT. What it checks, per executed step * the action digest and idempotency key are recomputed from the recorded action * the state history only uses legal transitions * AUTHORIZED carries an ALLOW decision signed by the pinned policy key and bound to this action * EFFECT_VERIFIED carries an observation signed by the pinned gate key, bound to this action, and the gate's signed ledger contains exactly that effect under the SAME decision id that authorized it * the evidence-log leaf for the step is included in the tree at the size and root of the witnessed checkpoint (Evolution 13 F13-01: an inclusion view that is not the witnessed view proves nothing) and globally * every effect in the gate's ledger has a verified, authorized step (an effect without one is INCONSISTENT) * the checkpoint is signed by the pinned evidence key * witness statements are signature-checked one by one against pinned witness keys, bound to this checkpoint, counted once per member, and compared with the threshold (Evolution 13 F13-04: an asserted count is not a quorum) * if an earlier checkpoint is supplied, the log only grew (RFC 6962 consistency proof; Evolution 13 F13-02) Statuses: VERIFIED, REJECTED, INCONSISTENT, UNAVAILABLE, INSUFFICIENT_EVIDENCE, OUTCOME_UNKNOWN, BLOCKED. Exit code: 0 all mandatory checks VERIFIED; 3 verified but steps remain OUTCOME_UNKNOWN (or --strict with any non-VERIFIED item); 1 anything REJECTED, INCONSISTENT, UNAVAILABLE or INSUFFICIENT_EVIDENCE; 2 usage error. Not independent of the producer in these respects (stated, not hidden): the export was produced by the deployment being verified; the public keys are only as trustworthy as the channel you got them from; this file was written by the same project as the code it checks; both use the `cryptography` library for Ed25519. """ from __future__ import annotations import argparse import base64 import hashlib import json import sys from pathlib import Path from typing import Any, Dict, List, Optional from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey D_ACTION = "CAIN42/LAUNCHPAD/ACTION/v1" D_DECISION = "CAIN42/LAUNCHPAD/DECISION/v1" D_GATE = "CAIN42/LAUNCHPAD/GATE-RECORD/v1" D_CHECKPOINT = "CAIN42/SOVEREIGN-WITNESS/CHECKPOINT/v1" D_QUORUM = "CAIN42/FEDERATION/QUORUM-STATEMENT/v1" ACTION_FIELDS = ("tenant", "chain_id", "step_id", "actor", "role", "action", "params") LEGAL = {None: {"PROPOSED"}, "PROPOSED": {"AUTHORIZED", "EFFECT_REJECTED", "NO_EFFECT"}, "AUTHORIZED": {"DISPATCHED", "NO_EFFECT"}, "DISPATCHED": {"EFFECT_VERIFIED", "NO_EFFECT", "OUTCOME_UNKNOWN"}, "OUTCOME_UNKNOWN": {"EFFECT_VERIFIED", "NO_EFFECT"}} BAD = ("REJECTED", "INCONSISTENT", "UNAVAILABLE", "INSUFFICIENT_EVIDENCE") def cj(o: Any) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def sha(o: Any) -> str: return hashlib.sha256(cj(o)).hexdigest() def dd(domain: str, body: Dict[str, Any]) -> str: return sha({"domain": domain, **body}) def sig_ok(rec: Any, domain: str, pub: Optional[str]) -> Optional[Dict[str, Any]]: """Body if `rec` is signed under `domain` by exactly `pub`; the record's own signer_pub is never trusted.""" if not isinstance(rec, dict) or not isinstance(rec.get("body"), dict) or not pub: return None if rec.get("signer_pub") != pub or not isinstance(rec.get("signature"), str): return None try: Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify( base64.b64decode(rec["signature"]), dd(domain, rec["body"]).encode()) except (InvalidSignature, ValueError): return None return rec["body"] def _lh(b: bytes) -> bytes: return hashlib.sha256(b"\x00" + b).digest() def _nh(a: bytes, b: bytes) -> bytes: return hashlib.sha256(b"\x01" + a + b).digest() def inclusion_ok(leaf: bytes, index: int, size: int, proof: List[bytes], root: bytes) -> bool: if not (0 <= index < size): return False fn, sn, r = index, size - 1, _lh(leaf) for p in proof: if sn == 0: return False if fn & 1 or fn == sn: r = _nh(p, r) while fn and not fn & 1: fn >>= 1 sn >>= 1 else: r = _nh(r, p) fn >>= 1 sn >>= 1 return sn == 0 and r == root def consistency_ok(old: int, new: int, old_root: bytes, new_root: bytes, proof: List[bytes]) -> bool: if old < 0 or old > new: return False if old == 0: return True if old == new: return old_root == new_root and not proof p = list(proof) if not p: return False if old & (old - 1) == 0: p = [old_root] + p fn, sn = old - 1, new - 1 while fn & 1: fn >>= 1 sn >>= 1 fr = sr = p[0] for c in p[1:]: if sn == 0: return False if fn & 1 or fn == sn: fr, sr = _nh(c, fr), _nh(c, sr) while fn and not fn & 1: fn >>= 1 sn >>= 1 else: sr = _nh(sr, c) fn >>= 1 sn >>= 1 return fr == old_root and sr == new_root and sn == 0 def verify_deployment(export: Dict[str, Any], trust: Dict[str, Any]) -> Dict[str, Any]: svc = trust.get("services") or {} def one(role: str) -> Optional[str]: pubs = [s.get("pub") for s in svc.values() if s.get("role") == role] return pubs[0] if len(pubs) == 1 else None pol, gate, evid = one("policy"), one("gate"), one("evidence") witnesses = {n: s.get("pub") for n, s in svc.items() if s.get("role") == "witness"} threshold = trust.get("witness_threshold") checks: List[Dict[str, Any]] = [] def add(name: str, status: str, detail: str = "") -> str: checks.append({"check": name, "status": status, "detail": detail}) return status if not (pol and gate and evid) or not isinstance(threshold, int) or threshold < 1 or threshold > len(witnesses): add("trust_material", "INSUFFICIENT_EVIDENCE", "need exactly one policy, gate and evidence key and a sane threshold") return {"overall": "INSUFFICIENT_EVIDENCE", "checks": checks, "steps": []} add("trust_material", "VERIFIED", f"{len(witnesses)} witness keys, threshold {threshold}") # ---- gate ledger (one signed record, or signed pages that must join up without a gap) --------------- pages = export.get("ledger_pages") if pages is None: pages = [export["ledger"]] if export.get("ledger") is not None else [] start = int((export.get("window") or {}).get("after_effect_seq", 0)) ledger: Dict[str, Dict[str, Any]] = {} ledger_ok, why, pos, n_eff = bool(pages), "ledger missing", start, 0 for i, pg in enumerate(pages): lb = sig_ok(pg, D_GATE, gate) if lb is None or lb.get("kind") != "LEDGER" or not isinstance(lb.get("effects"), list): ledger_ok, why = False, "a ledger page is not signed by the pinned gate key" break if "after" in lb and int(lb["after"]) != pos: ledger_ok, why = False, f"ledger page {i} starts at {lb['after']}, expected {pos}: effects are missing" break for e in lb["effects"]: if "effect_seq" in e and "after" in lb and int(e["effect_seq"]) != pos + 1: ledger_ok, why = False, f"effect sequence jumps from {pos} to {e['effect_seq']}" break pos = int(e.get("effect_seq", pos + 1)) n_eff += 1 ledger[e["idempotency_key"]] = e if not ledger_ok: break if i == len(pages) - 1 and lb.get("complete") is False: ledger_ok, why = False, "the last ledger page says more effects exist than were supplied" if ledger_ok and len(ledger) != n_eff: ledger_ok, why = False, "duplicate idempotency keys in ledger" if ledger_ok: add("gate_ledger_signature", "VERIFIED", f"{n_eff} effects, sequence {start + 1 if n_eff else start}..{pos}") elif not pages: add("gate_ledger_signature", "UNAVAILABLE", why) else: add("gate_ledger_signature", "INCONSISTENT" if "missing" in why or "jumps" in why or "duplicate" in why or "more effects" in why else "REJECTED", why) # ---- checkpoint + witness quorum ------------------------------------------------------------------- cp = export.get("checkpoint") cb = sig_ok(cp, D_CHECKPOINT, evid) cp_ok = cb is not None and cp.get("digest") == dd(D_CHECKPOINT, cb) add("checkpoint_signature", "VERIFIED" if cp_ok else ("UNAVAILABLE" if cp is None else "REJECTED")) accepted, problems = set(), [] for st in export.get("statements") or []: b = st.get("body") if isinstance(st, dict) else None mid = (b or {}).get("member_id") if sig_ok(st, D_QUORUM, witnesses.get(mid)) is None: problems.append(f"{mid}:bad-signature-or-not-a-member") elif not cp_ok or b.get("checkpoint_digest") != cp["digest"]: problems.append(f"{mid}:other-checkpoint") elif str(b.get("decision")).upper() != "ACCEPT": problems.append(f"{mid}:{b.get('decision')}:{b.get('reason')}") else: accepted.add(mid) if not cp_ok: add("witness_quorum", "INSUFFICIENT_EVIDENCE", "no verified checkpoint to witness") elif any(x.split(":")[-1] in ("EQUIVOCATION", "ROLLBACK", "INCONSISTENT") for x in problems): add("witness_quorum", "REJECTED", "a witness reported the log inconsistent: " + ",".join(problems)) elif len(accepted) >= threshold: add("witness_quorum", "VERIFIED", f"{len(accepted)} verified signatures >= {threshold}") else: add("witness_quorum", "INSUFFICIENT_EVIDENCE", f"{len(accepted)} verified signatures < {threshold}; " + ",".join(problems)) # ---- log growth ------------------------------------------------------------------------------------ if export.get("previous_checkpoint") is not None: pb = sig_ok(export["previous_checkpoint"], D_CHECKPOINT, evid) con = export.get("consistency") or {} try: ok = bool(cp_ok and pb is not None and consistency_ok( pb["tree_size"], cb["tree_size"], bytes.fromhex(pb["root_hash"]), bytes.fromhex(cb["root_hash"]), [bytes.fromhex(x) for x in con.get("proof", [])])) except (ValueError, TypeError, KeyError): ok = False add("log_append_only", "VERIFIED" if ok else "REJECTED", "" if ok else "no valid consistency proof from the earlier checkpoint to this one") # ---- steps ----------------------------------------------------------------------------------------- leaves = export.get("leaves") or {} steps_out, seen = [], set() for step in export.get("steps") or []: idem, hist = step.get("idempotency_key"), step.get("history") or [] seen.add(idem) res = {"idempotency_key": idem, "final_state": None, "status": "VERIFIED", "problems": []} def bad(status: str, why: str) -> None: if res["status"] == "VERIFIED" or status == "INCONSISTENT": res["status"] = status res["problems"].append(why) if not hist: bad("UNAVAILABLE", "no state history") steps_out.append(res) continue first = hist[0]["record"] action = first.get("action") or {} adig = dd(D_ACTION, {f: action.get(f) for f in ACTION_FIELDS}) if first.get("action_digest") != adig or \ sha({"t": action.get("tenant"), "c": action.get("chain_id"), "s": action.get("step_id")}) != idem: bad("REJECTED", "action does not hash to the recorded digest / idempotency key") prev, decision_id, obs = None, None, None for h in hist: if h["state"] not in LEGAL.get(prev, set()): bad("REJECTED", f"illegal transition {prev}->{h['state']}") prev = h["state"] rec = h["record"] if h["state"] in ("AUTHORIZED", "EFFECT_REJECTED"): d = sig_ok(rec.get("decision"), D_DECISION, pol) want = "ALLOW" if h["state"] == "AUTHORIZED" else "DENY" if d is None: bad("REJECTED", "decision not signed by the pinned policy key") elif d.get("decision") != want or d.get("action_digest") != adig or d.get("idempotency_key") != idem: bad("REJECTED", "decision does not cover this action") elif want == "ALLOW": decision_id = d.get("decision_id") if h["state"] in ("EFFECT_VERIFIED", "NO_EFFECT") and isinstance(rec.get("observation"), dict): o = sig_ok(rec["observation"], D_GATE, gate) if o is None or o.get("idempotency_key") != idem: bad("REJECTED", "observation not signed by the pinned gate key for this step") else: obs = (rec["observation"], o) res["final_state"] = prev in_ledger = ledger.get(idem) if prev == "EFFECT_VERIFIED": if obs is None or obs[1].get("present") is not True or obs[1].get("action_digest") != adig: bad("REJECTED", "no gate observation showing this effect") if decision_id is None: bad("REJECTED", "effect recorded without an ALLOW decision in its history") if ledger_ok and (in_ledger is None or in_ledger.get("action_digest") != adig or in_ledger.get("decision_id") != decision_id): bad("INCONSISTENT", "gate ledger does not show this effect under the decision that authorized it") leaf = leaves.get(sha({"state": prev, "observation": dd(D_GATE, obs[1])})) if obs else None try: inc = bool(leaf and cp_ok and leaf["tree_size"] == cb["tree_size"] and leaf["root_hash"] == cb["root_hash"] and inclusion_ok(cj(leaf["record"]), leaf["index"], leaf["tree_size"], [bytes.fromhex(x) for x in leaf["proof"]], bytes.fromhex(cb["root_hash"]))) except (ValueError, TypeError, KeyError): inc = False if not inc: bad("INSUFFICIENT_EVIDENCE", "no inclusion proof binding this step to the witnessed checkpoint") elif prev in ("NO_EFFECT", "EFFECT_REJECTED"): if ledger_ok and in_ledger is not None: bad("INCONSISTENT", f"state says {prev} but the gate ledger contains the effect") else: # PROPOSED / AUTHORIZED / DISPATCHED / OUTCOME_UNKNOWN: not resolved. Never upgraded by this verifier. if res["status"] == "VERIFIED": res["status"] = "OUTCOME_UNKNOWN" res["problems"].append(f"step is open in state {prev}" + ("; the gate ledger DOES contain the effect" if in_ledger else "")) steps_out.append(res) orphans = sorted(set(ledger) - seen) if ledger_ok: add("every_effect_has_an_authorized_step", "INCONSISTENT" if orphans else "VERIFIED", f"effects with no step record: {orphans[:5]}" if orphans else f"{len(ledger)} effects accounted for") counts: Dict[str, int] = {} for s in steps_out: counts[s["status"]] = counts.get(s["status"], 0) + 1 statuses = [c["status"] for c in checks] + [s["status"] for s in steps_out] if "INCONSISTENT" in statuses: overall = "INCONSISTENT" elif any(s in BAD for s in statuses): overall = next(s for s in BAD if s in statuses) elif "OUTCOME_UNKNOWN" in statuses: overall = "OUTCOME_UNKNOWN" else: overall = "VERIFIED" # the producer's own claims are compared, never used ps = (export.get("producer_summary") or {}) claimed = (ps.get("by_state") or {}).get("EFFECT_VERIFIED") proven = sum(1 for s in steps_out if s["final_state"] == "EFFECT_VERIFIED" and s["status"] == "VERIFIED") if claimed is not None and claimed != proven: add("producer_summary_matches_evidence", "INCONSISTENT", f"producer claims {claimed} verified effects, records prove {proven}") overall = "INCONSISTENT" if ps.get("quorum_met") is True and next(c["status"] for c in checks if c["check"] == "witness_quorum") != "VERIFIED": add("producer_quorum_claim_matches_evidence", "INCONSISTENT", "producer claims a quorum the signatures do not show") overall = "INCONSISTENT" return {"schema": "CAIN42/LAUNCHPAD/VERIFICATION/v1", "overall": overall, "checks": checks, "step_status_counts": counts, "steps": steps_out, "effects_proven": proven, "window": {"after_effect_seq": start, "last_effect_seq": pos, "after_step_seq": int((export.get("window") or {}).get("after_step_seq", 0)), "last_step_seq": (export.get("window") or {}).get("last_step_seq"), "tree_size": cb["tree_size"] if cp_ok else None, "root_hash": cb["root_hash"] if cp_ok else None}} # ===================================================================================================== # release verification: DSSE envelopes, in-toto/SLSA provenance, CycloneDX SBOM, release policy # ===================================================================================================== PT_INTOTO = "application/vnd.in-toto+json" PT_MANIFEST = "application/vnd.cain.launchpad.release+json" def _pae(payload_type: str, payload: bytes) -> bytes: t = payload_type.encode() return b"DSSEv1 %d %s %d %s" % (len(t), t, len(payload), payload) def _vt(v: Any) -> tuple: try: return tuple(int(x) for x in str(v).split(".")) except ValueError: return (-1,) def dsse_signers(env: Any, payload_type: str, policy: Dict[str, Any]) -> tuple: """Returns (payload bytes or None, {keyid: role} for every signature that verifies under a key the policy authorizes and has not revoked, [notes]). A signature by a key the policy does not list counts for nothing.""" notes: List[str] = [] if not isinstance(env, dict) or env.get("payloadType") != payload_type: return None, {}, ["not a DSSE envelope of the expected payload type"] try: payload = base64.b64decode(env["payload"], validate=True) except Exception: # noqa: BLE001 return None, {}, ["payload is not base64"] good: Dict[str, str] = {} for sg in env.get("signatures") or []: kid = sg.get("keyid") if isinstance(sg, dict) else None ent = (policy.get("signers") or {}).get(kid) if not ent: notes.append(f"{str(kid)[:12]}:not-in-policy") continue if ent.get("revoked"): notes.append(f"{kid[:12]}:revoked") continue try: raw = base64.b64decode(ent["public_key_b64"]) if hashlib.sha256(raw).hexdigest() != kid: notes.append(f"{kid[:12]}:policy-key-does-not-match-keyid") continue Ed25519PublicKey.from_public_bytes(raw).verify(base64.b64decode(sg["sig"]), _pae(payload_type, payload)) good[kid] = ent.get("role", "") except (InvalidSignature, ValueError, KeyError, TypeError): notes.append(f"{kid[:12]}:bad-signature") return payload, good, notes def verify_release(rel_dir: Path, policy: Dict[str, Any], now: float, expect_commit: Optional[str] = None, installed_version: Optional[str] = None, installed_mandatory: Optional[List[str]] = None) -> Dict[str, Any]: checks: List[Dict[str, Any]] = [] def add(name: str, status: str, detail: str = "") -> str: checks.append({"check": name, "status": status, "detail": detail}) return status def done(manifest: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: st = [c["status"] for c in checks] overall = "INCONSISTENT" if "INCONSISTENT" in st else next((s for s in BAD if s in st), "VERIFIED") return {"schema": "CAIN42/LAUNCHPAD/RELEASE-VERIFICATION/v1", "overall": overall, "checks": checks, "release": ({k: manifest.get(k) for k in ("name", "version", "development_release")} | {"artifact_sha256": (manifest.get("artifact") or {}).get("sha256"), "source_commit": (manifest.get("source") or {}).get("commit")}) if manifest else None} def read(name: str) -> Optional[bytes]: try: return (rel_dir / name).read_bytes() except OSError: return None if not isinstance(policy.get("signers"), dict) or not policy.get("required_roles"): add("release_policy", "INSUFFICIENT_EVIDENCE", "policy lists no signers or no required roles") return done() raw = read("release-manifest.dsse.json") if raw is None: add("manifest_present", "UNAVAILABLE", "release-manifest.dsse.json missing") return done() try: env = json.loads(raw) except ValueError: add("manifest_present", "REJECTED", "manifest envelope is not JSON") return done() payload, good, notes = dsse_signers(env, PT_MANIFEST, policy) roles = set(good.values()) need = set(policy["required_roles"]) # one key can only fill one role, and every required role needs its own authorized signature if payload is None or not need <= roles or len(good) < len(need): add("manifest_signatures", "REJECTED", f"authorized roles signed: {sorted(roles)}; required: {sorted(need)}; " + ",".join(notes)) return done() add("manifest_signatures", "VERIFIED", f"signed by authorized {sorted(roles)}" + (f"; ignored: {','.join(notes)}" if notes else "")) try: m = json.loads(payload) art, src = m["artifact"], m["source"] created, expires = int(m["created"]), int(m["expires"]) except (ValueError, KeyError, TypeError): add("manifest_wellformed", "REJECTED", "signed manifest is malformed") return done() max_age = float(policy.get("max_age_days", 0)) * 86400 if not (created <= now + 300): add("manifest_fresh", "REJECTED", "manifest is dated in the future") elif now > expires or (max_age and now - created > max_age): add("manifest_fresh", "REJECTED", "manifest is expired or older than the policy allows (stale release)") else: add("manifest_fresh", "VERIFIED", f"age {int((now - created) // 3600)} h") blob = read(str(art.get("name"))) if blob is None: add("artifact_hash", "UNAVAILABLE", "artifact file missing") elif hashlib.sha256(blob).hexdigest() != art.get("sha256") or len(blob) != art.get("size"): add("artifact_hash", "REJECTED", "artifact bytes do not match the signed manifest") else: add("artifact_hash", "VERIFIED", art["sha256"]) for key, label in (("sbom", "sbom_hash"), ("provenance", "provenance_hash")): b = read(str((m.get(key) or {}).get("name"))) if b is None: add(label, "UNAVAILABLE", "file missing") else: add(label, "VERIFIED" if hashlib.sha256(b).hexdigest() == m[key].get("sha256") else "REJECTED") # provenance: signed by an authorized BUILDER, about THIS artifact, from THIS commit, by an allowed builder id praw = read(str((m.get("provenance") or {}).get("name"))) try: ppay, pgood, pnotes = dsse_signers(json.loads(praw) if praw else None, PT_INTOTO, policy) except ValueError: ppay, pgood, pnotes = None, {}, ["provenance envelope is not JSON"] if ppay is None or "builder" not in pgood.values(): add("provenance_signature", "UNAVAILABLE" if praw is None else "REJECTED", "no valid signature by an authorized builder key; " + ",".join(pnotes)) else: add("provenance_signature", "VERIFIED") try: st = json.loads(ppay) pred = st["predicate"] subj = {s["name"]: s["digest"]["sha256"] for s in st["subject"]} commits = [d["digest"]["gitCommit"] for d in pred["buildDefinition"]["resolvedDependencies"] if "gitCommit" in (d.get("digest") or {})] bid = pred["runDetails"]["builder"]["id"] ok_type = st.get("_type") == "https://in-toto.io/Statement/v1" and \ st.get("predicateType") == "https://slsa.dev/provenance/v1" except (ValueError, KeyError, TypeError): add("provenance_wellformed", "REJECTED", "not an in-toto v1 statement with a SLSA provenance v1 predicate") else: add("provenance_wellformed", "VERIFIED" if ok_type else "REJECTED") add("provenance_subject_is_this_artifact", "VERIFIED" if subj.get(art.get("name")) == art.get("sha256") else "INCONSISTENT", "" if subj.get(art.get("name")) == art.get("sha256") else "provenance describes a different artifact") add("provenance_source_matches_manifest", "VERIFIED" if commits == [src.get("commit")] else "INCONSISTENT", ",".join(commits)) add("provenance_builder_allowed", "VERIFIED" if bid in (policy.get("allowed_builder_ids") or []) and bid == m.get("builder_id") else "REJECTED", bid) if expect_commit: add("source_commit_expected", "VERIFIED" if src.get("commit") == expect_commit else "INCONSISTENT", f"manifest {src.get('commit')} expected {expect_commit}") if policy.get("require_clean_source"): clean = src.get("clean") is True and not m.get("development_release") add("source_matches_commit", "VERIFIED" if clean else "REJECTED", "" if clean else "development release: inputs differ from the named commit, so it is not traceable to it") else: add("source_matches_commit", "VERIFIED" if src.get("clean") is True else "INSUFFICIENT_EVIDENCE", "" if src.get("clean") is True else "policy tolerates a development release; the artifact is NOT " "traceable to the named commit") add("version_minimum", "VERIFIED" if _vt(m.get("version")) >= _vt(policy.get("min_version", "0")) else "REJECTED", f"{m.get('version')} vs minimum {policy.get('min_version')}") if installed_version is not None: add("no_downgrade", "VERIFIED" if _vt(m.get("version")) > _vt(installed_version) else "REJECTED", f"{m.get('version')} vs installed {installed_version}") need_mv = set(policy.get("required_mandatory_verification") or []) | set(installed_mandatory or []) missing = sorted(need_mv - set(m.get("mandatory_verification") or [])) add("mandatory_verification_not_weakened", "REJECTED" if missing else "VERIFIED", f"release drops: {missing}" if missing else "") try: bom = json.loads(read("sbom.cdx.json") or b"") files = [c for c in bom["components"] if c.get("type") == "file"] ok = bom.get("bomFormat") == "CycloneDX" and bom["metadata"]["component"]["hashes"][0]["content"] == art.get("sha256") add("sbom_describes_this_artifact", "VERIFIED" if ok else "INCONSISTENT", f"{len(files)} files, " f"{sum(1 for c in bom['components'] if c.get('type') == 'library')} third-party packages") except (ValueError, KeyError, TypeError, IndexError): add("sbom_describes_this_artifact", "REJECTED", "SBOM missing or malformed") return done(m) # ===================================================================================================== # soak evidence: a hash-chained record of a long run # ===================================================================================================== def verify_soak(path: Path, require_hours: float) -> Dict[str, Any]: checks: List[Dict[str, Any]] = [] def add(name: str, status: str, detail: str = "") -> str: checks.append({"check": name, "status": status, "detail": detail}) return status def done(extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: st = [c["status"] for c in checks] overall = "INCONSISTENT" if "INCONSISTENT" in st else next((x for x in BAD if x in st), "VERIFIED") return {"schema": "CAIN42/LAUNCHPAD/SOAK-VERIFICATION/v1", "overall": overall, "checks": checks, "required_hours": require_hours, **(extra or {}), "not_independent": "timestamps come from the clock of the host that ran the soak; nothing outside " "that host witnessed the elapsed time"} try: lines = [l for l in path.read_text().splitlines() if l.strip()] except OSError as e: add("records_present", "UNAVAILABLE", str(e)) return done() recs: List[Dict[str, Any]] = [] prev = "GENESIS" for i, line in enumerate(lines): try: r = json.loads(line) h = r.pop("hash") ok = r["prev"] == prev and r["seq"] == i and \ hashlib.sha256(json.dumps(r, sort_keys=True, separators=(",", ":")).encode()).hexdigest() == h except (ValueError, KeyError, TypeError): ok, h = False, None if not ok: add("record_chain", "REJECTED", f"record {i} is malformed, out of order, or does not hash to its link") return done({"records": i}) prev = h recs.append(r) if not recs or recs[0]["kind"] != "start": add("record_chain", "INSUFFICIENT_EVIDENCE", "no start record") return done({"records": len(recs)}) add("record_chain", "VERIFIED", f"{len(recs)} records, each linked to the one before") iv = [r for r in recs if r["kind"] == "interval"] final = recs[-1] if recs[-1]["kind"] == "final" else None elapsed_h = (recs[-1]["mono"] - recs[0]["mono"]) / 3600 wall_h = (recs[-1]["wall"] - recs[0]["wall"]) / 3600 # the two clocks must tell the same story, record by record, and nothing may be missing in between drift = max((abs((b["wall"] - a["wall"]) - (b["mono"] - a["mono"])) for a, b in zip(recs, recs[1:])), default=0) back = any(b["mono"] < a["mono"] or b["wall"] < a["wall"] - 2 for a, b in zip(recs, recs[1:])) add("clocks_consistent", "INCONSISTENT" if back or drift > 30 else "VERIFIED", f"max wall-vs-monotonic disagreement between records {drift:.1f} s") gaps = [b["mono"] - a["mono"] for a, b in zip(iv, iv[1:])] typical = sorted(gaps)[len(gaps) // 2] if gaps else 0 biggest = max(gaps, default=0) add("no_gap_in_coverage", "INCONSISTENT" if gaps and biggest > max(5 * typical, typical + 600) else "VERIFIED", f"{len(iv)} interval records; typical spacing {typical:.0f} s, largest {biggest:.0f} s") tot = (final or (iv[-1] if iv else {})).get("totals") or {} viol = {k: tot.get(k) for k in ("authorization_violations", "duplicate_executions", "window_verify_failures")} if not tot: add("safety_counters", "INSUFFICIENT_EVIDENCE", "no totals recorded yet") else: add("safety_counters", "REJECTED" if any(viol.values()) else "VERIFIED", json.dumps(viol, sort_keys=True)) # the verdicts are recomputed from each interval, not read from the totals bad_w = [r["seq"] for r in iv if (r.get("window") or {}).get("verdict") in ("REJECTED", "INCONSISTENT")] add("no_window_rejected", "REJECTED" if bad_w else "VERIFIED", f"records {bad_w[:5]}" if bad_w else f"{sum(1 for r in iv if (r.get('window') or {}).get('verdict') == 'VERIFIED')} of {len(iv)} windows VERIFIED") if bool(bad_w) != bool(tot.get("window_verify_failures")) and tot: add("totals_match_records", "INCONSISTENT", "the totals disagree with the per-interval verdicts") # verified windows must join up: each one starts where the previous verified one ended pos, joined = 0, True for r in iv: w = r.get("window") or {} if w.get("verdict") == "VERIFIED": joined = joined and w["range"]["after_effect_seq"] == pos pos = w["range"]["last_effect_seq"] add("verified_windows_cover_the_ledger_without_a_gap", "VERIFIED" if joined else "INCONSISTENT", f"effects 1..{pos} covered") last_ag = next((r["agreement"] for r in reversed(iv) if "agree" in (r.get("agreement") or {})), None) if last_ag is None: add("final_agreement", "INSUFFICIENT_EVIDENCE", "no agreement measurement recorded") elif final and not (last_ag["agree"] and last_ag["open_steps"] == 0): add("final_agreement", "INCONSISTENT", json.dumps(last_ag, sort_keys=True)[:200]) else: add("final_agreement", "VERIFIED" if last_ag["agree"] else "INSUFFICIENT_EVIDENCE", f"ledger {last_ag['gate_ledger_effects']} = state {last_ag['state_effect_verified']} = " f"client {last_ag['runner_saw_executed']}" if last_ag["agree"] else "counts not yet in agreement (run in progress)") if final and pos != last_ag["gate_ledger_effects"]: add("every_effect_was_in_a_verified_window", "INSUFFICIENT_EVIDENCE", f"verified windows cover {pos} of {last_ag['gate_ledger_effects']} effects") if final is None: add("run_finished", "INSUFFICIENT_EVIDENCE", f"no final record: the run is still going or was killed; " f"{elapsed_h:.2f} h recorded so far") elif final.get("status") != "COMPLETED": add("run_finished", "REJECTED", f"final status {final.get('status')}: {final.get('stop_reason')}") else: add("run_finished", "VERIFIED") if tot.get("lost_deliveries"): add("no_lost_delivery", "REJECTED", f"{tot['lost_deliveries']} requests never reached a known outcome") add("elapsed_time", "VERIFIED" if elapsed_h >= require_hours else "INSUFFICIENT_EVIDENCE", f"{elapsed_h:.2f} h elapsed on the monotonic clock ({wall_h:.2f} h wall clock); {require_hours:g} h required") return done({"records": len(recs), "elapsed_hours": round(elapsed_h, 4), "totals": tot, "started_wall": recs[0]["wall"], "last_wall": recs[-1]["wall"], "build": recs[0].get("build")}) def main() -> int: ap = argparse.ArgumentParser() sub = ap.add_subparsers(dest="cmd", required=True) d = sub.add_parser("deployment") d.add_argument("export") d.add_argument("--trust", required=True) d.add_argument("--strict", action="store_true") r = sub.add_parser("release") r.add_argument("dir") r.add_argument("--policy", required=True) r.add_argument("--now", type=float) r.add_argument("--expect-commit") r.add_argument("--installed-version") r.add_argument("--installed-mandatory", help="comma-separated checks the installed release declares") k = sub.add_parser("soak") k.add_argument("records") k.add_argument("--require-hours", type=float, default=72.0) a = ap.parse_args() if a.cmd == "soak": out = verify_soak(Path(a.records), a.require_hours) print(json.dumps(out, indent=1, sort_keys=True)) return 0 if out["overall"] == "VERIFIED" else (3 if out["overall"] == "INSUFFICIENT_EVIDENCE" else 1) if a.cmd == "release": import time as _t try: pol = json.loads(Path(a.policy).read_text()) except (OSError, ValueError) as e: print(json.dumps({"overall": "UNAVAILABLE", "error": f"release policy: {e}"})) return 1 out = verify_release(Path(a.dir), pol, a.now if a.now is not None else _t.time(), a.expect_commit, a.installed_version, a.installed_mandatory.split(",") if a.installed_mandatory else None) print(json.dumps(out, indent=1, sort_keys=True)) return 0 if out["overall"] == "VERIFIED" else 1 try: export, trust = json.loads(Path(a.export).read_text()), json.loads(Path(a.trust).read_text()) except (OSError, ValueError) as e: print(json.dumps({"overall": "UNAVAILABLE", "error": str(e)})) return 1 out = verify_deployment(export, trust) print(json.dumps(out, indent=1, sort_keys=True)) if out["overall"] == "VERIFIED": return 0 if out["overall"] == "OUTCOME_UNKNOWN": return 3 return 1 if __name__ == "__main__": sys.exit(main())