#!/usr/bin/env python3 """ tools/verify_gate_log.py -- check an ARBITER gate's public log yourself, trusting nobody: not ARBITER, not its server, not the rest of ARBITER's code, not the gate's own answers (consequence-boundary plan, phase 4; gate/worker.js). This one file is all of it; a copy is published at https://consequenceaisystems.ai/runs/verify_gate_log.py python3 verify_gate_log.py https://arbiter-gate.jerrymcclintock1.workers.dev python3 verify_gate_log.py --file log.json --about about.json (saved copies) It needs Python 3.10+ and the `cryptography` package, and reads two public URLs: /.well-known/arbiter-gate and /v1/log. Then it replays the whole log: 1. the chain: every entry's hash is sha256(prev "\\n" seq "\\n" at "\\n" kind "\\n" body) and names the entry before it; the last one is the head the gate publishes 2. the owner: the first owner passkey, made at setup; from then on every change of authority (activating a passkey, a company, an approver's limit, a payee, ARBITER's key) is an owner's passkey signature over that exact change, checked here 3. every request: signed by an ARBITER key an owner granted for that company, for a payee an owner listed for it 4. every approval and rejection: a passkey signature over the exact terms (request, its body's hash, company, account, amount, currency), by a passkey an owner activated and granted for that company, up to the limit the owner set; a person present and verified (the flags), on the gate's own site (origin and rp id) 5. every receipt: signed by the gate's receipt key, naming what was approved 6. every payment the gate made follows from 1-5 Exit 0 when everything holds; 1 with each problem listed otherwise. """ import argparse import base64 import hashlib import json import sys import urllib.request from decimal import Decimal from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat UP, UV = 0x01, 0x04 def unb64u(s: str) -> bytes: return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) def b64u(b: bytes) -> str: return base64.urlsafe_b64encode(b).rstrip(b"=").decode() def canon(v) -> str: return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def fetch(url: str): # Its own user agent: Cloudflare refuses Python's default one ("Python-urllib", error 1010). headers = {"accept": "application/json", "user-agent": "arbiter-verify-gate-log/1"} with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=30) as r: return json.load(r) def jwk_key(jwk: dict, alg: int): if alg == -7 and jwk.get("kty") == "EC" and jwk.get("crv") == "P-256": return ec.EllipticCurvePublicNumbers(int.from_bytes(unb64u(jwk["x"]), "big"), int.from_bytes(unb64u(jwk["y"]), "big"), ec.SECP256R1()).public_key() if alg == -8 and jwk.get("kty") == "OKP" and jwk.get("crv") == "Ed25519": return Ed25519PublicKey.from_public_bytes(unb64u(jwk["x"])) if alg == -257 and jwk.get("kty") == "RSA": return rsa.RSAPublicNumbers(int.from_bytes(unb64u(jwk["e"]), "big"), int.from_bytes(unb64u(jwk["n"]), "big")).public_key() raise ValueError(f"unsupported key (alg {alg})") def fp_of(key) -> str: return hashlib.sha256(key.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)).hexdigest()[:16] class Verifier: def __init__(self, about: dict): self.about = about self.origin, self.rp_id = about["origin"], about["rp_id"] self.receipt_key = Ed25519PublicKey.from_public_bytes(unb64u(about["receipt_key"]["public_key"])) self.problems: list[str] = [] self.passkeys: dict[str, dict] = {} # fp -> {jwk, alg, cid, owner, status, name} self.by_cid: dict[str, str] = {} self.tenants: dict[str, str] = {} self.grants: dict[tuple, Decimal] = {} # (fp, tenant) -> max self.payees: set[tuple] = set() self.arbiter: dict[str, dict] = {} # fp -> {key, status, tenants} self.requests: dict[str, dict] = {} self.counts = {"owner changes": 0, "requests": 0, "approvals": 0, "rejections": 0, "payments made": 0, "failed payments": 0, "receipts": 0, "refusals logged": 0} self.payments: list[str] = [] def bad(self, seq, msg): self.problems.append(f"#{seq}: {msg}") def assertion(self, seq, a: dict, signed_text: str, what: str): """The passkey (fp) that made this assertion over sha256(signed_text), or None.""" fp = self.by_cid.get(str(a.get("credential_id", "")).rstrip("=")) if fp is None: self.bad(seq, f"{what}: signed by a passkey the log never registered") return None k = self.passkeys[fp] try: cd_bytes, ad = unb64u(a["client_data_json"]), unb64u(a["authenticator_data"]) cd = json.loads(cd_bytes) except (KeyError, ValueError): self.bad(seq, f"{what}: the assertion cannot be read") return None if cd.get("type") != "webauthn.get" or cd.get("challenge") != b64u(hashlib.sha256(signed_text.encode()).digest()): self.bad(seq, f"{what}: the passkey did not sign these exact terms") return None if cd.get("origin") != self.origin or cd.get("crossOrigin") is True: self.bad(seq, f"{what}: signed on {cd.get('origin')!r}, not the gate's own site") return None if ad[:32] != hashlib.sha256(self.rp_id.encode()).digest() or not ad[32] & UP or not ad[32] & UV: self.bad(seq, f"{what}: not this gate's passkey, or no person present and verified") return None data = ad + hashlib.sha256(cd_bytes).digest() try: key, sig = jwk_key(k["jwk"], k["alg"]), unb64u(a["signature"]) if k["alg"] == -7: key.verify(sig, data, ec.ECDSA(hashes.SHA256())) elif k["alg"] == -8: key.verify(sig, data) else: key.verify(sig, data, padding.PKCS1v15(), hashes.SHA256()) except (InvalidSignature, ValueError, KeyError): self.bad(seq, f"{what}: the passkey's signature does not verify") return None return fp def receipt(self, seq, rec: dict, expect: dict): try: self.receipt_key.verify(unb64u(rec["signature"]), rec["body"].encode()) body = json.loads(rec["body"]) except (InvalidSignature, KeyError, ValueError): self.bad(seq, "a receipt that the gate's key did not sign") return None for k, v in expect.items(): if body.get(k) != v: self.bad(seq, f"the receipt says {k}={body.get(k)!r}, not {v!r}") self.counts["receipts"] += 1 return body def entry(self, e: dict): seq, kind, b = e["seq"], e["kind"], json.loads(e["body"]) if kind == "gate_started": if b.get("receipt_key", {}).get("public_key") != self.about["receipt_key"]["public_key"]: self.bad(seq, "the gate started with another receipt key than the one it publishes") elif kind in ("owner_established", "passkey_registered"): try: key = jwk_key(b["jwk"], b["alg"]) except (KeyError, ValueError) as ex: self.bad(seq, f"a passkey whose key cannot be read: {ex}") return if fp_of(key) != b["fp"]: self.bad(seq, "a passkey whose fingerprint is not its key's") if kind == "owner_established" and any(p["owner"] for p in self.passkeys.values()): self.bad(seq, "a second owner made by the setup code (setup must close after the first)") self.passkeys[b["fp"]] = {"jwk": b["jwk"], "alg": b["alg"], "owner": kind == "owner_established", "status": "active" if kind == "owner_established" else "pending", "name": b.get("name")} self.by_cid[str(b["credential_id"]).rstrip("=")] = b["fp"] elif kind == "owner_op": t = json.loads(b["terms"]) by = self.assertion(seq, b["assertion"], b["terms"], f"owner change {t.get('op', {}).get('do')}") if by is None: return if not (self.passkeys[by]["owner"] and self.passkeys[by]["status"] == "active"): self.bad(seq, f"a change signed by {by}, which was not an active owner then") return if t.get("gate") != self.about["gate"]: self.bad(seq, "an owner change signed for another gate") self.counts["owner changes"] += 1 self.apply(seq, t["op"]) elif kind == "arbiter_key_registered": key = Ed25519PublicKey.from_public_bytes(unb64u(b["public_key"])) if fp_of(key) != b["fp"]: self.bad(seq, "an ARBITER key whose fingerprint is not its key's") self.arbiter[b["fp"]] = {"key": key, "status": "pending", "tenants": []} elif kind == "request_received": ak, body = self.arbiter.get(b["key"]), b["body"] r = json.loads(body) self.counts["requests"] += 1 if ak is None or ak["status"] != "active" or r.get("tenant") not in ak["tenants"]: self.bad(seq, f"request {r.get('id')}: not under an ARBITER key an owner granted for that company") else: try: ak["key"].verify(unb64u(b["signature"]), body.encode()) except (InvalidSignature, ValueError): self.bad(seq, f"request {r.get('id')}: ARBITER's signature does not verify") if (r.get("tenant"), r.get("account")) not in self.payees: self.bad(seq, f"request {r.get('id')}: account {r.get('account')} was not a listed payee then") self.requests[r["id"]] = {"r": r, "sha": hashlib.sha256(body.encode()).hexdigest(), "approved_by": None, "terms": None} elif kind in ("request_approved", "request_rejected"): q = self.requests.get(b["request"]) if q is None: self.bad(seq, f"a decision on request {b['request']}, which the log never received") return decision = "approve" if kind == "request_approved" else "reject" r = q["r"] expect = canon({"v": 1, "kind": decision, "gate": self.about["gate"], "request": r["id"], "request_sha256": q["sha"], "tenant": r["tenant"], "account": r["account"], "amount": r["amount"], "currency": r["currency"]}) if b["terms"] != expect: self.bad(seq, f"request {r['id']}: the signed terms are not the request's") return by = self.assertion(seq, b["assertion"], b["terms"], f"{decision} of request {r['id'][:8]}") if by is None: return limit = self.grants.get((by, r["tenant"])) if self.passkeys[by]["status"] != "active" or limit is None: self.bad(seq, f"request {r['id']}: decided by {by}, not granted for that company then") elif decision == "approve" and Decimal(r["amount"]) > limit: self.bad(seq, f"request {r['id']}: {r['amount']} is above {by}'s limit {limit}") self.counts["approvals" if decision == "approve" else "rejections"] += 1 q["approved_by"], q["terms"], q["decision"] = by, b["terms"], decision if decision == "reject": q["final"] = "rejected" elif kind in ("request_executed", "request_failed"): q = self.requests.get(b["request"]) if q is None or q.get("decision") != "approve": self.bad(seq, f"a payment for request {b['request']} that no approval preceded") return r = q["r"] outcome = kind.split("_", 1)[1] self.receipt(seq, b["receipt"], {"request": r["id"], "intent": r["intent"], "amount": r["amount"], "account": r["account"], "approver": q["approved_by"], "outcome": outcome, "terms_sha256": hashlib.sha256(q["terms"].encode()).hexdigest()}) if outcome == "executed": self.counts["payments made"] += 1 self.payments.append(f"#{seq} {r['currency']} {r['amount']} to {r['account']} " f"(company {self.tenants.get(r['tenant'], r['tenant'])}), approved by " f"{self.passkeys[q['approved_by']]['name']} ({q['approved_by']})") else: self.counts["failed payments"] += 1 elif kind == "request_expiry_receipt": q = self.requests.get(b["request"]) if q is not None: self.receipt(seq, b["receipt"], {"request": q["r"]["id"], "outcome": "expired"}) elif kind.endswith("_refused"): self.counts["refusals logged"] += 1 def apply(self, seq, op: dict): d = op.get("do") if d == "activate": k = self.passkeys.get(op["fp"]) if k is None or k["status"] != "pending": self.bad(seq, f"activated {op['fp']}, which was not a pending passkey") return k["status"], k["owner"] = "active", bool(op.get("owner")) elif d == "revoke": if op["fp"] in self.passkeys: self.passkeys[op["fp"]]["status"] = "revoked" elif d == "tenant": self.tenants[op["id"]] = op["label"] elif d == "grant": self.grants[(op["fp"], op["tenant"])] = Decimal(op["max"]) elif d == "ungrant": self.grants.pop((op["fp"], op["tenant"]), None) elif d == "payee": self.payees.add((op["tenant"], op["account"])) elif d == "unpayee": self.payees.discard((op["tenant"], op["account"])) elif d == "arbiter": if op["fp"] in self.arbiter and self.arbiter[op["fp"]]["status"] != "revoked": self.arbiter[op["fp"]].update(status="active", tenants=list(op["tenants"])) elif d == "unarbiter": if op["fp"] in self.arbiter: self.arbiter[op["fp"]]["status"] = "revoked" def main() -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("url", nargs="?", help="the gate, e.g. https://arbiter-gate.jerrymcclintock1.workers.dev") ap.add_argument("--file", help="a saved copy of the log (the JSON of /v1/log, entries from 1)") ap.add_argument("--about", help="a saved copy of /.well-known/arbiter-gate") a = ap.parse_args() if a.url: base = a.url.rstrip("/") about = fetch(base + "/.well-known/arbiter-gate") entries, after = [], 0 while True: page = fetch(f"{base}/v1/log?after={after}&limit=500") entries += page["entries"] if not page["entries"] or page["entries"][-1]["seq"] >= (page.get("head") or {}).get("seq", 0): break after = page["entries"][-1]["seq"] head = page.get("head") elif a.file and a.about: about = json.load(open(a.about)) saved = json.load(open(a.file)) entries, head = saved["entries"], saved.get("head") else: ap.error("give the gate's URL, or --file and --about") v = Verifier(about) prev = "0" * 64 for i, e in enumerate(entries): h = hashlib.sha256(f"{prev}\n{e['seq']}\n{e['at']}\n{e['kind']}\n{e['body']}".encode()).hexdigest() if e["seq"] != i + 1 or e["prev"] != prev or e["hash"] != h: v.bad(e["seq"], "the chain breaks here (an entry was changed, removed or inserted)") break prev = e["hash"] v.entry(e) if head and entries and (head["seq"] != entries[-1]["seq"] or head["hash"] != prev): v.problems.append("the log read is not the head the gate publishes") print(f"Gate {about['gate']} at {about['origin']}: {len(entries)} log entries, head {prev[:16]}") owners = [f"{k['name']} ({fp})" for fp, k in v.passkeys.items() if k["owner"] and k["status"] == "active"] print(f" owners now: {', '.join(owners) or 'none'}") for k, n in v.counts.items(): print(f" {k}: {n}") for p in v.payments: print(f" paid: {p}") if v.problems: print(f"\n{len(v.problems)} PROBLEM(S):") for p in v.problems: print(f" - {p}") return 1 print("\nEverything holds: the chain, every owner change, every request, every approval and every receipt." "\nEvery payment this gate made was approved, over its exact terms, by a passkey an owner granted.") return 0 if __name__ == "__main__": sys.exit(main())