DESIGN.md

@@ -201,15 +201,9 @@ 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, and a checkpoint from a stranger is
+checkpoint fails signature verification, a checkpoint from a stranger is refused
-refused at signing time.
+at signing time, and a rekey invalidates every earlier checkpoint — which is
-
+correct, and says so rather than failing quietly.
-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` /

docs/PROD_RUNBOOK.md

@@ -13,14 +13,6 @@ 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).

sigil/store.py

@@ -36,20 +36,6 @@ 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 (
@@ -240,17 +226,6 @@ 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:
@@ -287,19 +262,12 @@ 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, now),
+                (agent, pubkey, fingerprint, int(time.time())),
             )
-            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
 
@@ -311,33 +279,10 @@ 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
 
@@ -357,31 +302,12 @@ class Store:
         """
         with self._write_lock:
             if until is None:
-                # Permanent: retire the public key into agent_keys (so historical
+                # Permanent: wipe key material. Empty fingerprint means
-                # checkpoints stay verifiable), then wipe the live slot.
+                # "pending rekey" — see identity.has_live_key.
-                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),
+                    (int(time.time()), until, reason, b"", "", agent),
                 )
             else:
                 self.conn.execute(
@@ -737,26 +663,7 @@ 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:
@@ -857,20 +764,17 @@ class Store:
         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:
+            agent_row = self.get_agent(notice["signer"])
+            if agent_row 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(
+            if agent_row["fingerprint"] != notice.get("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"})
+                                "reason": "signer key changed since the tip was published"})
                 continue
-            public = _keys.from_public_bytes(key_row["pubkey"])
+            public = _keys.from_public_bytes(agent_row["pubkey"])
             checked = _anchor.verify_notice(notice, public)
             if not checked["ok"]:
                 results.append({"ok": False, "head": notice.get("head"),
@@ -890,17 +794,11 @@ class Store:
                               "(database rewrite cannot erase the off-box tip)",
                 })
                 continue
-            entry = {"ok": True, "head": notice["head"],
+            results.append({"ok": True, "head": notice["head"],
-                     "events": notice["events"], "signer": notice["signer"]}
+                            "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}
+                "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()
@@ -916,23 +814,22 @@ class Store:
         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:
+            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
-            key_row = self.get_key_by_fingerprint(row["signer"], row["fingerprint"])
+            if agent_row["fingerprint"] != row["fingerprint"]:
-            if key_row is None:
                 results.append({"id": row["id"], "ok": False,
-                                "reason": "signer key for this fingerprint is unknown"})
+                                "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(key_row["pubkey"])
+                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)
@@ -950,16 +847,9 @@ class Store:
                 results.append({"id": row["id"], "ok": False,
                                 "reason": "chain head at that position no longer matches"})
                 continue
-            entry = {"id": row["id"], "ok": True}
+            results.append({"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),
+        return {"ok": not bad, "checkpoints": len(results), "failures": bad}
-                "failures": bad, "warnings": warnings}
 
     # -- housekeeping ------------------------------------------------------
     def sweep(self) -> Dict[str, int]:

tests/test_production.py

@@ -142,16 +142,15 @@ 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_still_verifies_via_history(self):
+    def test_checkpoint_after_rekey_is_flagged(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.assertTrue(result["ok"], result)
+        self.assertFalse(result["ok"])
-        self.assertEqual(result["failures"], [])
+        self.assertIn("key changed", result["failures"][0]["reason"])
 
     def test_empty_log_checkpoints_cleanly(self):
         store, private = helpers.fresh()
@@ -492,42 +491,3 @@ 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)