Fix R1: invalidate view codes on revoke/rekey/suspend

Outstanding view codes survived revoke and rekey, so a pre-issued code
could mint a fresh human_via_agent session after compromise revoke or
key rotation. Invalidate unused codes when sessions are revoked (covers
revoke, suspend, and rekey), and refuse redeem when the agent is
currently revoked. Regression tests cover all three paths.
This commit is contained in:
leora 2026-10-06 06:10:42 +08:00
parent 73f802a5e6
commit 9eea815805
4 changed files with 86 additions and 4 deletions

View File

@ -131,8 +131,11 @@ when the timer ran out, which is a revocation that silently fails at exactly the
wrong moment. A noisy neighbour or a holiday wants a pause, and re-enrolling for wrong moment. A noisy neighbour or a holiday wants a pause, and re-enrolling for
that would destroy the audit history that says it was deliberate. that would destroy the audit history that says it was deliberate.
Revoking kills live sessions and drops outstanding challenges, so a revocation Revoking kills live sessions, drops outstanding challenges, and invalidates
that races an in-flight authentication cannot be overtaken by it. unused view codes, so a revocation that races an in-flight authentication
cannot be overtaken by it, and a pre-issued code cannot mint a fresh human
session after the agent is suspended or dead. Rekey invalidates codes the same
way: rotating a key must close every door the old key could open.
### Existing controls stay ### Existing controls stay

View File

@ -402,12 +402,30 @@ class Store:
(int(time.time()), token_hash), (int(time.time()), token_hash),
) )
def invalidate_codes_for_agent(self, agent: str) -> int:
"""Mark every unused view code for an agent as spent.
Revoke, suspend, and rekey must close outstanding codes the same way they
kill sessions. Leaving a pre-issued code redeemable after compromise
revoke or key rotation would mint a fresh human_via_agent session while
the agent is supposed to be dead.
"""
with self._write_lock:
cur = self.conn.execute(
"UPDATE codes SET used_at = ? WHERE agent = ? AND used_at IS NULL",
(int(time.time()), agent),
)
self.conn.commit()
return cur.rowcount
def revoke_sessions_for_agent(self, agent: str, reason: str) -> int: def revoke_sessions_for_agent(self, agent: str, reason: str) -> int:
"""Kill every live session for an agent. Used on revoke and rekey. """Kill every live session for an agent. Used on revoke and rekey.
Rotating a key must not leave old sessions usable, otherwise the new key Rotating a key must not leave old sessions usable, otherwise the new key
buys nothing after a compromise. buys nothing after a compromise. Outstanding view codes are invalidated
in the same breath so a pre-issued code cannot mint a fresh session.
""" """
self.invalidate_codes_for_agent(agent)
cur = self._exec( cur = self._exec(
"UPDATE sessions SET revoked_at = ?, revoked_reason = ?" "UPDATE sessions SET revoked_at = ?, revoked_reason = ?"
" WHERE on_behalf_of = ? AND revoked_at IS NULL", " WHERE on_behalf_of = ? AND revoked_at IS NULL",

View File

@ -155,6 +155,22 @@ def redeem_code(store: Store, code: str) -> dict:
agent = row["agent"] agent = row["agent"]
# Defense in depth: revoke/rekey already invalidate unused codes, but refuse
# here too if the agent is currently revoked or missing so a missed
# invalidation cannot mint a human_via_agent session.
from . import identity as _identity
agent_row = store.get_agent(agent)
if agent_row is None or _identity.is_revoked(agent_row):
if not store.allow(f"code_redeem_fail:{agent}"):
raise errors.RateLimited("too many redemption attempts")
store.record(
"view_code.redeem_refused",
actor=agent,
subject=f"{agent}'s human",
detail={"reason": "agent revoked or unknown"},
)
raise errors.CodeInvalid("that code is not redeemable")
# Known-agent attempts use the success quota (code_redeem), not the fail # Known-agent attempts use the success quota (code_redeem), not the fail
# bucket. Failures for this agent (expired / already used) charge a # bucket. Failures for this agent (expired / already used) charge a
# per-agent fail bucket so one noisy human cannot lock every agent out. # per-agent fail bucket so one noisy human cannot lock every agent out.

View File

@ -143,3 +143,48 @@ class TestAuditTrail(Base):
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
class TestRevokeAndRekeyCloseViewCodes(Base):
"""R1: revoke / rekey / suspend must close outstanding view codes."""
def test_redeem_refused_after_permanent_revoke(self):
issued = helpers.issue(self.store, self.private)
identity.revoke(self.store, "bigpickle", reason="compromise")
with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
viewcodes.redeem_code(self.store, issued["code"])
def test_redeem_refused_during_timed_suspension(self):
import time
issued = helpers.issue(self.store, self.private)
identity.revoke(
self.store, "bigpickle", until=int(time.time()) + 3600, reason="pause"
)
with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
viewcodes.redeem_code(self.store, issued["code"])
def test_redeem_refused_after_rekey(self):
issued = helpers.issue(self.store, self.private)
new_private, new_public = keys.generate()
secret = identity.seed_bootstrap(self.store, "bigpickle")
identity.enrol_rekey(
self.store,
"bigpickle",
keys.export_public_b64(new_public),
secret,
)
# Invalidate-on-rekey marks the code used; redeem must not mint a session.
with self.assertRaises((errors.CodeInvalid, errors.CodeUsed)):
viewcodes.redeem_code(self.store, issued["code"])
# New agent can still mint and redeem a fresh code.
fresh = helpers.issue(self.store, new_private)
session = viewcodes.redeem_code(self.store, fresh["code"])
self.assertEqual(session["kind"], "human_via_agent")
def test_unused_codes_marked_used_on_revoke(self):
issued = helpers.issue(self.store, self.private)
digest = viewcodes._hash_code(issued["code"])
row = self.store.get_code(digest)
self.assertIsNone(row["used_at"])
identity.revoke(self.store, "bigpickle")
row = self.store.get_code(digest)
self.assertIsNotNone(row["used_at"])