Fix R3: permanent revoke clears key; reinstate requires rekey #7

Merged
leora merged 1 commits from leora/fix-r3-reinstate-requires-rekey into sigil-initial 2026-10-06 06:13:25 +08:00
4 changed files with 124 additions and 16 deletions
Showing only changes of commit b49d2087d6 - Show all commits

View File

@ -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

View File

@ -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:

View File

@ -284,9 +284,23 @@ 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:
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 = ?",

View File

@ -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.