#!/usr/bin/env python3 """Independently verify a Sentinel evidence bundle. Standalone by design. This script imports nothing from Sentinel and has no third-party dependencies -- Python 3.8+ and the standard library only. Copy it anywhere, run it against a bundle, and check the result without installing our product or trusting our infrastructure. That independence is the point. A verifier that shared code with the producer would only prove the producer agrees with itself; a bug in canonicalisation would be invisible to it. Everything below is written from the published spec (docs/reference/evidence-bundle-spec.md), so if the spec is wrong or incomplete, this script disagrees with Sentinel and the disagreement is the finding. python3 verify_evidence.py bundle.json python3 verify_evidence.py bundle.json --json python3 verify_evidence.py bundle.json --quiet # exit code only Exit codes: 0 every check passed 1 a check failed (tampering, or a spec mismatch) 2 the bundle could not be read or is missing required fields What this proves, and what it does not, is printed with the result. Read that section: a bundle is a segment of an append-only log, and there are claims it cannot make about itself. """ from __future__ import annotations import argparse import hashlib import json import sys GENESIS_HASH = "0" * 64 # -------------------------------------------------------------------------- # Primitives -- the whole hashing spec is these three functions # -------------------------------------------------------------------------- def canonical_json(payload): """Serialise a payload to the exact string Sentinel hashes. Sorted keys, no whitespace, non-JSON types coerced via str(). All three details are load-bearing: change any one and every hash below differs. """ return json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) def sha256_hex(value): return hashlib.sha256(value.encode("utf-8")).hexdigest() def compute_chain_hash(previous_hash, payload_hash, event_id, created_at): """chain_hash = SHA256(previous_hash | payload_hash | event_id | created_at)""" return sha256_hex("|".join([previous_hash, payload_hash, event_id, created_at])) # -------------------------------------------------------------------------- # Result accumulation # -------------------------------------------------------------------------- class Report: def __init__(self): self.checks = [] def add(self, name, ok, detail="", severity="error"): self.checks.append( {"check": name, "ok": bool(ok), "detail": detail, "severity": severity} ) return ok def note(self, name, detail): self.checks.append( {"check": name, "ok": True, "detail": detail, "severity": "info"} ) @property def failures(self): return [c for c in self.checks if not c["ok"] and c["severity"] == "error"] @property def ok(self): return not self.failures # -------------------------------------------------------------------------- # Chain verification # -------------------------------------------------------------------------- def verify_events(events, report): """Verify each event's own hashes, then the links between them.""" if not events: report.add("events present", False, "bundle contains no events") return previous = None gaps = 0 for idx, event in enumerate(events): eid = str(event.get("event_id") or event.get("audit_id") or idx) payload = event.get("payload") if payload is None and "payload_json" in event: raw = event["payload_json"] payload = json.loads(raw) if isinstance(raw, str) else raw if payload is None: payload = {} expected_payload = sha256_hex(canonical_json(payload)) actual_payload = str(event.get("payload_hash", "")) report.add( "event %s payload_hash" % eid, expected_payload == actual_payload, "expected %s, found %s" % (expected_payload, actual_payload or "(absent)"), ) declared_previous = str(event.get("previous_hash", "")) expected_chain = compute_chain_hash( declared_previous, actual_payload or expected_payload, eid, str(event.get("created_at") or event.get("timestamp") or ""), ) actual_chain = str(event.get("chain_hash", "")) report.add( "event %s chain_hash" % eid, expected_chain == actual_chain, "expected %s, found %s" % (expected_chain, actual_chain or "(absent)"), ) if previous is not None: if declared_previous != previous: # A bundle is a filtered slice, so it may skip events. That is # expected and is NOT tampering -- each event above still had # its own hashes verified independently. gaps += 1 previous = actual_chain if gaps: report.note( "segment contiguity", "%d gap(s): the bundle skips intervening events, which is normal " "for a per-decision export and is not evidence of tampering" % gaps, ) else: report.note("segment contiguity", "events are consecutive in the chain") anchor = str(events[0].get("previous_hash", "")) if anchor == GENESIS_HASH: report.note("segment anchor", "this segment starts at the genesis of the log") else: report.note( "segment anchor", "anchored to prior chain_hash %s -- check this against the full log " "to place the segment" % anchor, ) # -------------------------------------------------------------------------- # Proof of Agent # -------------------------------------------------------------------------- #: Field order of the canonical evaluation string, per scheme version. EVALUATION_FIELDS = { "sentinel-poa-v1": [ "decision_id", "agent_id", "verdict", "rules_evaluated", "rules_triggered", "passed_rules", "failed_rules", ], "sentinel-poa-v2": [ "decision_id", "agent_id", "applied_verdict", "computed_verdict", "mode", "rules_evaluated", "rules_triggered", "passed_rules", "failed_rules", ], } REGISTRY_FIELDS = ["agent_id", "agent_type", "template_version", "rule_ids"] def verify_proof_of_agent(decision, report): """Verify the PoA digests, then tie them back to the decision record. Recomputing the digests only proves the proof is internally consistent. The check that matters is the last one: that the canonical strings describe THIS decision. Without it, a valid proof from some other decision would pass. """ poa = (decision or {}).get("proof_of_agent") if not poa: report.note("proof of agent", "no proof_of_agent in this bundle; skipped") return scheme = str(poa.get("scheme", "")) fields = EVALUATION_FIELDS.get(scheme) if fields is None: report.add( "poa scheme", False, "unknown scheme %r; this verifier knows %s" % (scheme, ", ".join(sorted(EVALUATION_FIELDS))), ) return report.note("poa scheme", scheme) reg_canonical = poa.get("registry_canonical") eval_canonical = poa.get("evaluation_canonical") if reg_canonical is None or eval_canonical is None: report.add( "poa canonical strings", False, "bundle omits registry_canonical/evaluation_canonical, so the " "bindings cannot be independently recomputed", ) return reg_binding = sha256_hex(reg_canonical) eval_binding = sha256_hex(eval_canonical) report.add( "poa registry_binding", reg_binding == poa.get("registry_binding"), "expected %s, found %s" % (reg_binding, poa.get("registry_binding")), ) report.add( "poa evaluation_binding", eval_binding == poa.get("evaluation_binding"), "expected %s, found %s" % (eval_binding, poa.get("evaluation_binding")), ) combined = sha256_hex(reg_binding + eval_binding) report.add( "poa combined_digest", combined == poa.get("combined_digest"), "expected %s, found %s" % (combined, poa.get("combined_digest")), ) _cross_check_evaluation(decision, eval_canonical, fields, report) _cross_check_registry(decision, reg_canonical, report) def _cross_check_evaluation(decision, canonical, fields, report): """Assert the canonical evaluation string describes this decision.""" parts = canonical.split("|") if len(parts) != len(fields): report.add( "poa evaluation shape", False, "expected %d fields for this scheme, found %d" % (len(fields), len(parts)), ) return parsed = dict(zip(fields, parts)) details = decision.get("details") or {} expected = { "decision_id": str(decision.get("decision_id", "")), "agent_id": str(decision.get("agent_id", "")), "verdict": str(decision.get("verdict", "")), "applied_verdict": str( decision.get("applied_verdict", decision.get("verdict", "")) ), "computed_verdict": str( decision.get("computed_verdict", decision.get("verdict", "")) ), "mode": str(decision.get("mode", "")), "rules_evaluated": str(decision.get("rules_evaluated", "")), "rules_triggered": str(decision.get("rules_triggered", "")), "passed_rules": ",".join(sorted(details.get("passed_rules", []) or [])), "failed_rules": ",".join(sorted(details.get("failed_rules", []) or [])), } for field in fields: report.add( "poa binds %s" % field, parsed[field] == expected[field], "proof says %r, decision record says %r" % (parsed[field], expected[field]), ) def _cross_check_registry(decision, canonical, report): """Assert the registry string names the agent this decision belongs to.""" parts = canonical.split("|") if len(parts) != len(REGISTRY_FIELDS): report.add( "poa registry shape", False, "expected %d fields, found %d" % (len(REGISTRY_FIELDS), len(parts)), ) return parsed = dict(zip(REGISTRY_FIELDS, parts)) for field in ("agent_id", "agent_type"): report.add( "poa registry binds %s" % field, parsed[field] == str(decision.get(field, "")), "proof says %r, decision record says %r" % (parsed[field], decision.get(field)), ) # -------------------------------------------------------------------------- # Decision record vs chained payload # -------------------------------------------------------------------------- def verify_decision_matches_chain(bundle, report): """The readable decision block must match what was actually hashed. A bundle presents the decision twice: once as a convenient `decision` object and once inside the chained event payload. Only the payload is covered by the hashes, so an edit to `decision` alone would otherwise slip through every check above. """ decision = bundle.get("decision") events = bundle.get("events") or [] if not decision or not events: return decision_id = str(bundle.get("decision_id") or decision.get("decision_id") or "") chained = None for event in events: payload = event.get("payload") or {} if str(payload.get("decision_id", "")) == decision_id: chained = payload break if chained is None: report.note( "decision record vs chain", "no chained payload for this decision_id; nothing to compare", ) return mismatches = [] for key, value in chained.items(): # Hashes and ids are attached to the entry after it is chained, so the # readable copy legitimately carries fields the payload does not. if key in ("event_id", "payload_hash", "previous_hash", "chain_hash"): continue if key in decision and decision[key] != value: mismatches.append(key) report.add( "decision record vs chain", not mismatches, "fields differ from the hashed payload: %s" % ", ".join(sorted(mismatches)) if mismatches else "readable decision matches the chained payload", ) # -------------------------------------------------------------------------- # Output # -------------------------------------------------------------------------- WHAT_THIS_PROVES = """\ What this verification does and does not establish Proven from the bundle alone: - Each event's payload has not been altered since it was hashed. - Each event's chain_hash is consistent with its payload, id and timestamp. - The Proof-of-Agent digests are correctly derived, AND the canonical strings behind them describe this decision and this agent -- not some other decision with a valid proof. - The readable decision block matches the payload that was actually hashed. NOT proven, and not provable from a bundle: - That this segment belongs to the institution's real audit chain. A bundle is a slice; check its anchor against the full log to place it. - That no events are missing from the wider log. Append-only storage is tamper-EVIDENT, not tamper-PROOF: it detects modification of retained events, and does not by itself prevent deletion. Put the log on immutable (object-lock/WORM) storage if you need that property. - Anything about the correctness of the decision itself. This verifies integrity, not judgement.""" def render_text(bundle, report, verbose): lines = [] lines.append("Sentinel evidence bundle verification") lines.append("=" * 52) lines.append("decision_id : %s" % bundle.get("decision_id", "(none)")) lines.append("exported_at : %s" % bundle.get("exported_at", "(none)")) lines.append("events : %d" % len(bundle.get("events") or [])) lines.append("") failures = report.failures for check in report.checks: if check["severity"] == "info": lines.append(" note %s: %s" % (check["check"], check["detail"])) elif check["ok"]: if verbose: lines.append(" ok %s" % check["check"]) else: lines.append(" FAIL %s" % check["check"]) lines.append(" %s" % check["detail"]) passed = sum(1 for c in report.checks if c["severity"] != "info" and c["ok"]) lines.append("") lines.append("-" * 52) if failures: lines.append("RESULT: FAILED -- %d of %d checks did not pass." % (len(failures), passed + len(failures))) lines.append("This bundle does not match its own hashes. Treat it as " "unverified.") else: lines.append("RESULT: VERIFIED -- all %d checks passed." % passed) lines.append("") lines.append(WHAT_THIS_PROVES) return "\n".join(lines) def main(argv=None): parser = argparse.ArgumentParser( description="Independently verify a Sentinel evidence bundle.", epilog="No dependencies, no network, no Sentinel install required.", ) parser.add_argument("bundle", help="path to an evidence bundle JSON file") parser.add_argument("--json", action="store_true", dest="as_json", help="emit machine-readable JSON") parser.add_argument("-v", "--verbose", action="store_true", help="list passing checks as well as failures") parser.add_argument("-q", "--quiet", action="store_true", help="print nothing; use the exit code") args = parser.parse_args(argv) try: with open(args.bundle, "r") as handle: bundle = json.load(handle) except (IOError, OSError) as exc: sys.stderr.write("cannot read bundle: %s\n" % exc) return 2 except ValueError as exc: sys.stderr.write("bundle is not valid JSON: %s\n" % exc) return 2 if not isinstance(bundle, dict) or "events" not in bundle: sys.stderr.write( "not an evidence bundle: expected a JSON object with an 'events' key\n" ) return 2 report = Report() verify_events(bundle.get("events") or [], report) verify_proof_of_agent(bundle.get("decision") or {}, report) verify_decision_matches_chain(bundle, report) if args.quiet: pass elif args.as_json: print(json.dumps({ "verified": report.ok, "decision_id": bundle.get("decision_id"), "events_checked": len(bundle.get("events") or []), "checks": report.checks, }, indent=2)) else: print(render_text(bundle, report, args.verbose)) return 0 if report.ok else 1 if __name__ == "__main__": sys.exit(main())