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
-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.
+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.
 
 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,18 +262,11 @@ 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),
-            )
-            if cur.rowcount == 1:
-                self.conn.execute(
-                    "INSERT OR IGNORE INTO agent_keys"
-                    " (agent, fingerprint, pubkey, active_from) VALUES (?, ?, ?, ?)",
-                    (agent, fingerprint, pubkey, now),
+                (agent, pubkey, fingerprint, int(time.time())),
             )
             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
-                # 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),
-                        )
+                # Permanent: wipe key material. Empty fingerprint means
+                # "pending rekey" — see identity.has_live_key.
                 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(
-                notice["signer"], notice.get("fingerprint") or ""
-            )
-            if key_row is None:
+            if agent_row["fingerprint"] != notice.get("fingerprint"):
                 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"],
-                     "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)
+            results.append({"ok": True, "head": notice["head"],
+                            "events": notice["events"], "signer": notice["signer"]})
         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 key_row is None:
+            if agent_row["fingerprint"] != row["fingerprint"]:
                 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}
-            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)
+            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, "warnings": warnings}
+        return {"ok": not bad, "checkpoints": len(results), "failures": bad}
 
     # -- 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):
-        # N2: historical checkpoints verify against retired public keys.
+    def test_checkpoint_after_rekey_is_flagged(self):
         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.assertEqual(result["failures"], [])
+        self.assertFalse(result["ok"])
+        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)