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.
This commit is contained in:
parent
a474882f61
commit
e993f2403b
@ -121,9 +121,11 @@ means a future route cannot pass a string where a proof belongs.
|
|||||||
|
|
||||||
### Suspension is not death
|
### Suspension is not death
|
||||||
|
|
||||||
`revoke <agent>` with no end time is permanent: the identity is dead until
|
`revoke <agent>` with no end time is permanent: the enrolled public key is
|
||||||
someone re-enrols it. `revoke <agent> --until <unix>` is a pause, and it lapses
|
cleared and the identity stays dead until someone bootstraps a fresh secret and
|
||||||
on its own.
|
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
|
The distinction matters because the two failure modes are different. A key
|
||||||
compromise wants death — a lapsing revocation would hand the attacker back in
|
compromise wants death — a lapsing revocation would hand the attacker back in
|
||||||
|
|||||||
@ -56,7 +56,9 @@ def seed_bootstrap(store: Store, agent: str, secret: Optional[str] = None) -> st
|
|||||||
"""
|
"""
|
||||||
_validate_agent_id(agent)
|
_validate_agent_id(agent)
|
||||||
row = store.get_agent(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")
|
raise errors.UnknownAgent(f"{agent} is revoked")
|
||||||
secret = secret or secrets.token_urlsafe(24)
|
secret = secret or secrets.token_urlsafe(24)
|
||||||
store.put_bootstrap(_hash_secret(secret), agent)
|
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
|
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,
|
def revoke(store: Store, agent: str, until: Optional[int] = None,
|
||||||
reason: Optional[str] = None, actor: str = "system") -> int:
|
reason: Optional[str] = None, actor: str = "system") -> int:
|
||||||
"""Revoke an agent, optionally until a time.
|
"""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:
|
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)
|
_validate_agent_id(agent)
|
||||||
if store.get_agent(agent) is None:
|
if store.get_agent(agent) is None:
|
||||||
raise errors.UnknownAgent(f"{agent} is not enrolled")
|
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")
|
raise errors.UnknownAgent(f"{agent} is not enrolled")
|
||||||
if is_revoked(row):
|
if is_revoked(row):
|
||||||
raise errors.UnknownAgent(f"{agent} is revoked")
|
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)
|
nonce = secrets.token_urlsafe(24)
|
||||||
expires = int(time.time()) + ttl
|
expires = int(time.time()) + ttl
|
||||||
store.put_nonce(nonce, agent, expires)
|
store.put_nonce(nonce, agent, expires)
|
||||||
@ -221,6 +246,10 @@ def verify(
|
|||||||
raise errors.UnknownAgent(f"{agent} is not enrolled")
|
raise errors.UnknownAgent(f"{agent} is not enrolled")
|
||||||
if is_revoked(agent_row):
|
if is_revoked(agent_row):
|
||||||
raise errors.UnknownAgent(f"{agent} is revoked")
|
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"])
|
public = keys.from_public_bytes(agent_row["pubkey"])
|
||||||
try:
|
try:
|
||||||
@ -259,7 +288,10 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
|
|||||||
row = store.get_agent(agent)
|
row = store.get_agent(agent)
|
||||||
if row is None:
|
if row is None:
|
||||||
raise errors.UnknownAgent(f"{agent} is not enrolled")
|
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")
|
raise errors.UnknownAgent(f"{agent} is revoked")
|
||||||
owner = store.spend_bootstrap(_hash_secret(secret))
|
owner = store.spend_bootstrap(_hash_secret(secret))
|
||||||
if owner != agent:
|
if owner != agent:
|
||||||
|
|||||||
@ -284,9 +284,23 @@ class Store:
|
|||||||
reason: Optional[str] = None) -> int:
|
reason: Optional[str] = None) -> int:
|
||||||
"""Suspend an agent. `until` of None is permanent.
|
"""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.
|
Returns how many live sessions were killed.
|
||||||
"""
|
"""
|
||||||
with self._write_lock:
|
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(
|
self.conn.execute(
|
||||||
"UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
|
"UPDATE agents SET revoked_at = ?, revoked_until = ?, revoked_reason = ?"
|
||||||
" WHERE agent = ?",
|
" WHERE agent = ?",
|
||||||
|
|||||||
@ -193,10 +193,19 @@ class TestRevocationSemantics(unittest.TestCase):
|
|||||||
def test_reinstating_works_and_is_audited(self):
|
def test_reinstating_works_and_is_audited(self):
|
||||||
identity.revoke(self.store, "bigpickle")
|
identity.revoke(self.store, "bigpickle")
|
||||||
identity.reinstate(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")
|
identity.challenge(self.store, "bigpickle")
|
||||||
kinds = [e["kind"] for e in self.store.events(limit=50)]
|
kinds = [e["kind"] for e in self.store.events(limit=50)]
|
||||||
self.assertIn("identity.revoked", kinds)
|
self.assertIn("identity.revoked", kinds)
|
||||||
self.assertIn("identity.reinstated", kinds)
|
self.assertIn("identity.reinstated", kinds)
|
||||||
|
self.assertIn("identity.rekeyed", kinds)
|
||||||
|
|
||||||
def test_revoking_an_unknown_agent_refused(self):
|
def test_revoking_an_unknown_agent_refused(self):
|
||||||
with self.assertRaises(errors.UnknownAgent):
|
with self.assertRaises(errors.UnknownAgent):
|
||||||
@ -230,6 +239,33 @@ class TestRevocationSemantics(unittest.TestCase):
|
|||||||
helpers.sign(self.private, chal["nonce"]))
|
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):
|
class TestViewSessionsCannotBeForged(unittest.TestCase):
|
||||||
"""Finding: human sessions were minted from an agent name.
|
"""Finding: human sessions were minted from an agent name.
|
||||||
|
|
||||||
@ -300,12 +336,14 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
|
|||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.store, self.private = helpers.fresh()
|
self.store, self.private = helpers.fresh()
|
||||||
|
|
||||||
def test_rekey_refuses_while_revoked(self):
|
def test_rekey_refuses_while_timed_suspended(self):
|
||||||
identity.revoke(self.store, "bigpickle", reason="incident")
|
import time
|
||||||
secret = None
|
# 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):
|
with self.assertRaises(errors.UnknownAgent):
|
||||||
secret = identity.seed_bootstrap(self.store, "bigpickle")
|
identity.seed_bootstrap(self.store, "bigpickle")
|
||||||
# Even with a secretly planted bootstrap, rekey must refuse.
|
|
||||||
planted = secrets_module.token_urlsafe(24)
|
planted = secrets_module.token_urlsafe(24)
|
||||||
self.store.put_bootstrap(
|
self.store.put_bootstrap(
|
||||||
__import__("hashlib").sha256(planted.encode()).hexdigest(),
|
__import__("hashlib").sha256(planted.encode()).hexdigest(),
|
||||||
@ -320,6 +358,28 @@ class TestRekeyDoesNotUnrevoke(unittest.TestCase):
|
|||||||
)
|
)
|
||||||
row = self.store.get_agent("bigpickle")
|
row = self.store.get_agent("bigpickle")
|
||||||
self.assertTrue(identity.is_revoked(row))
|
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):
|
def test_rekey_preserves_revocation_columns(self):
|
||||||
# Reinstate path: if somehow update ran under a revoked row, columns stay.
|
# Reinstate path: if somehow update ran under a revoked row, columns stay.
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user