#!/usr/bin/env python3 """ Verify a Kirelta audit log OFFLINE, without trusting Kirelta. python tools/verify_audit_anchor.py export.json python tools/verify_audit_anchor.py export.json --public-key <64-hex> `export.json` is the body of `GET /audit/anchor`: the tenant's audit entries, the signed checkpoints ("anchors") over them, and the Ed25519 public key. WHY THIS FILE SHARES NO CODE WITH KIRELTA, deliberately, down to re-deriving the hash format by hand: the claim being made is "you can check your own audit log without trusting us". A verifier that imported `kirelta.store_sql` would be checking our data with our code -- if the format were wrong, or the check subtly lenient, both sides would agree and both would be wrong. This file imports `hashlib`, `json`, `base64`, `binascii` and `cryptography`, and nothing else. Read it: it is short on purpose, and it is meant to be read by someone who does not trust the vendor. Port it to another language if you would rather not run ours at all -- everything it does is specified below. tests/test_audit_anchor.py::test_the_standalone_verifier_agrees_with_the_store keeps this independent implementation honest by running both over the same data and requiring the same answer. That test is the only link between them, and it exists to catch drift, not to create a dependency. WHAT IT CHECKS 1. Every entry's hash is recomputed from its own content. A stored hash is never believed. 2. Entries chain: each entry's prev_hash is the previous entry's hash. 3. Every anchor's Ed25519 signature verifies under the published public key. 4. Anchors chain to each other via prev_anchor_sig, with no gap in seq -- so a deleted checkpoint shows up even though the checkpoint itself is gone. 5. Each anchor's signed chain_head matches the entry it names. THIS is the check the hash chain cannot do for itself: a log rewritten end to end is internally consistent (1 and 2 pass) and still fails here, because the rewriter cannot produce a signature. 6. The GLOBAL sequence: each anchor also carries its position across Kirelta's whole deployment (`global_seq`) and the signature of whichever checkpoint preceded it there (`prev_global_sig`), both inside the signature. Your export contains only your own anchors, so you will see gaps in `global_seq` where other customers' checkpoints sit -- that is expected and is NOT a finding. What you CAN check offline is that your own anchors' global positions increase and that none of yours is missing from your own sequence. The gap-detection that matters -- an entire workspace's anchors deleted -- is visible to Kirelta's operators across all tenants, and to you if you have archived exports over time: a global position you saw before must never disappear. WHAT A PASS DOES NOT MEAN * Entries written before the first anchor are not covered. Their range is reported explicitly. * If every anchor was deleted, this reports NOT ANCHORED, not "tampered with". Compare against an export or a Decision Report PDF you already hold -- a copy outside the vendor's database is what makes that case detectable. * It says nothing about whether what was logged was true. Only that it has not changed since it was signed. Exit status: 0 verified, 1 a verification FAILURE, 2 nothing to verify (no anchors), 3 bad input. """ import argparse import base64 import binascii import hashlib import json import sys DOMAIN = b"kirelta-audit-anchor-v1\n" GENESIS_HASH = "genesis" GENESIS_ANCHOR = "genesis" def entry_hash(prev_hash, ts, tenant_id, actor, action, detail): """SHA-256 over canonical JSON of the entry plus its predecessor's hash. The float `ts` is serialised by json.dumps exactly as Python's repr renders it -- that is the format Kirelta has written since the chain existed, and changing it would invalidate every chain already stored, so it is reproduced rather than improved. (The ANCHOR payload below does not have this constraint and uses a fixed-width decimal string instead.) """ payload = json.dumps( {"prev": prev_hash, "ts": ts, "tenant_id": tenant_id, "actor": actor, "action": action, "detail": detail}, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def anchor_payload(a): """The exact bytes an anchor's signature covers.""" doc = { "v": int(a.get("v", 1)), "tenant_id": str(a["tenant_id"]), "seq": int(a["seq"]), "chain_head": str(a["chain_head"]), "entry_count": int(a["entry_count"]), "ts": f"{float(a['ts']):.6f}", "prev_anchor_sig": str(a.get("prev_anchor_sig") or GENESIS_ANCHOR), "global_seq": int(a["global_seq"]), "prev_global_sig": str(a.get("prev_global_sig") or GENESIS_ANCHOR), "key_id": str(a["key_id"]), "chain_state": str(a["chain_state"]), } return DOMAIN + json.dumps(doc, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def key_id_for(public_bytes): return hashlib.sha256(bytes(public_bytes)).hexdigest()[:16] def public_hex_supplied_out_of_band(export, public_hex): """True when the key came from somewhere other than the export file itself.""" return bool(public_hex) and public_hex != export.get("public_key") def _is_float4_exact(ts): """True if this timestamp survives a 32-bit float round trip unchanged -- see the call site.""" import struct try: return struct.unpack("f", struct.pack("f", float(ts)))[0] == float(ts) except (OverflowError, ValueError, TypeError): return False def _utc(ts): import datetime return datetime.datetime.fromtimestamp(float(ts), datetime.timezone.utc).strftime( "%Y-%m-%d %H:%M:%S UTC") def check(export, public_hex=None, expect_key=None): """Returns (exit_code, list of report lines).""" out, failures, notes = [], [], [] entries = sorted(export.get("audit") or [], key=lambda r: r["id"]) anchors = sorted(export.get("anchors") or [], key=lambda r: int(r["seq"])) public_hex = public_hex or export.get("public_key") # 1 + 2: recompute the chain from content alone. hashed = [e for e in entries if e.get("entry_hash")] bad_content, bad_link, quantised = [], [], [] expected_prev = None for e in hashed: if entry_hash(e.get("prev_hash"), e["ts"], e.get("tenant_id"), e.get("actor"), e.get("action"), e.get("detail")) != e["entry_hash"]: # A hash that cannot be reproduced is not automatically an altered record. Kirelta ran # on a PostgreSQL column that stored these timestamps as 32-bit floats, rounding them # into ~128-second buckets AFTER the hash was computed from the precise value. Those # entries can never re-verify and were never touched. The test is exact: a value that # has been through a 32-bit float round-trips through one unchanged, and a genuine # full-precision Unix timestamp essentially never does. Check it yourself -- that is # the whole point of this file being readable. (quantised if _is_float4_exact(e["ts"]) else bad_content).append(e["id"]) if expected_prev is None: expected_prev = e.get("prev_hash") or GENESIS_HASH if e.get("prev_hash") != expected_prev: bad_link.append(e["id"]) expected_prev = e["entry_hash"] out.append(f"entries: {len(entries)} ({len(hashed)} hashed, " f"{len(entries) - len(hashed)} predate integrity tracking)") if bad_content: failures.append(f"CONTENT ALTERED: {len(bad_content)} entr(y/ies) no longer hash to their " f"stored hash, first at id {bad_content[0]}") if quantised: out.append(f"unverifiable: {len(quantised)} early entr(y/ies) (up to id {quantised[-1]}, " f"{_utc(next(e['ts'] for e in hashed if e['id'] == quantised[-1]))}) cannot be " f"re-verified -- their timestamps were rounded by the database before a known " f"defect was fixed, so the values their hashes covered no longer exist. This is " f"NOT evidence of alteration, it cannot be repaired, and everything after that " f"point verifies normally.") if bad_link: notes.append(f"linkage broken at {len(bad_link)} entr(y/ies), first at id {bad_link[0]} " f"(content intact where CONTENT ALTERED is not also reported)") # 3 + 4 + 5: the anchors. if not anchors: out.append("anchors: NONE -- this log carries no external evidence. That is not the same " "as a finding, and it is not the same as verified.") for line in failures + notes: out.append(" " + line) return (1 if failures else 2), out if not public_hex: return 3, out + ["no public key: pass --public-key or include public_key in the export"] from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey pub_raw = binascii.unhexlify(public_hex) pub = Ed25519PublicKey.from_public_bytes(pub_raw) fingerprint = key_id_for(pub_raw) by_id = {e["id"]: e for e in entries} if expect_key: if expect_key.strip().lower() != fingerprint: failures.append(f"KEY MISMATCH: you expected key {expect_key.strip().lower()}, this " f"export is signed with {fingerprint}. Either the key rotated (check " f"the published one again) or something is impersonating it. Do not " f"accept the rest of this report as verification.") else: out.append(f"key pinned: matches the fingerprint you supplied ({fingerprint})") elif not public_hex_supplied_out_of_band(export, public_hex): notes.append("the public key came from the same file as the data. That is convenient and " "it is NOT independent evidence -- anyone who could rewrite the log could " "also rewrite the key beside it. Fetch the key from " "https://kirelta.com/security (different infrastructure) or " "/.well-known/kirelta-audit-key and re-run with --expect-key.") verified = 0 expected_seq, expected_sig = 1, GENESIS_ANCHOR last_global = 0 for a in anchors: seq = int(a["seq"]) gs = int(a.get("global_seq") or 0) if gs <= last_global: failures.append(f"anchor #{seq}: its deployment-wide position ({gs}) does not increase " f"-- checkpoints are append-only, so this ordering is impossible") last_global = gs if seq != expected_seq or str(a.get("prev_anchor_sig")) != expected_sig: failures.append(f"anchor #{seq}: the checkpoint sequence is broken (expected #" f"{expected_seq} chained to {expected_sig[:16]}...) -- a checkpoint " f"appears to have been removed") expected_seq, expected_sig = seq + 1, str(a["signature"]) if str(a.get("key_id")) != fingerprint: failures.append(f"anchor #{seq}: signed with key {a.get('key_id')}, but the public key " f"supplied is {fingerprint} -- cannot be checked with this key") continue try: pub.verify(binascii.unhexlify(a["signature"]), anchor_payload(a)) except Exception: failures.append(f"anchor #{seq}: SIGNATURE INVALID -- forged or altered") continue held = (by_id.get(a["head_id"]) or {}).get("entry_hash") if held != a["chain_head"]: failures.append( f"anchor #{seq}: THE LOG NO LONGER MATCHES WHAT WAS SIGNED. Entry {a['head_id']} " f"now holds {str(held)[:16]}..., the signature covers {str(a['chain_head'])[:16]}" f"... A rewritten chain still verifies against itself; this is the check that " f"catches it.") continue verified += 1 latest = anchors[-1] unanchored = len([e for e in hashed if e["id"] > latest["head_id"]]) out.append(f"anchors: {len(anchors)} present, {verified} verified under key {fingerprint}") out.append(f"covered: entries up to id {latest['head_id']} " f"({latest['entry_count']} entries, last signed {_utc(latest['ts'])})") if unanchored: out.append(f"NOT covered: {unanchored} entr(y/ies) written after the last checkpoint") first = anchors[0] if first["chain_state"] != "clean": out.append(f"note: the first checkpoint recorded the chain as '{first['chain_state']}' at " f"the time it was taken. History BEFORE that point is not covered by any " f"signature and is not made trustworthy by these checks.") for line in notes: out.append("note: " + line) for line in failures: out.append("FAIL: " + line) out.append("") out.append("VERIFIED — this audit log matches what was signed." if not failures else "FAILED — see above. Preserve this export and escalate.") return (1 if failures else 0), out def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("export", help="JSON body of GET /audit/anchor") ap.add_argument("--public-key", help="Ed25519 public key as 64 hex characters " "(overrides the one in the file -- use the copy you " "obtained independently)") ap.add_argument("--expect-key", help="the key FINGERPRINT you obtained out of band (from " "kirelta.com/security or /.well-known/kirelta-audit-key, " "which is served from different infrastructure). Fails if " "the export was signed with anything else -- this is what " "stops the same server supplying both the data and the " "key that blesses it.") args = ap.parse_args() try: with open(args.export, encoding="utf-8") as fh: export = json.load(fh) except (OSError, ValueError) as e: print(f"cannot read {args.export}: {e}", file=sys.stderr) return 3 key = args.public_key if key and len(key) != 64: # tolerate a base64 key, since exports print both pad = key + "=" * (-len(key) % 4) for dec in (base64.urlsafe_b64decode, base64.b64decode): try: raw = dec(pad) except (binascii.Error, ValueError): continue if len(raw) == 32: key = raw.hex() break code, lines = check(export, key, args.expect_key) print("=" * 88) print("KIRELTA AUDIT LOG — OFFLINE VERIFICATION") print("=" * 88) for line in lines: print(line) return code if __name__ == "__main__": raise SystemExit(main())