Close three review blockers: rekey, dual-enrol race, redeem quotas

- put_agent is INSERT OR IGNORE (enrol create-if-absent); update_agent_key
  preserves revocation columns so rekey cannot un-revoke
- enrol_rekey and seed_bootstrap refuse while the agent is revoked
- redeem_code uses per-agent code_redeem / code_redeem_fail; anonymous
  guesses no longer lock valid codes for other agents
- remove dead duplicate revoke_agent; tests for each finding (111 total)

Do not merge to main yet: independent crypto review, off-box chain head,
sealed store, and platform glue still open.

DESIGN.md

@@ -222,6 +222,8 @@ sessions mintable from a bare agent name, no revocation operation at all, an
 unkeyed chain that a database-level attacker could rebuild, and SQLite without
 a busy timeout.
 
+Leora's contracted prod pass (Quarry 18a5d762) closed three remaining review blockers on this branch: rekey no longer un-revokes, dual-bootstrap enrol is exactly-one under race, and redeem rate limits are per-agent (anonymous fails isolated from success). Independent review, off-box checkpoint anchor, sealed store, and platform glue remain.
+
 **Still not production ready**, and the reason is not a to-do list:
 
 - **No independent review.** Everything above is self-review, twice over.

README.md

@@ -53,7 +53,7 @@ python3 sigil.py sweep                                # prune stale rows
 python3 -m unittest discover -s tests -v
 ```
 
-106 tests. They cover the refusals, not just the happy paths: replayed
+111 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` and `test_production.py` have one test
 per defect found in review, so any of them regressing brings that defect back.
@@ -119,6 +119,9 @@ A hardening pass closed the defects below. Each is now covered by a test.
 | Rekey left old sessions alive | Rekey revokes every session for that agent |
 | Agent ids unvalidated, could forge log lines | Constrained alphabet and length |
 | 1-hour session anchored to issue time | Anchored to redemption |
+| Rekey cleared `revoked_at` (INSERT OR REPLACE) | Rekey refuses while revoked; `update_agent_key` preserves ban columns |
+| Dual-bootstrap enrol race (last writer won) | `put_agent` is INSERT OR IGNORE; exactly one enrol wins |
+| Redeem fail limit was one global bucket | Per-agent `code_redeem` / `code_redeem_fail`; anonymous fails isolated |
 
 Still open, and these are the reasons not to ship it yet:
 

sigil/identity.py

@@ -50,8 +50,14 @@ def seed_bootstrap(store: Store, agent: str, secret: Optional[str] = None) -> st
 
     This is the only moment a shared secret exists. In production this replaces
     the agent's static passcode, issued once and never transmitted again.
+
+    Refuses when the agent is currently revoked: a fresh bootstrap must not be
+    a path back in without an explicit reinstate.
     """
     _validate_agent_id(agent)
+    row = store.get_agent(agent)
+    if row is not None and is_revoked(row):
+        raise errors.UnknownAgent(f"{agent} is revoked")
     secret = secret or secrets.token_urlsafe(24)
     store.put_bootstrap(_hash_secret(secret), agent)
     store.record(
@@ -70,15 +76,18 @@ def enrol(
     _validate_agent_id(agent)
     if not store.allow(f"enrol:{agent}"):
         raise errors.RateLimited(f"too many enrolment attempts for {agent}")
-    if store.get_agent(agent) is not None:
+    # Spend the secret first so a burned secret cannot be retried. The agent
-        raise errors.EnrolmentFailed(f"{agent} is already enrolled")
+    # 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
+    # instead of silently replacing the winner's key.
     owner = store.spend_bootstrap(_hash_secret(secret))
     if owner != agent:
         raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")
 
     public = keys.import_public_b64(public_b64)
     fingerprint = keys.fingerprint(public)
-    store.put_agent(agent, keys.public_bytes(public), fingerprint)
+    if not store.put_agent(agent, keys.public_bytes(public), fingerprint):
+        raise errors.EnrolmentFailed(f"{agent} is already enrolled")
     store.record(
         "identity.enrolled",
         actor=agent,
@@ -241,17 +250,23 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
 
     Rotation cannot use the agent's existing key to authorise itself: if that
     key leaked, it would authorise its own replacement.
+
+    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.
     """
     row = store.get_agent(agent)
     if row is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
+    if is_revoked(row):
+        raise errors.UnknownAgent(f"{agent} is revoked")
     owner = store.spend_bootstrap(_hash_secret(secret))
     if owner != agent:
         raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")
     public = keys.import_public_b64(public_b64)
     fingerprint = keys.fingerprint(public)
     previous = row["fingerprint"]
-    store.put_agent(agent, keys.public_bytes(public), fingerprint)
+    if not store.update_agent_key(agent, keys.public_bytes(public), fingerprint):
+        raise errors.UnknownAgent(f"{agent} is not enrolled")
     # A new key must not leave old sessions alive, or rotating after a
     # compromise buys nothing.
     killed = store.revoke_sessions_for_agent(agent, "key rotated")

