From 73b9ab4515dad4384024b8ac4717033e53442d4b Mon Sep 17 00:00:00 2001 From: leora Date: Tue, 6 Oct 2026 06:22:13 +0800 Subject: [PATCH] Fix N2: agent_keys history keeps checkpoints verifiable after revoke Permanent revoke and rekey retired the live pubkey without retaining it, so historical checkpoints and off-box anchors became permanently unverifiable. Add append-only agent_keys (public material only); retire on revoke/rekey; verify against the fingerprint-named historical key. Document a dedicated checkpoint-signer identity in the runbook. --- DESIGN.md | 12 +++- docs/PROD_RUNBOOK.md | 8 +++ sigil/store.py | 148 ++++++++++++++++++++++++++++++++++----- tests/test_production.py | 46 +++++++++++- 4 files changed, 189 insertions(+), 25 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index f1f493a..7d7e1fe 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -201,9 +201,15 @@ history, forges replacements, and rebuilds the chain — `verify_chain` passes. So the head digest also gets **signed checkpoints**, by an enrolled agent whose private key the database does not contain. `verify-chain` runs both checks, and the rewrite is caught by the signature the attacker cannot produce. A forged -checkpoint fails signature verification, a checkpoint from a stranger is refused -at signing time, and a rekey invalidates every earlier checkpoint — which is -correct, and says so rather than failing quietly. +checkpoint fails signature verification, and a checkpoint from a stranger is +refused at signing time. + +Public keys are kept in an append-only `agent_keys` history. Revoke and rekey +*retire* a key rather than destroy it, so honest pre-compromise checkpoints and +off-box anchors stay verifiable against the fingerprint they name. Checkpoints +timestamped after a key's `retired_at` are flagged (not failed). Prefer a +**dedicated checkpoint-signer identity** that is never a working agent, so a +routine agent compromise-revoke does not touch the tip-signing key at all. What this still does not do is survive an attacker who can also delete the checkpoints, because they live in the same database. Set `anchor_path` / diff --git a/docs/PROD_RUNBOOK.md b/docs/PROD_RUNBOOK.md index bca4b37..92b0a65 100644 --- a/docs/PROD_RUNBOOK.md +++ b/docs/PROD_RUNBOOK.md @@ -13,6 +13,14 @@ enable platform glue until an integration review covers session minting. Optional: agent private keys stay **off** the platform host whenever possible; checkpoint signer key preferably offline. +## Dedicated checkpoint-signer identity + +Enrol a separate agent id used **only** for `checkpoint` / `publish_anchor` +(for example `sigil-checkpoint`). Do not use a working agent as the tip signer: +compromising and permanently revoking that agent would otherwise retire its key +(history still verifies, but ops alerting on "key retired" gets noisy). Never +store the checkpoint signer's private key in the Sigil DB — public history only. + ## Ops loop 1. Open with seal + anchor configured (CLI respects the env vars above). diff --git a/sigil/store.py b/sigil/store.py index 22b9f4c..66690c1 100644 --- a/sigil/store.py +++ b/sigil/store.py @@ -36,6 +36,20 @@ CREATE TABLE IF NOT EXISTS agents ( 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 ( @@ -226,6 +240,17 @@ class Store: } 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: @@ -262,12 +287,19 @@ class Store: 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, int(time.time())), + (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 @@ -279,10 +311,33 @@ class Store: 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 @@ -302,12 +357,31 @@ class Store: """ with self._write_lock: if until is None: - # Permanent: wipe key material. Empty fingerprint means - # "pending rekey" — see identity.has_live_key. + # 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 = ?", - (int(time.time()), until, reason, b"", "", agent), + (now, until, reason, b"", "", agent), ) else: self.conn.execute( @@ -663,7 +737,26 @@ class Store: } ) 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: @@ -764,17 +857,20 @@ class Store: if not notices or notices[-1].get("head") != tip.get("head"): notices = list(notices) + [tip] results = [] + warnings = [] for notice in notices: - agent_row = self.get_agent(notice["signer"]) - if agent_row is None: + 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 - if agent_row["fingerprint"] != notice.get("fingerprint"): + 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 changed since the tip was published"}) + "reason": "signer key for this fingerprint is unknown"}) continue - public = _keys.from_public_bytes(agent_row["pubkey"]) + 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"), @@ -794,11 +890,17 @@ class Store: "(database rewrite cannot erase the off-box tip)", }) continue - results.append({"ok": True, "head": notice["head"], - "events": notice["events"], "signer": notice["signer"]}) + 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, - "tip": tip, "path": self.anchor_path} + "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() @@ -814,22 +916,23 @@ class Store: from . import keys as _keys results = [] + warnings = [] 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: + if self.get_agent(row["signer"]) is None: results.append({"id": row["id"], "ok": False, "reason": f"signer {row['signer']} no longer enrolled"}) continue - if agent_row["fingerprint"] != row["fingerprint"]: + 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 changed since the checkpoint was made"}) + "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(agent_row["pubkey"]) + 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) @@ -847,9 +950,16 @@ class Store: results.append({"id": row["id"], "ok": False, "reason": "chain head at that position no longer matches"}) continue - results.append({"id": row["id"], "ok": True}) + 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} + return {"ok": not bad, "checkpoints": len(results), + "failures": bad, "warnings": warnings} # -- housekeeping ------------------------------------------------------ def sweep(self) -> Dict[str, int]: diff --git a/tests/test_production.py b/tests/test_production.py index a3d3121..650e55d 100644 --- a/tests/test_production.py +++ b/tests/test_production.py @@ -142,15 +142,16 @@ class TestSignedCheckpoints(unittest.TestCase): self.assertIn("does not match enrolled key", str(ctx.exception)) self.assertEqual(len(self.store.checkpoints()), before) - def test_checkpoint_after_rekey_is_flagged(self): + def test_checkpoint_after_rekey_still_verifies_via_history(self): + # N2: historical checkpoints verify against retired public keys. secret = identity.seed_bootstrap(self.store, "bigpickle") identity.enrol_rekey( self.store, "bigpickle", keys.export_public_b64(keys.generate()[1]), secret, ) result = self.store.verify_checkpoints() - self.assertFalse(result["ok"]) - self.assertIn("key changed", result["failures"][0]["reason"]) + self.assertTrue(result["ok"], result) + self.assertEqual(result["failures"], []) def test_empty_log_checkpoints_cleanly(self): store, private = helpers.fresh() @@ -491,3 +492,42 @@ class TestRedeemRateLimitIsolation(unittest.TestCase): self.fail("successes must not share the fail bucket") except errors.SigilError: pass + + +class TestKeyHistoryKeepsCheckpointsVerifiable(unittest.TestCase): + """N2: permanent revoke / rekey must not make honest checkpoints unverifiable.""" + + def test_checkpoints_verify_after_signer_permanent_revoke(self): + import tempfile, os + d = tempfile.mkdtemp() + anchor = os.path.join(d, "tip") + store = Store(":memory:", anchor_path=anchor) + private, _ = helpers.enrol(store, "signer") + store.record("identity.enrolled", actor="signer", subject="signer") + store.checkpoint(private, "signer", "pre-compromise") + before = store.verify_checkpoints() + self.assertTrue(before["ok"], before) + anchors_before = store.verify_external_anchors() + self.assertTrue(anchors_before["ok"], anchors_before) + identity.revoke(store, "signer", reason="compromise") + after = store.verify_checkpoints() + self.assertTrue(after["ok"], after) + self.assertEqual(after["failures"], []) + anchors_after = store.verify_external_anchors() + self.assertTrue(anchors_after["ok"], anchors_after) + # Live key wiped; history retained. + self.assertFalse(identity.has_live_key(store.get_agent("signer"))) + hist = store.conn.execute( + "SELECT fingerprint, retired_at FROM agent_keys WHERE agent = ?", + ("signer",), + ).fetchall() + self.assertEqual(len(hist), 1) + self.assertIsNotNone(hist[0]["retired_at"]) + + def test_private_key_never_in_agent_keys(self): + store, private = helpers.fresh("carol") + cols = [r["name"] for r in store.conn.execute("PRAGMA table_info(agent_keys)")] + self.assertNotIn("private", cols) + self.assertNotIn("privkey", cols) + row = store.conn.execute("SELECT * FROM agent_keys WHERE agent = ?", ("carol",)).fetchone() + self.assertEqual(len(row["pubkey"]), 32)