email-connect/scripts/inspect-sender-ledger.py

#!/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()