Merge pull request 'Fix R3: permanent revoke clears key; reinstate requires rekey' (#7) from leora/fix-r3-reinstate-requires-rekey into sigil-initial

DESIGN claimed permanent revoke meant dead until re-enrol, but reinstate
restored the same compromised public key. Permanent revoke now wipes
pubkey/fingerprint; reinstate lifts ban flags only. Timed suspend keeps
the key. Compromise recovery may bootstrap+rekey while still revoked,
then reinstate. Regression tests cover reinstate-alone and timed path.

DESIGN.md

@@ -121,9 +121,11 @@ means a future route cannot pass a string where a proof belongs.
 
 ### Suspension is not death
 
-`revoke <agent>` with no end time is permanent: the identity is dead until
-someone re-enrols it. `revoke <agent> --until <unix>` is a pause, and it lapses
-on its own.
+`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.
 
 The distinction matters because the two failure modes are different. A key
 compromise wants death — a lapsing revocation would hand the attacker back in

sigil/identity.py

@@ -56,7 +56,9 @@ def seed_bootstrap(store: Store, agent: str, secret: Optional[str] = None) -> st
     """
     _validate_agent_id(agent)
     row = store.get_agent(agent)
-    if row is not None and is_revoked(row):
+    # Timed suspension: refuse. Permanent revoke (no live key): allow so ops
+    # can stage a rekey before or after reinstate.
+    if row is not None and is_revoked(row) and has_live_key(row):
         raise errors.UnknownAgent(f"{agent} is revoked")
     secret = secret or secrets.token_urlsafe(24)
     store.put_bootstrap(_hash_secret(secret), agent)
@@ -114,6 +116,18 @@ def is_revoked(row, now: Optional[float] = None) -> bool:
     return (now if now is not None else time.time()) < until
 
 
+def has_live_key(row) -> bool:
+    """Does this agent row still hold an enrolled public key?
+
+    Permanent revoke clears pubkey/fingerprint so reinstate alone cannot bring
+    a compromised key back. Callers that mint challenges or sessions must
+    refuse until bootstrap + enrol_rekey restores a real key.
+    """
+    fp = row["fingerprint"] if "fingerprint" in row.keys() else None
+    pubkey = row["pubkey"] if "pubkey" in row.keys() else None
+    return bool(fp) and isinstance(pubkey, (bytes, bytearray)) and len(pubkey) == 32
+
+
 def revoke(store: Store, agent: str, until: Optional[int] = None,
            reason: Optional[str] = None, actor: str = "system") -> int:
     """Revoke an agent, optionally until a time.
@@ -144,7 +158,14 @@ def revoke(store: Store, agent: str, until: Optional[int] = None,
 
 
 def reinstate(store: Store, agent: str, actor: str = "system") -> None:
-    """Lift a revocation. Audited, because it undoes a security decision."""
+    """Lift a revocation. Audited, because it undoes a security decision.
+
+    After a *permanent* revoke the enrolled key was cleared. Reinstate lifts
+    the ban flags only — the agent still cannot authenticate until ops seed a
+    bootstrap secret and call enrol_rekey with a fresh key. Timed suspensions
+    keep the key, so reinstate (or waiting out the window) restores the same
+    identity without rotation.
+    """
     _validate_agent_id(agent)
     if store.get_agent(agent) is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
@@ -163,6 +184,10 @@ def challenge(store: Store, agent: str, ttl: int = DEFAULT_NONCE_TTL,
         raise errors.UnknownAgent(f"{agent} is not enrolled")
     if is_revoked(row):
         raise errors.UnknownAgent(f"{agent} is revoked")
+    if not has_live_key(row):
+        raise errors.UnknownAgent(
+            f"{agent} has no live key; bootstrap and enrol_rekey required"
+        )
     nonce = secrets.token_urlsafe(24)
     expires = int(time.time()) + ttl
     store.put_nonce(nonce, agent, expires)
@@ -221,6 +246,10 @@ def verify(
         raise errors.UnknownAgent(f"{agent} is not enrolled")
     if is_revoked(agent_row):
         raise errors.UnknownAgent(f"{agent} is revoked")
+    if not has_live_key(agent_row):
+        raise errors.UnknownAgent(
+            f"{agent} has no live key; bootstrap and enrol_rekey required"
+        )
 
     public = keys.from_public_bytes(agent_row["pubkey"])
     try:
@@ -259,7 +288,10 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
     row = store.get_agent(agent)
     if row is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
-    if is_revoked(row):
+    # Permanent revoke clears the key. Allow rekey while still revoked so ops
+    # can rotate before reinstate; a revoked agent that still has a live key
+    # (timed suspend) must not rekey as a back door.
+    if is_revoked(row) and has_live_key(row):
         raise errors.UnknownAgent(f"{agent} is revoked")
     owner = store.spend_bootstrap(_hash_secret(secret))
     if owner != agent:

sigil/store.py

@@ -284,14 +284,28 @@ class Store:
                      reason: Optional[str] = None) -> int:
         """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.
+
         Returns how many live sessions were killed.
         """
         with self._write_lock:
-            self.conn.execute(
-                "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
-                " WHERE agent = ?",
-                (int(time.time()), until, reason, agent),
-            )
+            if until is None:
+                # 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 = ?",
+                    (int(time.time()), until, reason, b"", "", agent),
+                )
+            else:
+                self.conn.execute(
+                    "UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
+                    " WHERE agent = ?",
+                    (int(time.time()), until, reason, agent),
+                )
             # Drop outstanding challenges so an in-flight exchange cannot
             # complete after the revocation lands.
             self.conn.execute(

tests/test_production.py

@@ -193,10 +193,19 @@ class TestRevocationSemantics(unittest.TestCase):
     def test_reinstating_works_and_is_audited(self):
         identity.revoke(self.store, "bigpickle")
         identity.reinstate(self.store, "bigpickle")
+        # Permanent revoke cleared the key; reinstate alone is not enough.
+        with self.assertRaises(errors.UnknownAgent):
+            identity.challenge(self.store, "bigpickle")
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(self.store, "bigpickle")
+        identity.enrol_rekey(
+            self.store, "bigpickle", keys.export_public_b64(new_pub), secret
+        )
         identity.challenge(self.store, "bigpickle")
         kinds = [e["kind"] for e in self.store.events(limit=50)]
         self.assertIn("identity.revoked", kinds)
         self.assertIn("identity.reinstated", kinds)
+        self.assertIn("identity.rekeyed", kinds)
 
     def test_revoking_an_unknown_agent_refused(self):
         with self.assertRaises(errors.UnknownAgent):
@@ -230,6 +239,33 @@ class TestRevocationSemantics(unittest.TestCase):
                             helpers.sign(self.private, chal["nonce"]))
 
 
+class TestCompromiseReinstateRequiresRekey(unittest.TestCase):
+    """R3: reinstate after permanent revoke must not restore the same key."""
+
+    def setUp(self):
+        self.store, self.private = helpers.fresh()
+
+    def test_reinstate_alone_cannot_use_old_key(self):
+        old_token_path_chal = identity.challenge(self.store, "bigpickle")
+        self.assertTrue(old_token_path_chal["fingerprint"])
+        identity.revoke(self.store, "bigpickle", reason="key leak")
+        identity.reinstate(self.store, "bigpickle")
+        with self.assertRaises(errors.UnknownAgent):
+            identity.challenge(self.store, "bigpickle")
+        # Old private key must not verify even if a nonce were planted.
+        self.assertFalse(identity.has_live_key(self.store.get_agent("bigpickle")))
+
+    def test_timed_suspend_reinstate_keeps_same_key(self):
+        import time
+        fp = self.store.get_agent("bigpickle")["fingerprint"]
+        identity.revoke(
+            self.store, "bigpickle", until=int(time.time()) + 60, reason="pause"
+        )
+        identity.reinstate(self.store, "bigpickle")
+        chal = identity.challenge(self.store, "bigpickle")
+        self.assertEqual(chal["fingerprint"], fp)
+
+
 class TestViewSessionsCannotBeForged(unittest.TestCase):
     """Finding: human sessions were minted from an agent name.
 
@@ -300,12 +336,14 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
     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
+    def test_rekey_refuses_while_timed_suspended(self):
+        import time
+        # Timed suspend keeps the key — rekey must not be a back door.
+        identity.revoke(
+            self.store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
+        )
         with self.assertRaises(errors.UnknownAgent):
-            secret = identity.seed_bootstrap(self.store, "bigpickle")
-        # Even with a secretly planted bootstrap, rekey must refuse.
+            identity.seed_bootstrap(self.store, "bigpickle")
         planted = secrets_module.token_urlsafe(24)
         self.store.put_bootstrap(
             __import__("hashlib").sha256(planted.encode()).hexdigest(),
@@ -320,6 +358,28 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
             )
         row = self.store.get_agent("bigpickle")
         self.assertTrue(identity.is_revoked(row))
+        self.assertTrue(identity.has_live_key(row))
+
+    def test_permanent_revoke_allows_rekey_but_stays_revoked(self):
+        """Compromise recovery: rotate key while banned, then reinstate."""
+        old_fp = self.store.get_agent("bigpickle")["fingerprint"]
+        identity.revoke(self.store, "bigpickle", reason="compromise")
+        row = self.store.get_agent("bigpickle")
+        self.assertTrue(identity.is_revoked(row))
+        self.assertFalse(identity.has_live_key(row))
+        new_pub = keys.generate()[1]
+        secret = identity.seed_bootstrap(self.store, "bigpickle")
+        new_fp = identity.enrol_rekey(
+            self.store, "bigpickle", keys.export_public_b64(new_pub), secret
+        )
+        row = self.store.get_agent("bigpickle")
+        self.assertTrue(identity.is_revoked(row))
+        self.assertTrue(identity.has_live_key(row))
+        self.assertNotEqual(new_fp, old_fp)
+        with self.assertRaises(errors.UnknownAgent):
+            identity.challenge(self.store, "bigpickle")
+        identity.reinstate(self.store, "bigpickle")
+        identity.challenge(self.store, "bigpickle")
 
     def test_rekey_preserves_revocation_columns(self):
         # Reinstate path: if somehow update ran under a revoked row, columns stay.