sigil/sigil/store.py

"""Storage for identities, challenges, sessions, one-time codes and audit events.

SQLite rather than a service: a component with one writer and a file is far
easier to reason about than a networked store, and Sigil's whole job is to be a
trust anchor, so its own storage should be the boring thing.

Time is always UTC epoch seconds from time.time().
"""

from __future__ import annotations

import base64
import json
import sqlite3
import threading
import time
from typing import Any, Dict, Iterable, List, Optional

# Chain state for the tamper-evident audit log. Genesis value; each event folds
# in the previous event's digest, so removing or editing any row breaks every
# link after it and the break is detectable.
GENESIS = "0" * 64

SCHEMA = """
CREATE TABLE IF NOT EXISTS agents (
    agent        TEXT PRIMARY KEY,
    pubkey       BLOB NOT NULL,
    fingerprint  TEXT NOT NULL,
    enrolled_at  INTEGER NOT NULL,
    revoked_at   INTEGER,
    -- NULL means the revocation is permanent; a timestamp means the agent
    -- comes back into force when it passes. "Suspended for a week" and
    -- "this identity is dead" are different statements and the schema should
    -- not pretend otherwise.
    revoked_until INTEGER,
    revoked_reason TEXT
);

-- Enrolment bootstrap secrets. Spent secrets are deleted, not marked, so a
-- spent secret cannot be replayed even if the table is read.
CREATE TABLE IF NOT EXISTS bootstrap (
    secret_hash  TEXT PRIMARY KEY,
    agent        TEXT NOT NULL,
    created_at   INTEGER NOT NULL,
    spent_at     INTEGER
);

CREATE TABLE IF NOT EXISTS nonces (
    nonce        TEXT PRIMARY KEY,
    agent        TEXT NOT NULL,
    issued_at    INTEGER NOT NULL,
    expires_at   INTEGER NOT NULL,
    used_at      INTEGER
);

CREATE TABLE IF NOT EXISTS sessions (
    token_hash    TEXT PRIMARY KEY,
    kind         TEXT NOT NULL,          -- 'agent' | 'human_via_agent'
    subject      TEXT NOT NULL,          -- who this session acts as
    on_behalf_of TEXT NOT NULL,          -- the agent, in both kinds
    granted_by   TEXT,                   -- agent that issued a view session
    fingerprint  TEXT,                   -- key the session was minted against
    revoked_reason TEXT,
    scopes       TEXT NOT NULL,          -- JSON list
    issued_at    INTEGER NOT NULL,
    expires_at   INTEGER NOT NULL,
    revoked_at   INTEGER
);

-- One-time codes for humans viewing through an agent. Stored as a hash: a
-- database leak must not yield usable codes.
CREATE TABLE IF NOT EXISTS codes (
    code_hash       TEXT PRIMARY KEY,
    agent           TEXT NOT NULL,
    scopes          TEXT NOT NULL,
    issued_at       INTEGER NOT NULL,
    redeem_expires_at INTEGER NOT NULL,  -- short: the code is a key
    view_ttl         INTEGER NOT NULL,   -- the door opens a full hour, but only
                                        -- once the key is used
    used_at         INTEGER
);

-- Append-only, hash-chained audit log. The chain is what makes this an
-- attestation substrate rather than a convenience log: an edit or deletion
-- breaks every digest after it, and verify_chain() says where.
CREATE TABLE IF NOT EXISTS events (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    at         INTEGER NOT NULL,
    kind       TEXT NOT NULL,
    actor      TEXT NOT NULL,
    subject    TEXT,
    fingerprint TEXT,
    detail     TEXT NOT NULL,
    prev_hash  TEXT NOT NULL,
    hash       TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS events_actor ON events(actor, at);
CREATE INDEX IF NOT EXISTS events_subject ON events(subject, at);

-- Sliding-window rate limits. Without these, challenge() is an unmetered
-- signing oracle for anyone who can reach it.
CREATE TABLE IF NOT EXISTS rate_limits (
    bucket      TEXT NOT NULL,
    at          INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS rate_limits_bucket ON rate_limits(bucket, at);

-- Signed anchors over the audit chain.
--
-- The chain alone stops an accidental edit or deletion. It does not stop
-- someone with write access to this database: they can rewrite events and
-- recompute every hash after them, and verify_chain will happily agree. A
-- checkpoint closes that, because it is signed by a key this database does not
-- contain. Rewriting the log without the signing key is detectable.
CREATE TABLE IF NOT EXISTS checkpoints (
    id       INTEGER PRIMARY KEY AUTOINCREMENT,
    at       INTEGER NOT NULL,
    label    TEXT,
    head     TEXT NOT NULL,          -- hash of the last event at signing time
    events   INTEGER NOT NULL,
    signer   TEXT NOT NULL,          -- agent whose key signed this
    fingerprint TEXT NOT NULL,
    signature TEXT NOT NULL          -- base64, over the canonical payload
);
"""

