Harden Sigil: sigil/store.py

sigil/store.py

@@ -11,9 +11,15 @@ from __future__ import annotations
 
 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,
@@ -41,11 +47,13 @@ CREATE TABLE IF NOT EXISTS nonces (
 );
 
 CREATE TABLE IF NOT EXISTS sessions (
-    token        TEXT PRIMARY KEY,
+    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,
@@ -60,12 +68,14 @@ CREATE TABLE IF NOT EXISTS codes (
     scopes          TEXT NOT NULL,
     issued_at       INTEGER NOT NULL,
     redeem_expires_at INTEGER NOT NULL,  -- short: the code is a key
-    session_expires_at INTEGER NOT NULL, -- longer: the session is the door
+    view_ttl         INTEGER NOT NULL,   -- the door opens a full hour, but only
+                                        -- once the key is used
     used_at         INTEGER
 );
 
--- Append-only attestation log. This is the raw material for signed work
+-- Append-only, hash-chained audit log. The chain is what makes this an
--- attestations: who did what, under which session, using which key.
+-- 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,
@@ -73,19 +83,73 @@ CREATE TABLE IF NOT EXISTS events (
     actor      TEXT NOT NULL,
     subject    TEXT,
     fingerprint TEXT,
-    detail     TEXT NOT NULL
+    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);
 """
 
+# 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:"):
         self.conn = sqlite3.connect(path, check_same_thread=False)
         self.conn.row_factory = sqlite3.Row
+        self.conn.execute("PRAGMA foreign_keys = ON")
+        self._write_lock = threading.RLock()
         self.conn.executescript(SCHEMA)
+        self._migrate()
+        self.conn.commit()
+
+    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 ''"
+                )
+        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:
@@ -93,15 +157,25 @@ class Store:
 
     # -- small helpers -----------------------------------------------------
     def _exec(self, sql: str, args: Iterable[Any] = ()) -> sqlite3.Cursor:
-        cur = self.conn.execute(sql, tuple(args))
+        """Run a statement under the write lock.
-        self.conn.commit()
+
-        return cur
+        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]:
-        return self.conn.execute(sql, tuple(args)).fetchone()
+        with self._write_lock:
+            return self.conn.execute(sql, tuple(args)).fetchone()
 
     def _all(self, sql: str, args: Iterable[Any] = ()) -> List[sqlite3.Row]:
-        return self.conn.execute(sql, tuple(args)).fetchall()
+        with self._write_lock:
+            return self.conn.execute(sql, tuple(args)).fetchall()
 
     # -- agents ------------------------------------------------------------
     def put_agent(self, agent: str, pubkey: bytes, fingerprint: str) -> None:
@@ -131,18 +205,25 @@ class Store:
         )
 
     def spend_bootstrap(self, secret_hash: str) -> Optional[str]:
-        """Atomically consume an enrolment secret. Returns the agent or None."""
+        """Atomically consume an enrolment secret. Returns the agent or None.
-        row = self._one(
+
-            "SELECT agent FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL",
+        Compare-and-set: the UPDATE only matches an unspent row, so exactly one
-            (secret_hash,),
+        concurrent caller can win. Check-then-update would let a secret enrol
-        )
+        more than one key.
-        if row is None:
+        """
-            return None
+        with self._write_lock:
-        self._exec(
+            cur = self.conn.execute(
-            "UPDATE bootstrap SET spent_at = ? WHERE secret_hash = ?",
+                "UPDATE bootstrap SET spent_at = ? WHERE secret_hash = ?"
-            (int(time.time()), secret_hash),
+                " AND spent_at IS NULL",
-        )
+                (int(time.time()), secret_hash),
-        return row["agent"]
+            )
+            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:
@@ -154,46 +235,73 @@ class Store:
     def get_nonce(self, nonce: str) -> Optional[sqlite3.Row]:
         return self._one("SELECT * FROM nonces WHERE nonce = ?", (nonce,))
 
-    def mark_nonce_used(self, nonce: str) -> None:
+    def claim_nonce(self, nonce: str) -> bool:
-        self._exec(
+        """Consume a nonce exactly once. False if it was already used.
-            "UPDATE nonces SET used_at = ? WHERE nonce = ?", (int(time.time()), nonce)
+
-        )
+        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: str,
+        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, kind, subject, on_behalf_of, granted_by,"
+            "INSERT INTO sessions (token_hash, kind, subject, on_behalf_of, granted_by,"
-            " scopes, issued_at, expires_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            " fingerprint, revoked_reason, scopes, issued_at, expires_at)"
+            " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
             (
-                token,
+                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: str) -> Optional[sqlite3.Row]:
+    def get_session(self, token_hash: str) -> Optional[sqlite3.Row]:
-        return self._one("SELECT * FROM sessions WHERE token = ?", (token,))
+        return self._one("SELECT * FROM sessions WHERE token_hash = ?", (token_hash,))
 
-    def revoke_session(self, token: str) -> None:
+    def revoke_session(self, token_hash: str) -> None:
         self._exec(
-            "UPDATE sessions SET revoked_at = ? WHERE token = ?",
+            "UPDATE sessions SET revoked_at = ? WHERE token_hash = ? AND revoked_at IS NULL",
-            (int(time.time()), token),
+            (int(time.time()), token_hash),
         )
 
+    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.
+        """
+        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")
 
@@ -204,31 +312,82 @@ class Store:
         agent: str,
         scopes: List[str],
         redeem_expires_at: int,
-        session_expires_at: int,
+        view_ttl: int,
     ) -> None:
         self._exec(
             "INSERT INTO codes (code_hash, agent, scopes, issued_at, redeem_expires_at,"
-            " session_expires_at) VALUES (?, ?, ?, ?, ?, ?)",
+            " view_ttl) VALUES (?, ?, ?, ?, ?, ?)",
             (
                 code_hash,
                 agent,
                 json.dumps(scopes),
                 int(time.time()),
                 redeem_expires_at,
-                session_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 mark_code_used(self, code_hash: str) -> None:
+    def claim_code(self, code_hash: str) -> bool:
-        self._exec(
+        """Consume a one-time code exactly once. False if already used.
-            "UPDATE codes SET used_at = ? WHERE code_hash = ?",
+
-            (int(time.time()), code_hash),
+        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,
@@ -236,19 +395,53 @@ class Store:
         subject: Optional[str] = None,
         fingerprint: Optional[str] = None,
         detail: Optional[Dict[str, Any]] = None,
-    ) -> None:
+    ) -> str:
-        self._exec(
+        """Append one hash-chained event. Returns its digest.
-            "INSERT INTO events (at, kind, actor, subject, fingerprint, detail)"
+
-            " VALUES (?, ?, ?, ?, ?, ?)",
+        Each row folds in the previous row's digest, so the log is
-            (
+        tamper-evident: removing or editing an event breaks every chain after
-                int(time.time()),
+        it, and verify_chain() reports the first bad id.
-                kind,
+        """
-                actor,
+        body = json.dumps(detail or {}, sort_keys=True)
-                subject,
+        with self._write_lock:
-                fingerprint,
+            last = self.conn.execute(
-                json.dumps(detail or {}, sort_keys=True),
+                "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
@@ -273,4 +466,11 @@ class Store:
                     "detail": json.loads(row["detail"]),
                 }
             )
-        return out
+        return out
+    # -- housekeeping ------------------------------------------------------
+    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()),),
+        )