Merge pull request 'Fix N3: dedicated rekey rate-limit bucket; charge only on failure' (#14) from leora/fix-n3-rekey-rate-bucket into sigil-initial

enrol_rekey shared the enrol: bucket and metered before auth, so junk
calls could lock out incident-response rekey. Use a separate rekey:
bucket and charge it only after a bad bootstrap secret, so a valid seed
always proceeds.

sigil/identity.py

@@ -285,8 +285,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 +295,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/store.py

@@ -148,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),

tests/test_hardening.py

@@ -381,18 +381,51 @@ 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):
-            identity.enrol_rekey(
-                store, "limited", keys.export_public_b64(keys.generate()[1]), secret
-            )
+        # 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):