#!/usr/bin/env python3 """Read-only, coherent ledger observation for a supervised sender transition. No application startup, migration, SMTP, receipt mutation or outcome repair. An observation while writers are active never proves a drained receiver. """ from __future__ import annotations import argparse import hashlib import json import sqlite3 import stat from contextlib import closing from datetime import UTC, datetime from pathlib import Path TABLES = { "deliveries", "verifications", "suppressions", "send_attempts", "verification_attempts", "caller_buckets", } MAX_BYTES = 128 * 1024**2 class Refused(RuntimeError): pass def inspect(path): path = Path(path).absolute() if (any(p.is_symlink() for p in [path, *path.parents]) or not stat.S_ISREG(path.stat().st_mode)): raise Refused("real_ledger_required") with closing(sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=5)) as db: db.execute("PRAGMA trusted_schema=OFF") db.execute("PRAGMA query_only=ON") db.execute("BEGIN") pages = db.execute("PRAGMA page_count").fetchone()[0] size = db.execute("PRAGMA page_size").fetchone()[0] if not 0 < pages * size <= MAX_BYTES: raise Refused("ledger_size_outside_contract") if db.execute("PRAGMA integrity_check").fetchall() != [("ok",)]: raise Refused("ledger_integrity_failed") schema = list(db.execute( "SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name")) tables = {row[1] for row in schema if row[0] == "table" and not row[1].startswith("sqlite_")} if tables != TABLES: raise Refused("unexpected_ledger_schema") # Same representation as ledger_archive.fingerprint, but this read and # every outcome query share one transaction, including committed WAL. fingerprint = { "schema_sha256": hashlib.sha256(json.dumps(schema).encode()).hexdigest(), "user_version": db.execute("PRAGMA user_version").fetchone()[0], "tables": {}, } for table in sorted(tables): rows = [json.dumps(row, separators=(",", ":"), default=lambda value: {"bytes_hex": value.hex()}).encode() for row in db.execute('SELECT * FROM "' + table + '"')] fingerprint["tables"][table] = { "rows": len(rows), "sha256": hashlib.sha256(b"\n".join(sorted(rows))).hexdigest(), } def count(sql): return db.execute(sql).fetchone()[0] outcomes = { "accepted_invitations": count("SELECT COUNT(*) FROM deliveries"), # Successful reserve_send rows stay in_flight; deliveries takes # precedence in the application's status and retry logic. "unknown_invitations": count( "SELECT COUNT(*) FROM send_attempts a WHERE state='in_flight' " "AND NOT EXISTS (SELECT 1 FROM deliveries d WHERE d.event_id=a.event_id)"), "failed_invitations": count( "SELECT COUNT(*) FROM send_attempts a WHERE state='failed' " "AND NOT EXISTS (SELECT 1 FROM deliveries d WHERE d.event_id=a.event_id)"), "accepted_verifications": count( "SELECT COUNT(*) FROM verification_attempts WHERE state='accepted'"), "unknown_verifications": count( "SELECT COUNT(*) FROM verification_attempts WHERE state='in_flight'"), "failed_verifications": count( "SELECT COUNT(*) FROM verification_attempts WHERE state='failed'"), "unlinked_verifications": count( "SELECT COUNT(*) FROM verifications v WHERE NOT EXISTS " "(SELECT 1 FROM verification_attempts a WHERE a.request_id=v.request_id)"), "unrecognized_attempt_states": count( "SELECT COUNT(*) FROM send_attempts WHERE state NOT IN ('in_flight','failed')" ) + count( "SELECT COUNT(*) FROM verification_attempts " "WHERE state NOT IN ('in_flight','accepted','failed')"), "orphan_verification_attempts": count( "SELECT COUNT(*) FROM verification_attempts a WHERE NOT EXISTS " "(SELECT 1 FROM verifications v WHERE v.request_id=a.request_id)"), "consumed_challenges": count( "SELECT COUNT(*) FROM verifications WHERE consumed_at IS NOT NULL"), "canceled_challenges": count( "SELECT COUNT(*) FROM verifications WHERE canceled_at IS NOT NULL"), "suppressed_recipients": count("SELECT COUNT(*) FROM suppressions"), } db.rollback() return { "schema": "email-connect.sender-ledger-observation.v1", "observed_at": datetime.now(UTC).isoformat(), "ledger_fingerprint": fingerprint, "outcomes": outcomes, "needs_outcome_reconciliation": any(outcomes[key] for key in ( "unknown_invitations", "unknown_verifications", "unrecognized_attempt_states", "orphan_verification_attempts")), "consistent_read_transaction": True, "writers_fenced": False, "rollout_authorized": False, "mail_sent": False, "private_rows_printed": False, } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--database", type=Path, required=True) args = parser.parse_args() try: result = inspect(args.database) except (OSError, ValueError, sqlite3.Error, Refused): raise SystemExit("ledger_observation_refused") from None print(json.dumps(result, sort_keys=True)) if __name__ == "__main__": main()