DESIGN.md

@@ -123,9 +123,11 @@ means a future route cannot pass a string where a proof belongs.
 
 `revoke <agent>` with no end time is permanent: the enrolled public key is
 cleared and the identity stays dead until someone bootstraps a fresh secret and
-rekeys (or re-enrols). `reinstate` alone lifts the ban flags but does **not**
-restore the compromised key — that was the footgun. `revoke <agent> --until
-<unix>` is a pause that keeps the key, and it lapses on its own.
+calls `enrol_rekey` (then `reinstate` if still banned). `enrol` will not work —
+the agent row still exists — and would only burn the seed. `reinstate` alone
+lifts the ban flags but does **not** restore the compromised key — that was the
+footgun. `revoke <agent> --until <unix>` is a pause that keeps the key, and it
+lapses on its own.
 
 The distinction matters because the two failure modes are different. A key
 compromise wants death — a lapsing revocation would hand the attacker back in
@@ -139,11 +141,12 @@ close every door the old key could open.
 
 A narrow check-then-insert race can still insert a session or code after the
 revoke/rekey statement has run. The choke point is `sessions.resolve`: every
-use re-checks that the agent is not revoked, still has a live key, and (for
-agent sessions) that the session fingerprint matches the current key epoch.
-View codes store the issuing fingerprint and refuse redeem after rekey or key
-wipe. That is what makes revocation overtake in-flight verify/issue/redeem,
-not the first-pass invalidation alone.
+use re-checks that the agent is not revoked, still has a live key, and that
+the session fingerprint (when present — always for agent sessions, and for
+human_via_agent sessions minted with a fingerprint) matches the current key
+epoch. View codes store the issuing fingerprint and refuse redeem after rekey
+or key wipe. That is what makes revocation overtake in-flight
+verify/issue/redeem, not the first-pass invalidation alone.
 
 ### Existing controls stay
 
@@ -201,9 +204,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` /
@@ -222,7 +231,7 @@ because the review looked for it.
 
 ## Status
 
-106 passing tests, and two hardening passes that closed real defects. The worst were
+158 passing tests, and two hardening passes that closed real defects. The worst were
 concurrency: one-time codes were redeemable repeatedly (753 sessions from 200
 codes under load), and a bootstrap secret could enrol more than one key. Those
 are now atomic compare-and-set operations, each covered by a test that races

README.md

@@ -55,7 +55,7 @@ python3 sigil.py sweep                                # prune stale rows
 python3 -m unittest discover -s tests -v
 ```
 
-130 tests. They cover the refusals, not just the happy paths: replayed
+158 tests. They cover the refusals, not just the happy paths: replayed
 signatures, wrong keys, expired nonces, single-use codes, scope escalation, and
 the invariant below. `test_hardening.py`, `test_production.py`, `test_anchor.py` and
 `test_sealed.py` have one test per defect found in review, so any of them
@@ -163,4 +163,4 @@ and its reasoning, and what is deferred.
 `docs/PROD_RUNBOOK.md` — production checklist: require `SIGIL_SEAL_PASS` and
 `SIGIL_ANCHOR` on separate storage, verify-chain alerts, integration sharp edges.
 
-`requirements.txt` — pinned `cryptography>=43.0.0,<45`.
+`requirements.txt` — `cryptography>=44.0.1,<45`; see `requirements.lock.txt` / `constraints.txt`.

constraints.txt

@@ -0,0 +1,3 @@
+# Exact pin matching requirements.lock.txt floor. Use with:
+#   pip install -r requirements.txt -c constraints.txt
+cryptography==44.0.1

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).
@@ -31,3 +39,8 @@ Optional: agent private keys stay **off** the platform host whenever possible; c
 - SQLCipher / page-level encryption for multi-tenant co-resident hosts.
 - Domain-separated challenge payloads (`sigil-challenge-v1` prefix) — backward-incompatible without a protocol version.
 - Shared cross-process rate limiter.
