Close the second round of review findings: checkpoints, revocation, WAL, sweep
This commit is contained in:
parent
910ddfde42
commit
65cea0d927
204
sigil/store.py
204
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(
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user