Close the second round of review findings: checkpoints, revocation, WAL, sweep

sigil/store.py

@@ -9,6 +9,7 @@ Time is always UTC epoch seconds from time.time().
 
 from __future__ import annotations
 
+import base64
 import json
 import sqlite3
 import threading
@@ -26,7 +27,13 @@ CREATE TABLE IF NOT EXISTS agents (
     pubkey       BLOB NOT NULL,
     fingerprint  TEXT NOT NULL,
     enrolled_at  INTEGER NOT NULL,
-    revoked_at   INTEGER
+    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
@@ -98,6 +105,24 @@ CREATE TABLE IF NOT EXISTS rate_limits (
     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.
@@ -113,14 +138,39 @@ RATE_LIMITS = {
 
 class Store:
     def __init__(self, path: str = ":memory:"):
-        self.conn = sqlite3.connect(path, check_same_thread=False)
+        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.
 
@@ -137,6 +187,15 @@ class Store:
                 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)")
         }
@@ -194,6 +253,33 @@ class Store:
             (int(time.time()), 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")]
 
@@ -467,7 +553,121 @@ class Store:
                 }
             )
         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.
+        """
+        from . import keys as _keys
+
+        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()
+        return {"at": at, "label": label, "head": head, "events": events,
+                "signer": signer, "fingerprint": fingerprint}
+
+    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(