+
+## Dependencies
+
+Install with `pip install -r requirements.txt -c constraints.txt`
+(or `--require-hashes -r requirements.lock.txt`). Floor is `cryptography>=44.0.1`.

requirements.lock.txt

@@ -0,0 +1,13 @@
+# Hash-pinned lock for Sigil runtime crypto deps.
+# Install: pip install --require-hashes -r requirements.lock.txt
+# Floor cryptography==44.0.1 addresses CVE-2024-12797 in older wheels.
+
+cryptography==44.0.1 \
+    --hash=sha256:53f23339864b617a3dfc2b0ac8d5c432625c80014c25caac9082314e9de56f41
+
+cffi==2.1.1 \
+    --hash=sha256:a931079504ecc49efed7744c476a5c343a92fabf66dec2db95edb1b2fdc770e2
+
+pycparser==3.0 \
+    --hash=sha256:b727414169a36b7d524c1c3e31839a521725078d7b2ff038656844266160a992
+

requirements.txt

@@ -1,3 +1,5 @@
 # Runtime dependency for Ed25519 + AES-GCM (SealedStore / optional key files).
-# Pin a known-good lower bound; upper bound avoids surprise major breaks in CI.
-cryptography>=43.0.0,<45
+# Lower bound 44.0.1: older 43.x/44.0.0 wheels ship OpenSSL affected by
+# CVE-2024-12797 (GHSA-79v4-65xg-pq4g). Upper bound <45 keeps upgrades deliberate.
+# For hash-pinned install see requirements.lock.txt.
+cryptography>=44.0.1,<45

sigil/identity.py

