"""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 ); -- Append-only public-key history. Revoke and rekey *retire* a key instead of -- destroying it, so historical checkpoints and off-box anchors stay verifiable. -- Never stores private keys. CREATE TABLE IF NOT EXISTS agent_keys ( agent TEXT NOT NULL, fingerprint TEXT NOT NULL, pubkey BLOB NOT NULL, active_from INTEGER NOT NULL, retired_at INTEGER, retired_reason TEXT, PRIMARY KEY (agent, fingerprint) ); CREATE INDEX IF NOT EXISTS agent_keys_agent ON agent_keys(agent, active_from); -- 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, -- Fingerprint of the agent key at issue time. Redeem refuses if the agent -- has since rekeyed (or been wiped), so an in-flight issue_code cannot -- outlive a revoke/rekey that lands before put_code commits. fingerprint TEXT ); -- 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), # Separate from enrol: junk pre-auth enrol_rekey must not lock out ops # rotating a compromised agent during incident response. "rekey": (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" ) code_cols = { row["name"] for row in self.conn.execute("PRAGMA table_info(codes)") } if "fingerprint" not in code_cols: self.conn.execute("ALTER TABLE codes ADD COLUMN fingerprint TEXT") # Backfill agent_keys from current agents rows (live or wiped-empty skipped). for row in self.conn.execute( "SELECT agent, pubkey, fingerprint, enrolled_at FROM agents" " WHERE fingerprint IS NOT NULL AND fingerprint != ''" " AND length(pubkey) = 32" ): self.conn.execute( "INSERT OR IGNORE INTO agent_keys" " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)", (row["agent"], row["fingerprint"], row["pubkey"], row["enrolled_at"]), ) 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: now = int(time.time()) cur = self.conn.execute( "INSERT OR IGNORE INTO agents" " (agent, pubkey, fingerprint, enrolled_at)" " VALUES (?, ?, ?, ?)", (agent, pubkey, fingerprint, now), ) if cur.rowcount == 1: self.conn.execute( "INSERT OR IGNORE INTO agent_keys" " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)", (agent, fingerprint, pubkey, now), ) 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: now = int(time.time()) prev = self.conn.execute( "SELECT pubkey, fingerprint, enrolled_at FROM agents WHERE agent = ?", (agent,), ).fetchone() cur = self.conn.execute( "UPDATE agents SET pubkey = ?, fingerprint = ? WHERE agent = ?", (pubkey, fingerprint, agent), ) if cur.rowcount == 1 and prev is not None: old_fp = prev["fingerprint"] or "" if old_fp and len(prev["pubkey"] or b"") == 32: self.conn.execute( "INSERT OR IGNORE INTO agent_keys" " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)", (agent, old_fp, prev["pubkey"], prev["enrolled_at"] or now), ) self.conn.execute( "UPDATE agent_keys SET retired_at = ?, retired_reason = ?" " WHERE agent = ? AND fingerprint = ? AND retired_at IS NULL", (now, "rekey", agent, old_fp), ) self.conn.execute( "INSERT OR IGNORE INTO agent_keys" " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)", (agent, fingerprint, pubkey, now), ) 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. A permanent revoke clears the enrolled public key so `reinstate` alone cannot restore a compromised identity — ops must bootstrap + rekey (or enrol) before the agent can authenticate again. Timed suspension keeps the key: when the pause lapses, the same key works. Returns how many live sessions were killed. """ with self._write_lock: if until is None: # Permanent: retire the public key into agent_keys (so historical # checkpoints stay verifiable), then wipe the live slot. now = int(time.time()) prev = self.conn.execute( "SELECT pubkey, fingerprint, enrolled_at FROM agents WHERE agent = ?", (agent,), ).fetchone() if prev is not None: old_fp = prev["fingerprint"] or "" if old_fp and len(prev["pubkey"] or b"") == 32: self.conn.execute( "INSERT OR IGNORE INTO agent_keys" " (agent, fingerprint, pubkey, active_from)" " VALUES (?, ?, ?, ?)", (agent, old_fp, prev["pubkey"], prev["enrolled_at"] or now), ) self.conn.execute( "UPDATE agent_keys SET retired_at = ?, retired_reason = ?" " WHERE agent = ? AND fingerprint = ? AND retired_at IS NULL", (now, reason or "permanent revoke", agent, old_fp), ) self.conn.execute( "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?," " pubkey = ?, fingerprint = ? WHERE agent = ?", (now, until, reason, b"", "", agent), ) else: 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( "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL", (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: """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( "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL", (agent,), ) else: cur = self.conn.execute( "DELETE FROM bootstrap WHERE agent = ? AND spent_at IS NULL" " AND secret_hash != ?", (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 then delete: exactly one concurrent caller can win, and the spent row is removed so a table dump cannot offline-crack a low-entropy custom secret. Matches the schema comment. """ with self._write_lock: row = self.conn.execute( "SELECT agent FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL", (secret_hash,), ).fetchone() if row is None: return None agent = row["agent"] cur = self.conn.execute( "DELETE FROM bootstrap WHERE secret_hash = ? AND spent_at IS NULL", (secret_hash,), ) self.conn.commit() if cur.rowcount != 1: return None return agent # -- 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, fingerprint: Optional[str] = None, ) -> None: self._exec( "INSERT INTO codes (code_hash, agent, scopes, issued_at, redeem_expires_at," " view_ttl, fingerprint) VALUES (?, ?, ?, ?, ?, ?, ?)", ( code_hash, agent, json.dumps(scopes), int(time.time()), redeem_expires_at, view_ttl, fingerprint, ), ) 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 or expired. Atomic CAS includes the expiry predicate so a race at the TTL boundary cannot mint a session slightly after redeem_expires_at. """ with self._write_lock: now = int(time.time()) cur = self.conn.execute( "UPDATE codes SET used_at = ? WHERE code_hash = ? AND used_at IS NULL" " AND redeem_expires_at > ?", (now, code_hash, now), ) 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 def get_key_by_fingerprint(self, agent: str, fingerprint: str): """Return a historical (or live) public key row for verify. Prefers agent_keys history. Falls back to the live agents row when the fingerprint still matches (covers DBs created before agent_keys). Never returns private key material — pubkey only. """ row = self._one( "SELECT * FROM agent_keys WHERE agent = ? AND fingerprint = ?", (agent, fingerprint), ) if row is not None: return row live = self.get_agent(agent) if live is not None and (live["fingerprint"] or "") == fingerprint: return live return None # -- 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) agent_row = self.get_agent(signer) if agent_row is None: raise ValueError(f"{signer} is not enrolled, so cannot sign a checkpoint") enrolled_fp = agent_row["fingerprint"] or "" if not enrolled_fp or fingerprint != enrolled_fp: raise ValueError( f"private key fingerprint does not match enrolled key for {signer}; " "refusing to sign a checkpoint with a stranger key" ) 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 = [] warnings = [] for notice in notices: if self.get_agent(notice["signer"]) is None: results.append({"ok": False, "head": notice.get("head"), "reason": f"signer {notice['signer']} no longer enrolled"}) continue key_row = self.get_key_by_fingerprint( notice["signer"], notice.get("fingerprint") or "" ) if key_row is None: results.append({"ok": False, "head": notice.get("head"), "reason": "signer key for this fingerprint is unknown"}) continue public = _keys.from_public_bytes(key_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 entry = {"ok": True, "head": notice["head"], "events": notice["events"], "signer": notice["signer"]} retired_at = key_row["retired_at"] if "retired_at" in key_row.keys() else None if retired_at is not None and notice.get("at", 0) > retired_at: entry["flagged"] = True entry["reason"] = "notice timestamp is after the signing key was retired" warnings.append(entry) results.append(entry) bad = [r for r in results if not r["ok"]] return {"ok": not bad, "anchors": len(results), "failures": bad, "warnings": warnings, "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 = [] warnings = [] for row in self.conn.execute("SELECT * FROM checkpoints ORDER BY id ASC"): if self.get_agent(row["signer"]) is None: results.append({"id": row["id"], "ok": False, "reason": f"signer {row['signer']} no longer enrolled"}) continue key_row = self.get_key_by_fingerprint(row["signer"], row["fingerprint"]) if key_row is None: results.append({"id": row["id"], "ok": False, "reason": "signer key for this fingerprint is unknown"}) continue payload = self._checkpoint_payload( row["at"], row["label"], row["head"], row["events"], row["signer"], row["fingerprint"], ) try: public = _keys.from_public_bytes(key_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 entry = {"id": row["id"], "ok": True} retired_at = key_row["retired_at"] if "retired_at" in key_row.keys() else None if retired_at is not None and row["at"] > retired_at: entry["flagged"] = True entry["reason"] = "checkpoint timestamp is after the signing key was retired" warnings.append(entry) results.append(entry) bad = [r for r in results if not r["ok"]] return {"ok": not bad, "checkpoints": len(results), "failures": bad, "warnings": warnings} # -- 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()),), )