sigil/store.py

@@ -237,22 +237,41 @@ class Store:
             return self.conn.execute(sql, tuple(args)).fetchall()
 
     # -- agents ------------------------------------------------------------
-    def put_agent(self, agent: str, pubkey: bytes, fingerprint: str) -> None:
+    def put_agent(self, agent: str, pubkey: bytes, fingerprint: str) -> bool:
-        self._exec(
+        """Create an agent row if absent. Returns True on insert, False if already enrolled.
-            "INSERT OR REPLACE INTO agents (agent, pubkey, fingerprint, enrolled_at)"
+
+        INSERT OR IGNORE so two concurrent enrols cannot both win: compare-and-set on
+        the primary key. Callers that need to replace a key must use update_agent_key,
+        which preserves revocation columns.
+        """
+        with self._write_lock:
+            cur = self.conn.execute(
+                "INSERT OR IGNORE INTO agents"
+                " (agent, pubkey, fingerprint, enrolled_at)"
                 " VALUES (?, ?, ?, ?)",
                 (agent, pubkey, fingerprint, int(time.time())),
             )
+            self.conn.commit()
+            return cur.rowcount == 1
+
+    def update_agent_key(self, agent: str, pubkey: bytes, fingerprint: str) -> bool:
+        """Replace pubkey/fingerprint without touching revocation columns.
+
+        INSERT OR REPLACE would NULL out revoked_at / revoked_until / revoked_reason
+        because those columns are omitted from the insert list — that is how rekey
+        used to un-revoke. An UPDATE of only the key columns keeps the ban intact.
+        """
+        with self._write_lock:
+            cur = self.conn.execute(
+                "UPDATE agents SET pubkey = ?, fingerprint = ? WHERE agent = ?",
+                (pubkey, fingerprint, agent),
+            )
+            self.conn.commit()
+            return cur.rowcount == 1
 
     def get_agent(self, agent: str) -> Optional[sqlite3.Row]:
         return self._one("SELECT * FROM agents WHERE agent = ?", (agent,))
 
-    def revoke_agent(self, agent: str) -> None:
-        self._exec(
-            "UPDATE agents SET revoked_at = ? WHERE agent = ?",
-            (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.

sigil/viewcodes.py

@@ -140,14 +140,14 @@ def issue_code(
 
 def redeem_code(store: Store, code: str) -> dict:
     """Trade a one-time code for a human_via_agent session."""
-    # Rate-limit before the lookup so a guessing attacker cannot use response
-    # differences to probe for valid codes for free.
-    if not store.allow("code_redeem_fail"):
-        raise errors.RateLimited("too many redemption attempts")
-
     digest = _hash_code(normalise(code))
     row = store.get_code(digest)
     if row is None:
+        # Unknown guesses share one anonymous fail bucket so probing still
+        # saturates — but that bucket is not the same as any agent's success
+        # path, so ten bad guesses cannot lock out a valid code for someone else.
+        if not store.allow("code_redeem_fail:unknown"):
+            raise errors.RateLimited("too many redemption attempts")
         store.record(
             "view_code.redeem_refused", actor="unknown", detail={"reason": "unrecognised"}
         )
@@ -155,7 +155,12 @@ def redeem_code(store: Store, code: str) -> dict:
 
     agent = row["agent"]
 
+    # Known-agent attempts use the success quota (code_redeem), not the fail
+    # bucket. Failures for this agent (expired / already used) charge a
+    # per-agent fail bucket so one noisy human cannot lock every agent out.
     if time.time() >= row["redeem_expires_at"]:
+        if not store.allow(f"code_redeem_fail:{agent}"):
+            raise errors.RateLimited("too many redemption attempts")
         store.record(
             "view_code.redeem_refused",
             actor=agent,
@@ -164,10 +169,16 @@ def redeem_code(store: Store, code: str) -> dict:
         )
         raise errors.CodeExpired("that code has expired")
 
+    if not store.allow(f"code_redeem:{agent}"):
+        raise errors.RateLimited("too many redemption attempts")
+
     # Claim the code before issuing anything. Compare-and-set, so a code cannot
     # mint two sessions under a race, and an expired or replayed code is refused
     # before any session exists.
     if not store.claim_code(digest):
+        # Already-used is a fail for this agent; refund the success slot by
+        # charging the fail bucket instead so retries do not starve real redeems.
+        store.allow(f"code_redeem_fail:{agent}")
         store.record(
             "view_code.redeem_refused",
             actor=agent,

tests/test_production.py

@@ -5,6 +5,7 @@ defect back rather than quietly passing.
 """
 
 import base64
+import secrets as secrets_module
 import os
 import tempfile
 import time
@@ -292,3 +293,125 @@ class TestDatabaseRobustness(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+class TestRekeyDoesNotUnrevoke(unittest.TestCase):
+    """Finding: enrol_rekey used INSERT OR REPLACE and cleared revoked_at."""
+
+    def setUp(self):
+        self.store, self.private = helpers.fresh()
+
+    def test_rekey_refuses_while_revoked(self):
+        identity.revoke(self.store, "bigpickle", reason="incident")
+        secret = None
+        with self.assertRaises(errors.UnknownAgent):
+            secret = identity.seed_bootstrap(self.store, "bigpickle")
+        # Even with a secretly planted bootstrap, rekey must refuse.
+        planted = secrets_module.token_urlsafe(24)
+        self.store.put_bootstrap(
+            __import__("hashlib").sha256(planted.encode()).hexdigest(),
+            "bigpickle",
+        )
+        with self.assertRaises(errors.UnknownAgent):
+            identity.enrol_rekey(
+                self.store,
+                "bigpickle",
+                keys.export_public_b64(keys.generate()[1]),
+                planted,
+            )
+        row = self.store.get_agent("bigpickle")
+        self.assertTrue(identity.is_revoked(row))
+
+    def test_rekey_preserves_revocation_columns(self):
+        # Reinstate path: if somehow update ran under a revoked row, columns stay.
+        identity.revoke(self.store, "bigpickle", reason="incident")
+        row = self.store.get_agent("bigpickle")
+        self.assertIsNotNone(row["revoked_at"])
+        new_pub = keys.generate()[1]
+        self.store.update_agent_key(
+            "bigpickle",
+            keys.public_bytes(new_pub),
+            keys.fingerprint(new_pub),
+        )
+        row2 = self.store.get_agent("bigpickle")
+        self.assertEqual(row2["revoked_at"], row["revoked_at"])
+        self.assertEqual(row2["revoked_reason"], "incident")
+        self.assertTrue(identity.is_revoked(row2))
+
+
+class TestDualBootstrapEnrolRace(unittest.TestCase):
+    """Finding: two secrets for one unenrolled agent could both enrol."""
+
+    def test_exactly_one_enrol_wins_under_race(self):
+        import threading
+
+        store = Store(":memory:")
+        secrets = [
+            identity.seed_bootstrap(store, "racer"),
+            identity.seed_bootstrap(store, "racer"),
+        ]
+        pubs = [keys.generate()[1] for _ in range(2)]
+        results = []
+        lock = threading.Lock()
+
+        def attempt(i):
+            try:
+                fp = identity.enrol(
+                    store,
+                    "racer",
+                    keys.export_public_b64(pubs[i]),
+                    secrets[i],
+                )
+                with lock:
+                    results.append(("ok", fp, keys.fingerprint(pubs[i])))
+            except errors.EnrolmentFailed as e:
+                with lock:
+                    results.append(("fail", str(e)))
+
+        threads = [threading.Thread(target=attempt, args=(i,)) for i in range(2)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        wins = [r for r in results if r[0] == "ok"]
+        fails = [r for r in results if r[0] == "fail"]
+        self.assertEqual(len(wins), 1, results)
+        self.assertEqual(len(fails), 1, results)
+        row = store.get_agent("racer")
+        self.assertEqual(row["fingerprint"], wins[0][2])
+
+
+class TestRedeemRateLimitIsolation(unittest.TestCase):
+    """Finding: one global code_redeem_fail bucket locked every agent."""
+
+    def test_failed_guesses_do_not_lock_other_agents_valid_code(self):
+        store_a, private_a = helpers.fresh("alice")
+        # Exhaust anonymous fail bucket
+        hit_limit = False
+        for i in range(50):
+            try:
+                viewcodes.redeem_code(store_a, f"ZZZZ-{i:04d}")
+            except errors.RateLimited:
+                hit_limit = True
+                break
+            except errors.SigilError:
+                pass
+        self.assertTrue(hit_limit, "anonymous guessing should still rate-limit")
+
+        # A valid code for alice must still redeem — success uses code_redeem:alice
+        issued = helpers.issue(store_a, private_a, agent="alice", limit=5000, window=60)
+        view = viewcodes.redeem_code(store_a, issued["code"])
+        self.assertEqual(view["on_behalf_of"], "alice")
+
+    def test_successful_redeems_do_not_consume_fail_bucket(self):
+        store, private = helpers.fresh()
+        for _ in range(5):
+            issued = helpers.issue(store, private, limit=5000, window=60)
+            viewcodes.redeem_code(store, issued["code"])
+        # Fail bucket should still accept unknown guesses (not exhausted by successes)
+        try:
+            viewcodes.redeem_code(store, "AAAA-AAAA")
+        except errors.RateLimited:
+            self.fail("successes must not share the fail bucket")
+        except errors.SigilError:
+            pass