#!/usr/bin/env python3 """Check a Gatesi receipt (and optionally its trace) on your own machine. python3 agentgate_verify.py receipt.json python3 agentgate_verify.py receipt.json trace.json python3 agentgate_verify.py receipt.json trace.json --key-id 0123456789abcdef No Gatesi server, no internet, no packages needed (Python 3.7+). The format is described in receipt-v1.md. Exit code 0 = everything checked passed, 1 = something failed, 2 = bad usage. """ import hashlib import json import re import sys # ---- Ed25519 signature check (RFC 8032, verification only, plain Python) ---- _P = 2 ** 255 - 19 _L = 2 ** 252 + 27742317777372353535851937790883648493 _D = -121665 * pow(121666, _P - 2, _P) % _P _I = pow(2, (_P - 1) // 4, _P) def _recover_x(y, sign): if y >= _P: return None x2 = (y * y - 1) * pow(_D * y * y + 1, _P - 2, _P) % _P if x2 == 0: return None if sign else 0 x = pow(x2, (_P + 3) // 8, _P) if (x * x - x2) % _P != 0: x = x * _I % _P if (x * x - x2) % _P != 0: return None if (x & 1) != sign: x = _P - x return x _GY = 4 * pow(5, _P - 2, _P) % _P _GX = _recover_x(_GY, 0) _G = (_GX, _GY, 1, _GX * _GY % _P) def _add(a, b): A = (a[1] - a[0]) * (b[1] - b[0]) % _P B = (a[1] + a[0]) * (b[1] + b[0]) % _P C = 2 * a[3] * b[3] * _D % _P Dd = 2 * a[2] * b[2] % _P E, F, G, H = B - A, Dd - C, Dd + C, B + A return (E * F % _P, G * H % _P, F * G % _P, E * H % _P) def _mul(s, pt): q = (0, 1, 1, 0) while s > 0: if s & 1: q = _add(q, pt) pt = _add(pt, pt) s >>= 1 return q def _eq(a, b): return (a[0] * b[2] - b[0] * a[2]) % _P == 0 and (a[1] * b[2] - b[1] * a[2]) % _P == 0 def _decode(s): if len(s) != 32: return None y = int.from_bytes(s, "little") sign = y >> 255 y &= (1 << 255) - 1 x = _recover_x(y, sign) if x is None: return None return (x, y, 1, x * y % _P) def ed25519_verify(public_key, message, signature): if len(public_key) != 32 or len(signature) != 64: return False A = _decode(public_key) R = _decode(signature[:32]) if A is None or R is None: return False s = int.from_bytes(signature[32:], "little") if s >= _L: return False h = int.from_bytes(hashlib.sha512(signature[:32] + public_key + message).digest(), "little") % _L return _eq(_mul(s, _G), _add(R, _mul(h, A))) # ---- canonical JSON and hashing (see receipt-v1.md, sections 2 and 3) ---- def _num(v): if isinstance(v, bool): return "true" if v else "false" if isinstance(v, int): return str(v) if v != v or v in (float("inf"), float("-inf")): return "null" if v == int(v) and abs(v) < 1e21: return str(int(v)) t = repr(v) m = re.match(r"^(-?[0-9.]+)e([+-])0*([0-9]+)$", t) if m: # python writes 1e-07, JavaScript writes 1e-7 t = m.group(1) + "e" + m.group(2) + m.group(3) return t def _js(v): """JSON text the way JavaScript's JSON.stringify writes it (compact).""" if v is None: return "null" if isinstance(v, (bool, int, float)): return _num(v) if isinstance(v, str): return json.dumps(v, ensure_ascii=False) if isinstance(v, list): return "[" + ",".join(_js(x) for x in v) + "]" return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + _js(x) for k, x in v.items()) + "}" def canon(v): if v is None or isinstance(v, (bool, int, float, str)): return _js(v) if isinstance(v, list): return "[" + ",".join(canon(x) for x in v) + "]" keys = sorted(v.keys(), key=lambda k: k.encode("utf-16-be")) return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + canon(v[k]) for k in keys) + "}" def sha256(text): return hashlib.sha256(text.encode("utf-8")).hexdigest() GENESIS = "0" * 64 SETTLED = ("approved", "denied", "expired", "cancelled") def chain_text(prev, e): decision = e.get("decision") if e.get("type") == "check" and decision in SETTLED: decision = "pending" approval = None if e.get("type") == "check" else e.get("approval_id") arr = [e.get("run_id"), e.get("seq"), e.get("agent_id"), e.get("ts"), e.get("type"), e.get("tool"), e.get("action"), e.get("args"), decision, e.get("reason"), e.get("cost") or 0, e.get("result"), approval] return prev + "|" + _js(arr) def check_trace(trace): """Returns (ok, message, chained_count, head_seq, head_hash).""" if not isinstance(trace, dict) or trace.get("format") != "agentgate-trace/2" or not isinstance(trace.get("events"), list): return False, "this is not an agentgate-trace/2 document", 0, None, None rows = [e for e in trace["events"] if e.get("seq") is not None] rows.sort(key=lambda e: e["seq"]) prev, expect, settled_by = GENESIS, 1, {} for e in rows: if e["seq"] != expect: return False, "event %d is missing" % expect, len(rows), None, None if e.get("prev_hash") != prev: return False, "link to the previous event is broken at event %d" % e["seq"], len(rows), None, None if sha256(chain_text(prev, e)) != e.get("hash"): return False, "content of event %d was changed" % e["seq"], len(rows), None, None if e.get("type") == "approval" and e.get("decision") in SETTLED and e.get("approval_id"): settled_by[e["approval_id"]] = e["decision"] prev, expect = e["hash"], expect + 1 for e in rows: if e.get("type") == "check" and e.get("approval_id") and e.get("decision") in SETTLED: s = settled_by.get(e["approval_id"]) if s and s != e["decision"]: return False, "check decision differs from the recorded approval (event %d)" % e["seq"], len(rows), None, None n = len(rows) return True, "%d chained events link up" % n, n, (rows[-1]["seq"] if n else None), (prev if n else None) def verify(receipt, trace=None, key_id=None): """Returns (all_ok, [(ok, text), ...]).""" out = [] def add(ok, text): out.append((bool(ok), text)) return bool(ok) if not isinstance(receipt, dict) or receipt.get("format") != "agentgate-receipt/1": add(False, "this is not an agentgate-receipt/1 document") return False, out try: iss = receipt["issuer"] pub, sig = iss["public_key"], receipt["signature"] if not (re.fullmatch(r"[0-9a-fA-F]{64}", pub) and re.fullmatch(r"[0-9a-fA-F]{128}", sig)): raise ValueError("bad key or signature text") if iss.get("algorithm") != "Ed25519": raise ValueError("unknown algorithm") receipt["run"]["id"], receipt["chain"]["head_hash"], receipt["calls"]["digest"] except (KeyError, TypeError, ValueError) as e: add(False, "the receipt is missing required fields or they are malformed (%s)" % e) return False, out payload = {k: v for k, v in receipt.items() if k not in ("signature", "calls_detail")} add(ed25519_verify(bytes.fromhex(pub), canon(payload).encode("utf-8"), bytes.fromhex(sig)), "signature matches the receipt contents and the key inside it") kid = sha256(pub)[:16] add(iss.get("key_id") == kid, "key id %s is the hash of the public key" % kid) if key_id: add(key_id.lower() == kid, "key id matches the one you trust (%s)" % key_id) else: out.append((True, "NOTE: you did not say which key id you trust (use --key-id); this only proves the receipt is self-consistent, key id %s" % kid)) if "calls_detail" in receipt: add(sha256(canon(receipt["calls_detail"])) == receipt["calls"]["digest"], "list of proxied calls matches the signed digest") if trace is not None: ok, msg, n, head_seq, head_hash = check_trace(trace) add(ok, msg) if ok: add(trace.get("run", {}).get("id") == receipt["run"]["id"], "trace is for the same run as the receipt") ch = receipt["chain"] add(n == ch.get("events") and head_seq == ch.get("head_seq") and head_hash == ch.get("head_hash"), "trace has exactly the chain head and event count the receipt signed (%d events)" % n) return all(o for o, _ in out), out def main(argv): args, key_id, files = argv[1:], None, [] i = 0 while i < len(args): if args[i] == "--key-id" and i + 1 < len(args): key_id = args[i + 1] i += 2 else: files.append(args[i]) i += 1 if not 1 <= len(files) <= 2: print(__doc__) return 2 try: docs = [json.load(open(f, encoding="utf-8")) for f in files] except (OSError, ValueError) as e: print("Could not read the file: %s" % e) return 2 ok, lines = verify(docs[0], docs[1] if len(docs) > 1 else None, key_id) for good, text in lines: print(("OK: " if good else "PROBLEM: ") + text) print("\nRESULT: " + ("everything checked passed" if ok else "DO NOT TRUST THIS RECEIPT")) return 0 if ok else 1 if __name__ == "__main__": sys.exit(main(sys.argv))