diff --git a/sigil/store.py b/sigil/store.py index d7687a0..f355c33 100644 --- a/sigil/store.py +++ b/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 --- attestations: who did what, under which session, using which key. +-- 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, @@ -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)) - self.conn.commit() - return cur + """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]: - 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.""" - row = self._one( - "SELECT agent FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL", - (secret_hash,), - ) - if row is None: - return None - self._exec( - "UPDATE bootstrap SET spent_at = ? WHERE secret_hash = ?", - (int(time.time()), secret_hash), - ) - return row["agent"] + """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: @@ -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: - self._exec( - "UPDATE nonces SET used_at = ? WHERE nonce = ?", (int(time.time()), 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: 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," - " scopes, issued_at, expires_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT INTO sessions (token_hash, kind, subject, on_behalf_of, granted_by," + " 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]: - return self._one("SELECT * FROM sessions WHERE token = ?", (token,)) + 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: str) -> None: + def revoke_session(self, token_hash: str) -> None: self._exec( - "UPDATE sessions SET revoked_at = ? WHERE token = ?", - (int(time.time()), token), + "UPDATE sessions SET revoked_at = ? WHERE token_hash = ? AND revoked_at IS NULL", + (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: - self._exec( - "UPDATE codes SET used_at = ? WHERE code_hash = ?", - (int(time.time()), 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, @@ -236,19 +395,53 @@ class Store: subject: Optional[str] = None, fingerprint: Optional[str] = None, detail: Optional[Dict[str, Any]] = None, - ) -> None: - self._exec( - "INSERT INTO events (at, kind, actor, subject, fingerprint, detail)" - " VALUES (?, ?, ?, ?, ?, ?)", - ( - int(time.time()), - kind, - actor, - subject, - fingerprint, - json.dumps(detail or {}, sort_keys=True), - ), - ) + ) -> 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 @@ -273,4 +466,11 @@ class Store: "detail": json.loads(row["detail"]), } ) - return out \ No newline at end of file + 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()),), + )