265 lines
11 KiB
Python
265 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Ledger anchor check: verify a hash chain against an external anchor.
|
|
|
|
Read-only by construction, and that is the point:
|
|
|
|
- ``--db`` opens SQLite with ``mode=ro`` (read-only URI). The
|
|
script never appends, never updates, never deletes, and never
|
|
calls ``AuthStore.audit`` -- writing a row would move the chain
|
|
head and make the check itself unreproducible (the same rule
|
|
the ``GET /admin/audit-ledger/verify`` endpoint follows on its
|
|
success path). A check that writes is not a check.
|
|
- ``--export`` reads a JSON export (e.g. from
|
|
``GET /admin/audit-ledger/export``) and verifies it in memory.
|
|
This is the path for production Postgres: this script does NOT
|
|
open a Postgres connection, and it never pretends to. PG
|
|
verification is: export via the admin endpoint, then run this
|
|
script in ``--export`` mode against the stored anchor.
|
|
- ``--write-anchor`` (``--export`` mode only) writes an ANCHOR
|
|
FILE (outside the DB), never the chain, and only after the
|
|
internal chain verification has passed. Writing an anchor from
|
|
a failing chain would launder tampering into a fresh "trusted"
|
|
head, so the script refuses (no file written) when verification
|
|
fails. With ``--db``, take the anchor from the passing check's
|
|
reported head/count via the documented procedure instead.
|
|
|
|
What is checked, in order (all failures are reported, exit 1):
|
|
|
|
1. Internal integrity: linkage (each ``prev_hash`` equals the
|
|
previous ``entry_hash``) + recomputed ``entry_hash`` per row,
|
|
via ``cloud.ledger.verify_chain`` / ``verify_export``. This
|
|
catches middle tampering, reordering, and middle deletion.
|
|
2. Anchor prefix (only when ``--anchor`` is given): the anchor
|
|
records ``{head_hash, count}`` taken at an earlier time. The
|
|
current chain must still contain that prefix: it must have
|
|
at least ``count`` entries, and the entry at position
|
|
``count`` (1-based) must still hash to ``head_hash``. A
|
|
shorter chain means the anchored tail was truncated; a
|
|
different hash at that position means the chain was replaced
|
|
or rewritten before the anchor point. A head that has ADVANCED
|
|
past the anchor (more entries, same prefix) is NORMAL and
|
|
passes -- anchors are points in history, not a frozen head.
|
|
The honest limit, stated not hidden: an anchor cannot detect
|
|
deletion of entries added AFTER that anchor was taken (the
|
|
chain then still ends exactly at the anchor head). That is
|
|
why anchors must be taken frequently: the anchor cadence
|
|
bounds the undetectable window, and the next anchor catches
|
|
what this one could not.
|
|
|
|
Anchor file format (JSON)::
|
|
|
|
{"version": 1, "head_hash": "<64 hex>", "count": <int>,
|
|
"anchored_at": "<ISO timestamp, informational only>",
|
|
"source": "<db path / export description, informational>"}
|
|
|
|
``anchored_at`` and ``source`` are never trusted by the check:
|
|
only ``head_hash`` + ``count`` decide pass/fail.
|
|
|
|
Exit codes: 0 = pass, 1 = verification failure (failures listed
|
|
in the JSON output), 2 = usage / I/O error (missing file, bad
|
|
anchor format, Postgres URL passed as ``--db``). Output is one
|
|
JSON object on stdout, so a cron job or operator can parse it
|
|
without scraping prose.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(
|
|
os.path.abspath(__file__))))
|
|
|
|
from cloud.ledger import verify_chain, verify_export # noqa: E402
|
|
|
|
_HEX64 = re.compile(r"^[0-9a-f]{64}$")
|
|
|
|
|
|
def _load_anchor(path):
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
except (OSError, ValueError) as exc:
|
|
raise ValueError("cannot read anchor %r: %s" % (path, exc))
|
|
if not isinstance(data, dict):
|
|
raise ValueError("anchor %r is not a JSON object" % path)
|
|
head = data.get("head_hash")
|
|
count = data.get("count")
|
|
if not isinstance(head, str) or not _HEX64.match(head):
|
|
raise ValueError(
|
|
"anchor %r: head_hash must be a 64-char lowercase hex"
|
|
" sha256" % path)
|
|
if isinstance(count, bool) or not isinstance(count, int) \
|
|
or count < 0:
|
|
raise ValueError(
|
|
"anchor %r: count must be a non-negative int" % path)
|
|
if count == 0:
|
|
raise ValueError(
|
|
"anchor %r: count 0 anchors nothing (empty chain); "
|
|
"take an anchor only after at least one entry exists"
|
|
% path)
|
|
return {"head_hash": head, "count": count}
|
|
|
|
|
|
def _check_prefix(entry_hashes, anchor):
|
|
"""Anchor-prefix check. Returns a list of failures (empty = pass)."""
|
|
count = anchor["count"]
|
|
if len(entry_hashes) < count:
|
|
return [
|
|
"chain shorter than anchor: current count %d < anchored"
|
|
" count %d (anchored tail truncated or chain replaced)"
|
|
% (len(entry_hashes), count)]
|
|
if entry_hashes[count - 1] != anchor["head_hash"]:
|
|
return [
|
|
"anchor prefix mismatch: entry %d hashes to %s, not the"
|
|
" anchored head %s (chain replaced or rewritten before"
|
|
" the anchor point)"
|
|
% (count, entry_hashes[count - 1], anchor["head_hash"])]
|
|
return []
|
|
|
|
|
|
def _report(ok, failures, count, head_hash, anchor):
|
|
report = {
|
|
"ok": ok,
|
|
"failures": failures,
|
|
"count": count,
|
|
"head_hash": head_hash,
|
|
"anchor": anchor,
|
|
"head_advanced_by": (
|
|
count - anchor["count"] if anchor is not None else None),
|
|
}
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
return 0 if ok else 1
|
|
|
|
|
|
def _run_db(db_path, anchor):
|
|
if "://" in db_path or db_path.startswith("postgres"):
|
|
raise ValueError(
|
|
"--db takes a SQLite file path, not a Postgres URL. "
|
|
"For Postgres, export via GET /admin/audit-ledger/export"
|
|
" and run this script with --export.")
|
|
if not os.path.exists(db_path):
|
|
raise ValueError("db file not found: %r" % db_path)
|
|
uri = "file:%s?mode=ro" % _quote(db_path)
|
|
conn = sqlite3.connect(uri, uri=True)
|
|
try:
|
|
result = verify_chain(conn)
|
|
rows = conn.execute(
|
|
"SELECT entry_hash FROM audit_ledger ORDER BY id ASC"
|
|
).fetchall() if result["count"] else []
|
|
finally:
|
|
conn.close()
|
|
failures = list(result.get("failures") or [])
|
|
entry_hashes = [row[0] for row in rows]
|
|
if anchor is not None:
|
|
failures.extend(_check_prefix(entry_hashes, anchor))
|
|
return _report(not failures, failures, result["count"],
|
|
result["head_hash"], anchor)
|
|
|
|
|
|
def _quote(path):
|
|
# Minimal URI escaping for the read-only open: '#' and '?'
|
|
# would otherwise terminate the URI. A path containing them
|
|
# is unusual; escaping keeps the open honest instead of
|
|
# silently opening a different file.
|
|
return path.replace("?", "%3F").replace("#", "%23")
|
|
|
|
|
|
def _run_export(export_path, anchor):
|
|
try:
|
|
with open(export_path, "r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
except (OSError, ValueError) as exc:
|
|
raise ValueError(
|
|
"cannot read export %r: %s" % (export_path, exc))
|
|
result = verify_export(data)
|
|
entries = data.get("entries") if isinstance(data, dict) else None
|
|
entry_hashes = []
|
|
if isinstance(entries, list):
|
|
entry_hashes = [e.get("entry_hash") for e in entries
|
|
if isinstance(e, dict)]
|
|
failures = list(result.get("failures") or [])
|
|
failures.extend(_check_prefix(entry_hashes, anchor)
|
|
if anchor is not None else [])
|
|
return _report(not failures, failures, result["count"],
|
|
result["head_hash"], anchor)
|
|
|
|
|
|
def _write_anchor(out_path, head_hash, count, source):
|
|
import datetime
|
|
payload = {
|
|
"version": 1,
|
|
"head_hash": head_hash,
|
|
"count": count,
|
|
"anchored_at": datetime.datetime.now(
|
|
datetime.timezone.utc).isoformat(),
|
|
"source": source,
|
|
}
|
|
tmp_path = out_path + ".tmp"
|
|
with open(tmp_path, "w", encoding="utf-8") as handle:
|
|
json.dump(payload, handle, indent=2, sort_keys=True)
|
|
handle.write("\n")
|
|
os.replace(tmp_path, out_path)
|
|
print(json.dumps({"anchor_written": out_path, **payload},
|
|
indent=2, sort_keys=True))
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(
|
|
description="Verify an audit hash chain against an external"
|
|
" anchor (read-only; exit 0 pass, 1 fail, 2 usage).")
|
|
group = parser.add_mutually_exclusive_group(required=True)
|
|
group.add_argument("--db", help="SQLite control.db file path")
|
|
group.add_argument("--export",
|
|
help="JSON file from /admin/audit-ledger/export")
|
|
parser.add_argument("--anchor",
|
|
help="anchor JSON file ({head_hash, count})")
|
|
parser.add_argument("--write-anchor",
|
|
help="after a passing check, write a fresh"
|
|
" anchor JSON to this path (anchor file"
|
|
" only, never the chain; refused on failure)")
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
anchor = _load_anchor(args.anchor) if args.anchor else None
|
|
except ValueError as exc:
|
|
print("error: %s" % exc, file=sys.stderr)
|
|
return 2
|
|
if args.db and args.write_anchor:
|
|
# Refused before any check runs, so a passing report is
|
|
# never followed by a contradictory exit code. Anchor
|
|
# taking from a DB check uses the documented procedure:
|
|
# run the check, then write the reported head/count as a
|
|
# new anchor file (the file, never the chain).
|
|
print("error: --write-anchor is not combined with --db;"
|
|
" take the anchor from the passing check's"
|
|
" head_hash/count via the documented procedure",
|
|
file=sys.stderr)
|
|
return 2
|
|
try:
|
|
if args.db:
|
|
return _run_db(args.db, anchor)
|
|
code = _run_export(args.export, anchor)
|
|
if code == 0 and args.write_anchor:
|
|
# Re-read the export to obtain the verified head.
|
|
with open(args.export, "r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
_write_anchor(args.write_anchor, data["head_hash"],
|
|
len(data["entries"]), args.export)
|
|
elif args.write_anchor:
|
|
print("error: refusing --write-anchor: verification"
|
|
" failed, an anchor from a failing chain would"
|
|
" launder tampering", file=sys.stderr)
|
|
return code
|
|
except ValueError as exc:
|
|
print("error: %s" % exc, file=sys.stderr)
|
|
return 2
|
|
except sqlite3.Error as exc:
|
|
print("error: sqlite: %s" % exc, file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|