Fix R1: invalidate view codes on revoke/rekey/suspend #5
@ -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
|
||||
that would destroy the audit history that says it was deliberate.
|
||||
|
||||
Revoking kills live sessions and drops outstanding challenges, so a revocation
|
||||
that races an in-flight authentication cannot be overtaken by it.
|
||||
Revoking kills live sessions, drops outstanding challenges, and invalidates
|
||||
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
|
||||
|
||||
|
||||
@ -402,12 +402,30 @@ class Store:
|
||||
(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:
|
||||
"""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
|
||||
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(
|
||||
"UPDATE sessions SET revoked_at = ?, revoked_reason = ?"
|
||||
" WHERE on_behalf_of = ? AND revoked_at IS NULL",
|
||||
|
||||
@ -155,6 +155,22 @@ def redeem_code(store: Store, code: str) -> dict:
|
||||
|
||||
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
|
||||
# bucket. Failures for this agent (expired / already used) charge a
|
||||
# per-agent fail bucket so one noisy human cannot lock every agent out.
|
||||
|
||||
@ -143,3 +143,48 @@ class TestAuditTrail(Base):
|
||||
|
||||
if __name__ == "__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"])
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user