# Defaults: generous enough not to bite an agent, tight enough to matter.
RATE_LIMITS = {
    "challenge": (30, 60),      # 30 per 60s
    "verify": (60, 60),
    "enrol": (10, 300),
    "code_issue": (20, 300),
    "code_redeem": (20, 60),
    "code_redeem_fail": (10, 300),
}


class Store:
    def __init__(self, path: str = ":memory:", anchor_path: Optional[str] = None):
        """Open a Sigil store.

        `anchor_path`, when set, is a filesystem path *outside* this database
        where every checkpoint also publishes its signed tip. Rewriting or
        deleting rows in the SQLite file cannot erase that published tip.
        """
        self.path = path
        self.anchor_path = anchor_path
        self.conn = sqlite3.connect(path, check_same_thread=False, timeout=10.0)
        self.conn.row_factory = sqlite3.Row
        # Writers wait rather than failing instantly. Without this a second
        # process gets "database is locked" the moment a transaction overlaps,
        # which turns ordinary concurrency into an outage.
        self.conn.execute("PRAGMA busy_timeout = 10000")
        self.conn.execute("PRAGMA foreign_keys = ON")
        # WAL lets a reader run while a writer holds the file, so an audit read
        # cannot block an authentication. Not available for :memory:.
        if path != ":memory:":
            self.conn.execute("PRAGMA journal_mode = WAL")
        # The in-process lock serialises threads sharing this object. It does
        # nothing across processes; the single-statement compare-and-set writes
        # carry that guarantee, since SQLite serialises the write itself.
        self._write_lock = threading.RLock()
        self.conn.executescript(SCHEMA)
        self._migrate()
        self.conn.commit()

    def _txn(self, work):
        """Run `work` under the write lock, rolling back if it raises.

        sqlite3 leaves an implicit transaction open on error. Without an
        explicit rollback the next unrelated commit would then persist a
        half-applied change from a failed call.
        """
        with self._write_lock:
            try:
                return work()
            except Exception:
                self.conn.rollback()
                raise

    def _migrate(self) -> None:
        """Bring an older file forward.

        Two changes since the first draft: the events table gained
        prev_hash/hash so the log can be tamper-evident, and codes replaced a
        stored session_expires_at with view_ttl, because the session window must
        start when the code is redeemed rather than when it was minted.
        """
        have = {
            row["name"] for row in self.conn.execute("PRAGMA table_info(events)")
        }
        for column in ("prev_hash", "hash"):
            if column not in have:
                self.conn.execute(
                    f"ALTER TABLE events ADD COLUMN {column} TEXT NOT NULL DEFAULT ''"
                )
        agent_cols = {
            row["name"] for row in self.conn.execute("PRAGMA table_info(agents)")
        }
        for column, ddl in (
            ("revoked_until", "INTEGER"),
            ("revoked_reason", "TEXT"),
        ):
            if column not in agent_cols:
                self.conn.execute(f"ALTER TABLE agents ADD COLUMN {column} {ddl}")
        code_cols = {
            row["name"] for row in self.conn.execute("PRAGMA table_info(codes)")
        }
        if "view_ttl" not in code_cols:
            self.conn.execute("ALTER TABLE codes ADD COLUMN view_ttl INTEGER")
            if "session_expires_at" in code_cols:
                self.conn.execute(
                    "UPDATE codes SET view_ttl ="
                    " MAX(0, session_expires_at - issued_at) WHERE view_ttl IS NULL"
                )
            self.conn.execute(
                "UPDATE codes SET view_ttl = 3600 WHERE view_ttl IS NULL"
            )
        self.conn.commit()

    def close(self) -> None:
        self.conn.close()

    # -- small helpers -----------------------------------------------------
    def _exec(self, sql: str, args: Iterable[Any] = ()) -> sqlite3.Cursor:
        """Run a statement under the write lock.

        Single-use claims depend on compare-and-set. SQLite serialises
        statements on one connection, but callers share this object across
        threads, so the lock is what makes read-modify-write sequences here
        actually atomic rather than merely interleaved.
        """
        with self._write_lock:
            cur = self.conn.execute(sql, tuple(args))
            self.conn.commit()
            return cur

    def _one(self, sql: str, args: Iterable[Any] = ()) -> Optional[sqlite3.Row]:
        with self._write_lock:
            return self.conn.execute(sql, tuple(args)).fetchone()

    def _all(self, sql: str, args: Iterable[Any] = ()) -> List[sqlite3.Row]:
        with self._write_lock:
            return self.conn.execute(sql, tuple(args)).fetchall()

    # -- agents ------------------------------------------------------------
    def put_agent(self, agent: str, pubkey: bytes, fingerprint: str) -> bool:
        """Create an agent row if absent. Returns True on insert, False if already enrolled.

        INSERT OR IGNORE so two concurrent enrols cannot both win: compare-and-set on
        the primary key. Callers that need to replace a key must use update_agent_key,
        which preserves revocation columns.
        """
        with self._write_lock:
            cur = self.conn.execute(
                "INSERT OR IGNORE INTO agents"
                " (agent, pubkey, fingerprint, enrolled_at)"
                " VALUES (?, ?, ?, ?)",
                (agent, pubkey, fingerprint, int(time.time())),
            )
            self.conn.commit()
            return cur.rowcount == 1

    def update_agent_key(self, agent: str, pubkey: bytes, fingerprint: str) -> bool:
        """Replace pubkey/fingerprint without touching revocation columns.

        INSERT OR REPLACE would NULL out revoked_at / revoked_until / revoked_reason
        because those columns are omitted from the insert list — that is how rekey
        used to un-revoke. An UPDATE of only the key columns keeps the ban intact.
        """
        with self._write_lock:
            cur = self.conn.execute(
                "UPDATE agents SET pubkey = ?, fingerprint = ? WHERE agent = ?",
                (pubkey, fingerprint, agent),
            )
            self.conn.commit()
            return cur.rowcount == 1

    def get_agent(self, agent: str) -> Optional[sqlite3.Row]:
        return self._one("SELECT * FROM agents WHERE agent = ?", (agent,))

    def revoke_agent(self, agent: str, until: Optional[int] = None,
                     reason: Optional[str] = None) -> int:
        """Suspend an agent. `until` of None is permanent.

        Returns how many live sessions were killed.
        """
        with self._write_lock:
            self.conn.execute(
                "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
                " WHERE agent = ?",
                (int(time.time()), until, reason, agent),
            )
            # Drop outstanding challenges so an in-flight exchange cannot
            # complete after the revocation lands.
            self.conn.execute(
                "DELETE FROM nonces WHERE agent = ? AND used_at IS NULL", (agent,)
            )
            self.conn.commit()
        return self.revoke_sessions_for_agent(agent, reason or "agent revoked")

    def reinstate(self, agent: str) -> None:
        self._exec(
            "UPDATE agents SET revoked_at = NULL, revoked_until = NULL,"
            " revoked_reason = NULL WHERE agent = ?",
            (agent,),
        )

    def agents(self) -> List[str]:
        return [r["agent"] for r in self._all("SELECT agent FROM agents ORDER BY agent")]

    # -- bootstrap ---------------------------------------------------------
    def put_bootstrap(self, secret_hash: str, agent: str) -> None:
        """Register one unspent bootstrap secret for an agent.

        Any prior unspent secrets for the same agent are spent first so only
        one live seed exists at a time (ops double-seed or a chat leak of an
        unused seed must not leave a standing rekey capability).
        """
        with self._write_lock:
            now = int(time.time())
            self.conn.execute(
                "UPDATE bootstrap SET spent_at = ? WHERE agent = ? AND spent_at IS NULL",
                (now, agent),
            )
            self.conn.execute(
                "INSERT INTO bootstrap (secret_hash, agent, created_at) VALUES (?, ?, ?)",
                (secret_hash, agent, now),
            )
            self.conn.commit()

    def clear_bootstrap_for_agent(self, agent: str, *, keep_hash: Optional[str] = None) -> int:
        """Spend (or delete) every unspent bootstrap secret for an agent.

        After a successful enrol / enrol_rekey, leftover seeds must not remain
        usable. keep_hash, when set, is left alone (already consumed by CAS).
        """
        with self._write_lock:
            if keep_hash is None:
                cur = self.conn.execute(
                    "UPDATE bootstrap SET spent_at = ? WHERE agent = ? AND spent_at IS NULL",
                    (int(time.time()), agent),
                )
            else:
                cur = self.conn.execute(
                    "UPDATE bootstrap SET spent_at = ? WHERE agent = ? AND spent_at IS NULL"
                    " AND secret_hash != ?",
                    (int(time.time()), agent, keep_hash),
                )
            self.conn.commit()
            return cur.rowcount

    def spend_bootstrap(self, secret_hash: str) -> Optional[str]:
        """Atomically consume an enrolment secret. Returns the agent or None.

        Compare-and-set: the UPDATE only matches an unspent row, so exactly one
        concurrent caller can win. Check-then-update would let a secret enrol
        more than one key.
        """
        with self._write_lock:
            cur = self.conn.execute(
                "UPDATE bootstrap SET spent_at = ? WHERE secret_hash = ?"
                " AND spent_at IS NULL",
                (int(time.time()), secret_hash),
            )
            self.conn.commit()
            if cur.rowcount != 1:
                return None
            row = self.conn.execute(
                "SELECT agent FROM bootstrap WHERE secret_hash = ?", (secret_hash,)
            ).fetchone()
            return row["agent"] if row else None

    # -- nonces ------------------------------------------------------------
    def put_nonce(self, nonce: str, agent: str, expires_at: int) -> None:
        self._exec(
            "INSERT INTO nonces (nonce, agent, issued_at, expires_at) VALUES (?, ?, ?, ?)",
            (nonce, agent, int(time.time()), expires_at),
        )

    def get_nonce(self, nonce: str) -> Optional[sqlite3.Row]:
        return self._one("SELECT * FROM nonces WHERE nonce = ?", (nonce,))

    def claim_nonce(self, nonce: str) -> bool:
        """Consume a nonce exactly once. False if it was already used.

        Called before the signature is judged, so a bad attempt still burns the
        nonce. Atomic, so two concurrent verifies cannot both win.
        """
        with self._write_lock:
            cur = self.conn.execute(
                "UPDATE nonces SET used_at = ? WHERE nonce = ? AND used_at IS NULL",
                (int(time.time()), nonce),
            )
            self.conn.commit()
            return cur.rowcount == 1

    # -- sessions ----------------------------------------------------------
    def put_session(
        self,
        token_hash: str,
        kind: str,
        subject: str,
        on_behalf_of: str,
        scopes: List[str],
        expires_at: int,
        granted_by: Optional[str] = None,
        fingerprint: Optional[str] = None,
        revoked_reason: Optional[str] = None,
    ) -> None:
        self._exec(
            "INSERT INTO sessions (token_hash, kind, subject, on_behalf_of, granted_by,"
            " fingerprint, revoked_reason, scopes, issued_at, expires_at)"
            " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (
                token_hash,
                kind,
                subject,
                on_behalf_of,
                granted_by,
                fingerprint,
                revoked_reason,
                json.dumps(scopes),
                int(time.time()),
                expires_at,
            ),
        )

    def get_session(self, token_hash: str) -> Optional[sqlite3.Row]:
        return self._one("SELECT * FROM sessions WHERE token_hash = ?", (token_hash,))

    def revoke_session(self, token_hash: str) -> None:
        self._exec(
            "UPDATE sessions SET revoked_at = ? WHERE token_hash = ? AND revoked_at IS NULL",
            (int(time.time()), token_hash),
        )

    def invalidate_codes_for_agent(self, agent: str) -> int:
        """Mark every unused view code for an agent as spent.

        Revoke, suspend, and rekey must close outstanding codes the same way they
        kill sessions. Leaving a pre-issued code redeemable after compromise
        revoke or key rotation would mint a fresh human_via_agent session while
        the agent is supposed to be dead.
        """
        with self._write_lock:
            cur = self.conn.execute(
                "UPDATE codes SET used_at = ? WHERE agent = ? AND used_at IS NULL",
                (int(time.time()), agent),
            )
            self.conn.commit()
            return cur.rowcount

    def revoke_sessions_for_agent(self, agent: str, reason: str) -> int:
        """Kill every live session for an agent. Used on revoke and rekey.

        Rotating a key must not leave old sessions usable, otherwise the new key
        buys nothing after a compromise. Outstanding view codes are invalidated
        in the same breath so a pre-issued code cannot mint a fresh session.
        """
        self.invalidate_codes_for_agent(agent)
        cur = self._exec(
            "UPDATE sessions SET revoked_at = ?, revoked_reason = ?"
            " WHERE on_behalf_of = ? AND revoked_at IS NULL",
            (int(time.time()), reason, agent),
        )
        return cur.rowcount

    def sessions(self) -> List[sqlite3.Row]:
        return self._all("SELECT * FROM sessions ORDER BY issued_at")

    # -- one-time codes ----------------------------------------------------
    def put_code(
        self,
        code_hash: str,
        agent: str,
        scopes: List[str],
        redeem_expires_at: int,
        view_ttl: int,
    ) -> None:
        self._exec(
            "INSERT INTO codes (code_hash, agent, scopes, issued_at, redeem_expires_at,"
            " view_ttl) VALUES (?, ?, ?, ?, ?, ?)",
            (
                code_hash,
                agent,
                json.dumps(scopes),
                int(time.time()),
                redeem_expires_at,
                view_ttl,
            ),
        )

    def get_code(self, code_hash: str) -> Optional[sqlite3.Row]:
        return self._one("SELECT * FROM codes WHERE code_hash = ?", (code_hash,))

    def claim_code(self, code_hash: str) -> bool:
        """Consume a one-time code exactly once. False if already used.

        Atomic, so a code cannot mint two sessions under a race.
        """
        with self._write_lock:
            cur = self.conn.execute(
                "UPDATE codes SET used_at = ? WHERE code_hash = ? AND used_at IS NULL",
                (int(time.time()), code_hash),
            )
            self.conn.commit()
            return cur.rowcount == 1

    # -- rate limits -------------------------------------------------------
    def allow(self, bucket: str, limit: Optional[int] = None,
              window: Optional[int] = None) -> bool:
        """Consume one unit of a rate limit. False when exhausted.

        The bucket is "kind:key", where the caller namespaces by agent or by
        source address and the kind selects the configured quota. "challenge:alice"
        and "challenge:bob" share the "challenge" quota but are counted
        separately, so one noisy agent cannot lock everyone else out.
        """
        kind, _, _key = bucket.partition(":")
        default = RATE_LIMITS.get(kind)
        if default is None:
            return True
        limit = limit if limit is not None else default[0]
        window = window if window is not None else default[1]
        now = int(time.time())
        cutoff = now - window
        with self._write_lock:
            self.conn.execute("DELETE FROM rate_limits WHERE at < ?", (now - 3600,))
            used = self.conn.execute(
                "SELECT COUNT(*) AS n FROM rate_limits WHERE bucket = ? AND at > ?",
                (bucket, cutoff),
            ).fetchone()["n"]
            if used >= limit:
                self.conn.commit()
                return False
            self.conn.execute(
                "INSERT INTO rate_limits (bucket, at) VALUES (?, ?)", (bucket, now)
            )
            self.conn.commit()
            return True

    # -- events ------------------------------------------------------------
    @staticmethod
    def _event_hash(prev_hash: str, at: int, kind: str, actor: str,
                    subject: Optional[str], fingerprint: Optional[str],
                    detail: str) -> str:
        import hashlib

        payload = "\x1f".join([
            prev_hash, str(at), kind, actor, subject or "", fingerprint or "", detail
        ])
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()

    def record(
        self,
        kind: str,
        actor: str,
        subject: Optional[str] = None,
        fingerprint: Optional[str] = None,
        detail: Optional[Dict[str, Any]] = None,
    ) -> str:
        """Append one hash-chained event. Returns its digest.

        Each row folds in the previous row's digest, so the log is
        tamper-evident: removing or editing an event breaks every chain after
        it, and verify_chain() reports the first bad id.
        """
        body = json.dumps(detail or {}, sort_keys=True)
        with self._write_lock:
            last = self.conn.execute(
                "SELECT hash FROM events ORDER BY id DESC LIMIT 1"
            ).fetchone()
            prev = last["hash"] if last and last["hash"] else GENESIS
            at = int(time.time())
            digest = self._event_hash(prev, at, kind, actor, subject, fingerprint, body)
            self.conn.execute(
                "INSERT INTO events (at, kind, actor, subject, fingerprint, detail,"
                " prev_hash, hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                (at, kind, actor, subject, fingerprint, body, prev, digest),
            )
            self.conn.commit()
            return digest

    def verify_chain(self) -> Dict[str, Any]:
        """Walk the whole log and report the first inconsistency.

        Returns {"ok": True} or {"ok": False, "broken_at": id, "reason": ...}.
        An empty log is valid: it is simply the genesis state.
        """
        with self._write_lock:
            rows = self.conn.execute(
                "SELECT * FROM events ORDER BY id ASC"
            ).fetchall()
        prev = GENESIS
        for row in rows:
            if row["prev_hash"] != prev:
                return {"ok": False, "broken_at": row["id"],
                        "reason": "prev_hash does not match previous event"}
            expected = self._event_hash(
                prev, row["at"], row["kind"], row["actor"],
                row["subject"], row["fingerprint"], row["detail"],
            )
            if expected != row["hash"]:
                return {"ok": False, "broken_at": row["id"],
                        "reason": "event content does not match its hash"}
            prev = row["hash"]
        return {"ok": True, "events": len(rows)}

    def events(
        self, actor: Optional[str] = None, limit: int = 100
    ) -> List[Dict[str, Any]]:
        if actor:
            rows = self._all(
                "SELECT * FROM events WHERE actor = ? OR subject = ?"
                " ORDER BY id DESC LIMIT ?",
                (actor, actor, limit),
            )
        else:
            rows = self._all("SELECT * FROM events ORDER BY id DESC LIMIT ?", (limit,))
        out = []
        for row in rows:
            out.append(
                {
                    "at": row["at"],
                    "kind": row["kind"],
                    "actor": row["actor"],
                    "subject": row["subject"],
                    "fingerprint": row["fingerprint"],
                    "detail": json.loads(row["detail"]),
                }
            )
        return out
    # -- checkpoints -------------------------------------------------------
    @staticmethod
    def _checkpoint_payload(at: int, label: str, head: str, events: int,
                            signer: str, fingerprint: str) -> bytes:
        import json as _json

        return _json.dumps(
            {"at": at, "label": label or "", "head": head, "events": events,
             "signer": signer, "fingerprint": fingerprint},
            sort_keys=True, separators=(",", ":"),
        ).encode("utf-8")

    def checkpoint(self, private, signer: str, label: Optional[str] = None) -> dict:
        """Sign the current chain head and record it.

        `private` is an Ed25519 private key. The signature covers the head
        digest, so later verifying that the chain still ends at the same
        committed head is a check against anyone who cannot forge a signature
        with this key.

        When this Store was opened with `anchor_path`, the same signed tip is
        also published to that path. The published tip survives a rewrite of
        this database file.
        """
        from . import keys as _keys
        from . import anchor as _anchor

        public = _keys.public_from_private(private)
        fingerprint = _keys.fingerprint(public)
        if self.get_agent(signer) is None:
            raise ValueError(f"{signer} is not enrolled, so cannot sign a checkpoint")
        with self._write_lock:
            last = self.conn.execute(
                "SELECT id, hash FROM events ORDER BY id DESC LIMIT 1"
            ).fetchone()
            head = last["hash"] if last and last["hash"] else GENESIS
            events = self.conn.execute("SELECT COUNT(*) AS n FROM events").fetchone()["n"]
            at = int(time.time())
            payload = self._checkpoint_payload(at, label or "", head, events,
                                               signer, fingerprint)
            signature = base64.b64encode(private.sign(payload)).decode("ascii")
            self.conn.execute(
                "INSERT INTO checkpoints (at, label, head, events, signer,"
                " fingerprint, signature) VALUES (?, ?, ?, ?, ?, ?, ?)",
                (at, label or "", head, events, signer, fingerprint, signature),
            )
            self.conn.commit()
        result = {"at": at, "label": label, "head": head, "events": events,
                  "signer": signer, "fingerprint": fingerprint,
                  "signature": signature}
        if self.anchor_path:
            published = _anchor.write_tip(self.anchor_path, result)
            result["anchor_path"] = published
        return result

    def publish_anchor(self, private, signer: str, label: Optional[str] = None) -> dict:
        """Checkpoint and require an off-box publish.

        Same as checkpoint(), but refuses if this Store has no anchor_path, so
        a deployment that intends to keep the tip off-box cannot silently fall
        back to in-DB-only.
        """
        if not self.anchor_path:
            raise ValueError(
                "publish_anchor requires Store(..., anchor_path=...); "
                "a checkpoint inside the database alone is not off-box"
            )
        return self.checkpoint(private, signer, label)

    def verify_external_anchors(self) -> Dict[str, Any]:
        """Verify the published tip (and history) against enrolled keys + chain.

        Returns ok=False when:
        - the tip file is missing but an anchor_path was configured
        - a notice signature fails
        - the tip head is no longer present at the recorded event count
          (the DB was rewritten or truncated past the published tip)
        """
        from . import keys as _keys
        from . import anchor as _anchor

        if not self.anchor_path:
            return {"ok": True, "anchors": 0, "skipped": True,
                    "reason": "no anchor_path configured"}
        notices = _anchor.read_log(self.anchor_path)
        tip = _anchor.read_tip(self.anchor_path)
        if tip is None:
            return {"ok": False, "anchors": 0, "failures": [
                {"reason": f"anchor tip missing at {self.anchor_path}"}
            ]}
        # Ensure the tip itself is represented even if the log write failed.
        if not notices or notices[-1].get("head") != tip.get("head"):
            notices = list(notices) + [tip]
        results = []
        for notice in notices:
            agent_row = self.get_agent(notice["signer"])
            if agent_row is None:
                results.append({"ok": False, "head": notice.get("head"),
                                "reason": f"signer {notice['signer']} no longer enrolled"})
                continue
            if agent_row["fingerprint"] != notice.get("fingerprint"):
                results.append({"ok": False, "head": notice.get("head"),
                                "reason": "signer key changed since the tip was published"})
                continue
            public = _keys.from_public_bytes(agent_row["pubkey"])
            checked = _anchor.verify_notice(notice, public)
            if not checked["ok"]:
                results.append({"ok": False, "head": notice.get("head"),
                                "reason": checked["reason"]})
                continue
            found = self.conn.execute(
                "SELECT hash FROM events WHERE id = ?", (notice["events"],)
            ).fetchone()
            actual = GENESIS if notice["events"] == 0 else (
                found["hash"] if found else None
            )
            if actual != notice["head"]:
                results.append({
                    "ok": False,
                    "head": notice["head"],
                    "reason": "published tip head no longer matches the chain "
                              "(database rewrite cannot erase the off-box tip)",
                })
                continue
            results.append({"ok": True, "head": notice["head"],
                            "events": notice["events"], "signer": notice["signer"]})
        bad = [r for r in results if not r["ok"]]
        return {"ok": not bad, "anchors": len(results), "failures": bad,
                "tip": tip, "path": self.anchor_path}

    def checkpoints(self) -> List[Dict[str, Any]]:
        rows = self.conn.execute("SELECT * FROM checkpoints ORDER BY id ASC").fetchall()
        return [dict(r) for r in rows]

    def verify_checkpoints(self) -> Dict[str, Any]:
        """Verify every checkpoint's signature and that its head is still real.

        A checkpoint fails if its signature does not verify, or if the chain no
        longer contains the head it committed to at that position. Both mean the
        log changed in a way the signer did not authorise.
        """
        from . import keys as _keys

        results = []
        for row in self.conn.execute("SELECT * FROM checkpoints ORDER BY id ASC"):
            agent_row = self.get_agent(row["signer"])
            if agent_row is None:
                results.append({"id": row["id"], "ok": False,
                                "reason": f"signer {row['signer']} no longer enrolled"})
                continue
            if agent_row["fingerprint"] != row["fingerprint"]:
                results.append({"id": row["id"], "ok": False,
                                "reason": "signer key changed since the checkpoint was made"})
                continue
            payload = self._checkpoint_payload(
                row["at"], row["label"], row["head"], row["events"],
                row["signer"], row["fingerprint"],
            )
            try:
                public = _keys.from_public_bytes(agent_row["pubkey"])
                # verify() returns None on success and raises on failure, so it
                # must be called, not inspected.
                public.verify(base64.b64decode(row["signature"]), payload)
            except Exception as exc:
                results.append({"id": row["id"], "ok": False,
                                "reason": f"signature check failed: {exc}"})
                continue
            found = self.conn.execute(
                "SELECT hash FROM events WHERE id = ?", (row["events"],)
            ).fetchone()
            actual = found["hash"] if found else GENESIS
            if row["events"] == 0:
                actual = GENESIS
            if actual != row["head"]:
                results.append({"id": row["id"], "ok": False,
                                "reason": "chain head at that position no longer matches"})
                continue
            results.append({"id": row["id"], "ok": True})
        bad = [r for r in results if not r["ok"]]
        return {"ok": not bad, "checkpoints": len(results), "failures": bad}

    # -- housekeeping ------------------------------------------------------
    def sweep(self) -> Dict[str, int]:
        """Drop rows that can no longer affect a decision.

        Sessions and codes are retained after they stop working because the
        audit trail refers to them, but nonces and rate-limit rows are pure
        bookkeeping and grow forever otherwise.
        """
        now = int(time.time())
        with self._write_lock:
            nonces = self.conn.execute(
                "DELETE FROM nonces WHERE expires_at < ?", (now,)
            ).rowcount
            limits = self.conn.execute(
                "DELETE FROM rate_limits WHERE at < ?", (now - 3600,)
            ).rowcount
            self.conn.commit()
        return {"nonces": nonces, "rate_limits": limits, "sessions": 0}

    def lapsed_codes(self) -> List[sqlite3.Row]:
        """Unredeemed codes whose window has closed, for reporting only."""
        return self._all(
            "SELECT * FROM codes WHERE used_at IS NULL AND redeem_expires_at <= ?",
            (int(time.time()),),
        )