#!/usr/bin/env python3 """Check an exported Punchless audit chain. Standard library only. python verify_audit_export.py audit-chain-42.ndjson [--tokens DIR] Why this file exists, and why it imports nothing ------------------------------------------------ A verification endpoint we host proves nothing to anyone who doubts us. "Your own software says your own records are fine" is not evidence; it is a claim about a claim. What is worth something is a file the other side can check on their own machine, with a script they can read in five minutes and run without installing anything. So: no dependencies, no network, no configuration. Copy this file and the export onto any machine with Python and run it. If it says the chain is intact, that means every entry's recorded hash is exactly the hash of its own contents plus the entry before it, computed here rather than taken on trust. What an intact chain does and does not prove -------------------------------------------- It proves the export is internally consistent: no entry was edited and none was removed, unless every later entry was rewritten to match. It does not prove the records were truthful when written. Anchors close the "unless" (H-524). Each night the chain's tip was sent to an RFC 3161 timestamp authority (DigiCert's, by default), which signed "this hash existed at this time". The export carries those tokens. This script checks that the chain, recomputed here, reaches each anchored hash at that entry - so nothing up to it was rewritten after it was timestamped. The token's signature is checked by OpenSSL, not by anything we wrote: with --tokens DIR this writes each token to DIR and prints the command. Entries after the last anchor are only as good as the chain. Exit codes: 0 intact, 1 broken, 2 the file could not be read. """ from __future__ import annotations import base64 import hashlib import json import os import sys GENESIS_HASH = "0" * 64 def canonical_payload(entry: dict) -> str: """Rebuild the exact bytes the server hashed. Sorted keys and a fixed separator. If this disagrees with the server by so much as a space, every entry reads as tampered - so the shape here is the contract, and the server's canonical_payload must match it exactly. """ return json.dumps( { "seq": entry["seq"], "company_id": entry["company_id"], "actor_user_id": entry["actor_user_id"], "entity_type": entry["entity_type"], "entity_id": entry["entity_id"], "event_type": entry["event_type"], "summary": entry["summary"], "changes": entry["changes"] or {}, "created_at": entry["created_at"], }, sort_keys=True, separators=(",", ":"), default=str, ) def entry_digest(prev_hash: str, payload: str) -> str: return hashlib.sha256(f"{prev_hash}{payload}".encode("utf-8")).hexdigest() def verify(path: str, tokens_dir: str | None = None) -> int: try: with open(path, encoding="utf-8") as handle: lines = [line for line in (raw.strip() for raw in handle) if line] except OSError as error: print(f"Could not read {path}: {error}", file=sys.stderr) return 2 if not lines: print(f"{path} is empty.", file=sys.stderr) return 2 try: header = json.loads(lines[0]) except json.JSONDecodeError as error: print(f"The first line is not the export header: {error}", file=sys.stderr) return 2 if header.get("format") != "punchless-audit-chain": print( f"{path} is not a Punchless audit export " f"(its format is {header.get('format')!r}).", file=sys.stderr, ) return 2 genesis = header.get("genesis_hash", GENESIS_HASH) print(f"Company {header.get('company_id')}, exported {header.get('exported_at')}") anchors = {a["tip_seq"]: a for a in header.get("anchors") or []} anchors_reached = 0 prev_hash = genesis expected_seq = None checked = 0 for number, line in enumerate(lines[1:], start=2): try: entry = json.loads(line) except json.JSONDecodeError as error: print(f"\nLine {number} is not valid JSON: {error}") return 1 seq = entry.get("seq") if expected_seq is None: expected_seq = seq elif seq != expected_seq: # A gap means an entry was removed. The surviving hashes would still # chain to each other, so only the sequence catches this. print( f"\nBROKEN at entry {expected_seq}: it is missing - the sequence " f"jumps from {expected_seq - 1} to {seq}." ) print(f"{checked} entries verified before the break.") return 1 if entry.get("prev_hash") != prev_hash: print( f"\nBROKEN at entry {seq}: it does not follow the entry before it. " "Its recorded predecessor hash is not the hash of the previous entry." ) print(f"{checked} entries verified before the break.") return 1 recomputed = entry_digest(prev_hash, canonical_payload(entry)) if recomputed != entry.get("entry_hash"): print( f"\nBROKEN at entry {seq}: its contents do not match its hash, so it " "was changed after it was written." ) print(f" recorded: {entry.get('entry_hash')}") print(f" actual: {recomputed}") print(f" summary: {entry.get('summary')!r}") print(f"\n{checked} entries verified before the break.") return 1 anchor = anchors.get(seq) if anchor is not None: if anchor.get("tip_hash") != recomputed: print( f"\nBROKEN at entry {seq}: it was timestamped by {anchor.get('authority')} " f"at {anchor.get('stamped_at')} with a different hash, so the chain up to " "it was rewritten after that." ) print(f" timestamped: {anchor.get('tip_hash')}") print(f" recomputed: {recomputed}") return 1 token = base64.b64decode(anchor.get("token_tsr_base64") or "") if bytes.fromhex(recomputed) not in token: print(f"\nBROKEN at entry {seq}: its timestamp token is not for this hash.") return 1 anchors_reached += 1 prev_hash = entry["entry_hash"] expected_seq = seq + 1 checked += 1 last = expected_seq - 1 if expected_seq is not None else 0 beyond = sorted(s for s in anchors if s > last) if beyond: anchor = anchors[beyond[0]] print( f"\nBROKEN: the export ends at entry {last}, but entry {beyond[0]} was " f"timestamped at {anchor.get('stamped_at')} - entries were removed from the end." ) return 1 print(f"\nINTACT: {checked} entries verified.") print(f"Chain tip: {prev_hash}") print( "\nEvery entry's hash matches its own contents and the entry before it, " "recomputed here rather than taken on trust." ) if anchors: newest = anchors[max(anchors)] print( f"\n{anchors_reached} timestamped tip(s) reached: nothing up to entry {newest['tip_seq']} " f"was changed after {newest['stamped_at']}, when {newest['authority']} signed its hash.\n" f"Entries after {newest['tip_seq']} are consistent with the chain, not yet timestamped." ) if tokens_dir: os.makedirs(tokens_dir, exist_ok=True) print("\nCheck each token's signature with OpenSSL (any CA bundle that trusts the authority):") for seq, anchor in sorted(anchors.items()): name = os.path.join(tokens_dir, f"anchor-{seq}.tsr") with open(name, "wb") as handle: handle.write(base64.b64decode(anchor["token_tsr_base64"])) print( f" openssl ts -verify -digest {anchor['tip_hash']} -in {name} " "-CAfile /etc/ssl/certs/ca-certificates.crt" ) else: print( "\nNo timestamped tips in this export, so this shows consistency only:\n" "nothing was edited or removed without rewriting everything after it." ) return 0 if __name__ == "__main__": args = sys.argv[1:] tokens = None if "--tokens" in args: at = args.index("--tokens") if at + 1 >= len(args): print(__doc__) sys.exit(2) tokens = args[at + 1] del args[at : at + 2] if len(args) != 1: print(__doc__) sys.exit(2) sys.exit(verify(args[0], tokens))