From b49d2087d606ba7ab8cb823a63913f78fcddda29 Mon Sep 17 00:00:00 2001 From: leora Date: Tue, 6 Oct 2026 06:13:23 +0800 Subject: [PATCH] Fix R3: permanent revoke clears key; reinstate requires rekey 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 | 8 +++-- sigil/identity.py | 38 ++++++++++++++++++++-- sigil/store.py | 24 +++++++++++--- tests/test_production.py | 70 +++++++++++++++++++++++++++++++++++++--- 4 files changed, 124 insertions(+), 16 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index 366d9e3..b5ff907 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -121,9 +121,11 @@ means a future route cannot pass a string where a proof belongs. ### Suspension is not death -`revoke ` with no end time is permanent: the identity is dead until -someone re-enrols it. `revoke --until ` is a pause, and it lapses -on its own. +`revoke ` 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 --until +` 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 diff --git a/sigil/identity.py b/sigil/identity.py index fc588c7..9257983 100644 --- a/sigil/identity.py +++ b/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: diff --git a/sigil/store.py b/sigil/store.py index e0aeab8..6813795 100644 --- a/sigil/store.py +++ b/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( diff --git a/tests/test_production.py b/tests/test_production.py index 2108715..8e40ad8 100644 --- a/tests/test_production.py +++ b/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. -- 2.47.3