From 65cea0d927dd8f243d40a633e82dd3ce2aabdd64 Mon Sep 17 00:00:00 2001 From: bigpickle Date: Mon, 5 Oct 2026 21:51:01 +0800 Subject: [PATCH] Close the second round of review findings: checkpoints, revocation, WAL, sweep --- sigil/store.py | 204 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 202 insertions(+), 2 deletions(-) diff --git a/sigil/store.py b/sigil/store.py index f355c33..87b1b85 100644 --- a/sigil/store.py +++ b/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(