@@ -78,6 +78,10 @@ def enrol(
     _validate_agent_id(agent)
     if not store.allow(f"enrol:{agent}"):
         raise errors.RateLimited(f"too many enrolment attempts for {agent}")
+    # Already enrolled (e.g. after permanent revoke): refuse *before* spending
+    # the seed so ops following a wrong "re-enrol" path do not burn it.
+    if store.get_agent(agent) is not None:
+        raise errors.EnrolmentFailed(f"{agent} is already enrolled; use enrol_rekey")
     # Spend the secret first so a burned secret cannot be retried. The agent
     # row insert below is INSERT OR IGNORE: if two secrets race for the same
     # unenrolled agent, exactly one put_agent wins and the loser fails closed
@@ -285,8 +289,6 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
     Rekey does not clear revocation. A revoked agent stays revoked until an
     explicit reinstate; rotating the key must not be a back door back in.
     """
-    if not store.allow(f"enrol:{agent}"):
-        raise errors.RateLimited(f"too many enrolment attempts for {agent}")
     row = store.get_agent(agent)
     if row is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
@@ -297,6 +299,10 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
         raise errors.UnknownAgent(f"{agent} is revoked")
     owner = store.spend_bootstrap(_hash_secret(secret))
     if owner != agent:
+        # Rate-limit only failed attempts so junk cannot burn the ops path, and
+        # use a dedicated rekey: bucket so exhausting enrol: cannot lock rekey.
+        if not store.allow(f"rekey:{agent}"):
+            raise errors.RateLimited(f"too many rekey attempts for {agent}")
         raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")
     public = keys.import_public_b64(public_b64)
     fingerprint = keys.fingerprint(public)

sigil/sessions.py

@@ -207,11 +207,14 @@ def resolve(store: Store, token: str) -> dict:
         raise errors.SessionExpired("agent revoked")
     if not _identity.has_live_key(agent_row):
         raise errors.SessionExpired("agent has no live key")
-    # Agent sessions are bound to the key epoch they were minted against.
-    # After rekey, old agent:act bearers must die even if revoke_sessions raced.
-    if row["kind"] == AGENT:
+    # Every session that carries a fingerprint is bound to that key epoch.
+    # After rekey, race-surviving agent:act *and* human_via_agent bearers die
+    # even if revoke_sessions missed them (N7).
     sess_fp = row["fingerprint"] or ""
     live_fp = agent_row["fingerprint"] or ""
+    if sess_fp or row["kind"] == AGENT:
+        # Agent sessions always require a matching fingerprint; human sessions
+        # require a match whenever a fingerprint was recorded at mint time.
         if not sess_fp or sess_fp != live_fp:
             raise errors.SessionExpired("session key epoch no longer current")
 

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 (
@@ -134,6 +148,9 @@ 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),
@@ -226,6 +243,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,11 +290,18 @@ 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 +314,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
 
@@ -294,20 +352,39 @@ class Store:
         """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.
+        cannot restore a compromised identity — ops must bootstrap + enrol_rekey
+        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: 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 +740,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:
@@ -692,9 +788,13 @@ class Store:
 
         public = _keys.public_from_private(private)
         fingerprint = _keys.fingerprint(public)
+        from . import identity as _identity
+
         agent_row = self.get_agent(signer)
         if agent_row is None:
             raise ValueError(f"{signer} is not enrolled, so cannot sign a checkpoint")
+        if _identity.is_revoked(agent_row):
+            raise ValueError(f"{signer} is revoked, so cannot sign a checkpoint")
         enrolled_fp = agent_row["fingerprint"] or ""
         if not enrolled_fp or fingerprint != enrolled_fp:
             raise ValueError(
@@ -764,17 +864,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 +897,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 +923,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 +957,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]:

sigil/viewcodes.py

@@ -144,7 +144,6 @@ def issue_code(
         "code_ttl": code_ttl,
         "view_ttl": view_ttl,
         "redeem_expires_at": now + code_ttl,
-        "view_ttl": view_ttl,
     }
 
 

tests/test_hardening.py

@@ -381,19 +381,52 @@ class TestClaimCodeExpiryCAS(unittest.TestCase):
 
 
 class TestEnrolRekeyRateLimit(unittest.TestCase):
-    """R7: enrol_rekey must share the enrol rate-limit bucket."""
+    """N3: enrol_rekey uses a dedicated rekey: bucket; junk cannot lock ops."""
 
-    def test_enrol_rekey_is_rate_limited(self):
+    def test_exhausting_enrol_bucket_does_not_block_rekey(self):
         store, _ = helpers.fresh("limited")
-        # Exhaust enrol bucket.
-        for i in range(20):
+        for _ in range(20):
             store.allow("enrol:limited")
         secret = identity.seed_bootstrap(store, "limited")
-        with self.assertRaises(errors.RateLimited):
+        # Valid rekey must succeed even if enrol: is exhausted.
         identity.enrol_rekey(
             store, "limited", keys.export_public_b64(keys.generate()[1]), secret
         )
 
+    def test_junk_rekey_fills_rekey_bucket_not_enrol(self):
+        store, _ = helpers.fresh("victim")
+        for _ in range(15):
+            try:
+                identity.enrol_rekey(
+                    store, "victim", keys.export_public_b64(keys.generate()[1]), "junk"
+                )
+            except (errors.EnrolmentFailed, errors.RateLimited):
+                pass
+        # enrol: bucket still usable
+        self.assertTrue(store.allow("enrol:victim"))
+        # Valid ops rekey still works: failures charge rekey: only after bad secret,
+        # and a valid secret bypasses the limiter.
+        secret = identity.seed_bootstrap(store, "victim")
+        identity.enrol_rekey(
+            store, "victim", keys.export_public_b64(keys.generate()[1]), secret
+        )
+
+    def test_failed_rekey_eventually_rate_limited(self):
+        store, _ = helpers.fresh("hammer")
+        hit = False
+        for _ in range(30):
+            try:
+                identity.enrol_rekey(
+                    store, "hammer", keys.export_public_b64(keys.generate()[1]), "nope"
+                )
+            except errors.RateLimited:
+                hit = True
+                break
+            except errors.EnrolmentFailed:
+                pass
+        self.assertTrue(hit, "junk rekey attempts should eventually RateLimited")
+
+
 
 class TestBootstrapRowDeletedOnSpend(unittest.TestCase):
     """R8: spent bootstrap secrets are deleted, not retained as hashes."""
@@ -495,3 +528,47 @@ class TestResolveRechecksAgentState(unittest.TestCase):
         identity.revoke(store, "bigpickle", reason="compromise")
         with self.assertRaises(errors.SessionExpired):
             sessions.resolve(store, sess["token"])
+
+
+class TestHumanSessionFingerprintEpoch(unittest.TestCase):
+    """N7: rekey must kill race-surviving human_via_agent sessions too."""
+
+    def test_resolve_refuses_stale_human_fingerprint_after_rekey(self):
+        store, private = helpers.fresh()
+        old_fp = store.get_agent("bigpickle")["fingerprint"]
+        # Plant a human session as if redeem's put_session won a race against rekey.
+        import time as _t
+        token = "human-race-survivor"
+        store.put_session(
+            sessions.hash_token(token),
+            sessions.HUMAN_VIA_AGENT,
+            "bigpickle's human",
+            "bigpickle",
+            list(sessions.HUMAN_VIEW_SCOPES),
+            int(_t.time()) + 3600,
+            granted_by="bigpickle",
+            fingerprint=old_fp,
+        )
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(store, "bigpickle")
+        identity.enrol_rekey(
+            store, "bigpickle", keys.export_public_b64(new_pub), secret
+        )
+        # Clear revoked_at that rekey set, to isolate the fingerprint check.
+        store.conn.execute(
+            "UPDATE sessions SET revoked_at = NULL, revoked_reason = NULL"
+            " WHERE token_hash = ?",
+            (sessions.hash_token(token),),
+        )
+        store.conn.commit()
+        with self.assertRaises(errors.SessionExpired) as ctx:
+            sessions.resolve(store, token)
+        self.assertIn("key epoch", str(ctx.exception))
+
+    def test_issue_code_return_has_single_view_ttl(self):
+        """N8: issue_code return dict must not duplicate view_ttl."""
+        import inspect
+        from sigil import viewcodes as vc
+        src = inspect.getsource(vc.issue_code)
+        # Count literal "view_ttl" keys in the return dict construction region.
+        self.assertEqual(src.count('"view_ttl":'), 1, src[src.rfind("return"):])

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,75 @@ 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)
+
+
+class TestEnrolAfterPermanentRevoke(unittest.TestCase):
+    """N4: enrol must not burn the seed when the agent row still exists."""
+
+    def test_enrol_after_permanent_revoke_does_not_burn_seed(self):
+        store, private = helpers.fresh()
+        identity.revoke(store, "bigpickle", reason="compromise")
+        identity.reinstate(store, "bigpickle")
+        secret = identity.seed_bootstrap(store, "bigpickle")
+        with self.assertRaises(errors.EnrolmentFailed) as ctx:
+            identity.enrol(
+                store, "bigpickle", keys.export_public_b64(keys.generate()[1]), secret
+            )
+        self.assertIn("enrol_rekey", str(ctx.exception))
+        # Seed still usable for the correct path.
+        identity.enrol_rekey(
+            store, "bigpickle", keys.export_public_b64(keys.generate()[1]), secret
+        )
+
+
+class TestSuspendedSignerCannotCheckpoint(unittest.TestCase):
+    """N5: timed-suspended signer must not checkpoint."""
+
+    def test_timed_suspend_refuses_checkpoint(self):
+        import time
+        store, private = helpers.fresh()
+        identity.revoke(
+            store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
+        )
+        with self.assertRaises(ValueError) as ctx:
+            store.checkpoint(private, "bigpickle")
+        self.assertIn("revoked", str(ctx.exception).lower())