Fix N3: dedicated rekey rate-limit bucket; charge only on failure
enrol_rekey shared the enrol: bucket and metered before auth, so junk calls could lock out incident-response rekey. Use a separate rekey: bucket and charge it only after a bad bootstrap secret, so a valid seed always proceeds.
This commit is contained in:
parent
a1cc241923
commit
fae03444af
@ -285,8 +285,6 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
|
|||||||
Rekey does not clear revocation. A revoked agent stays revoked until an
|
Rekey does not clear revocation. A revoked agent stays revoked until an
|
||||||
explicit reinstate; rotating the key must not be a back door back in.
|
explicit reinstate; rotating the key must not be a back door back in.
|
||||||
"""
|
"""
|
||||||
if not store.allow(f"enrol:{agent}"):
|
|
||||||
raise errors.RateLimited(f"too many enrolment attempts for {agent}")
|
|
||||||
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")
|
||||||
@ -297,6 +295,10 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
|
|||||||
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:
|
||||||
|
# Rate-limit only failed attempts so junk cannot burn the ops path, and
|
||||||
|
# use a dedicated rekey: bucket so exhausting enrol: cannot lock rekey.
|
||||||
|
if not store.allow(f"rekey:{agent}"):
|
||||||
|
raise errors.RateLimited(f"too many rekey attempts for {agent}")
|
||||||
raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")
|
raise errors.EnrolmentFailed("bootstrap secret invalid or already spent")
|
||||||
public = keys.import_public_b64(public_b64)
|
public = keys.import_public_b64(public_b64)
|
||||||
fingerprint = keys.fingerprint(public)
|
fingerprint = keys.fingerprint(public)
|
||||||
|
|||||||
@ -148,6 +148,9 @@ RATE_LIMITS = {
|
|||||||
"challenge": (30, 60), # 30 per 60s
|
"challenge": (30, 60), # 30 per 60s
|
||||||
"verify": (60, 60),
|
"verify": (60, 60),
|
||||||
"enrol": (10, 300),
|
"enrol": (10, 300),
|
||||||
|
# Separate from enrol: junk pre-auth enrol_rekey must not lock out ops
|
||||||
|
# rotating a compromised agent during incident response.
|
||||||
|
"rekey": (10, 300),
|
||||||
"code_issue": (20, 300),
|
"code_issue": (20, 300),
|
||||||
"code_redeem": (20, 60),
|
"code_redeem": (20, 60),
|
||||||
"code_redeem_fail": (10, 300),
|
"code_redeem_fail": (10, 300),
|
||||||
|
|||||||
@ -381,19 +381,52 @@ class TestClaimCodeExpiryCAS(unittest.TestCase):
|
|||||||
|
|
||||||
|
|
||||||
class TestEnrolRekeyRateLimit(unittest.TestCase):
|
class TestEnrolRekeyRateLimit(unittest.TestCase):
|
||||||
"""R7: enrol_rekey must share the enrol rate-limit bucket."""
|
"""N3: enrol_rekey uses a dedicated rekey: bucket; junk cannot lock ops."""
|
||||||
|
|
||||||
def test_enrol_rekey_is_rate_limited(self):
|
def test_exhausting_enrol_bucket_does_not_block_rekey(self):
|
||||||
store, _ = helpers.fresh("limited")
|
store, _ = helpers.fresh("limited")
|
||||||
# Exhaust enrol bucket.
|
for _ in range(20):
|
||||||
for i in range(20):
|
|
||||||
store.allow("enrol:limited")
|
store.allow("enrol:limited")
|
||||||
secret = identity.seed_bootstrap(store, "limited")
|
secret = identity.seed_bootstrap(store, "limited")
|
||||||
with self.assertRaises(errors.RateLimited):
|
# Valid rekey must succeed even if enrol: is exhausted.
|
||||||
identity.enrol_rekey(
|
identity.enrol_rekey(
|
||||||
store, "limited", keys.export_public_b64(keys.generate()[1]), secret
|
store, "limited", keys.export_public_b64(keys.generate()[1]), secret
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def test_junk_rekey_fills_rekey_bucket_not_enrol(self):
|
||||||
|
store, _ = helpers.fresh("victim")
|
||||||
|
for _ in range(15):
|
||||||
|
try:
|
||||||
|
identity.enrol_rekey(
|
||||||
|
store, "victim", keys.export_public_b64(keys.generate()[1]), "junk"
|
||||||
|
)
|
||||||
|
except (errors.EnrolmentFailed, errors.RateLimited):
|
||||||
|
pass
|
||||||
|
# enrol: bucket still usable
|
||||||
|
self.assertTrue(store.allow("enrol:victim"))
|
||||||
|
# Valid ops rekey still works: failures charge rekey: only after bad secret,
|
||||||
|
# and a valid secret bypasses the limiter.
|
||||||
|
secret = identity.seed_bootstrap(store, "victim")
|
||||||
|
identity.enrol_rekey(
|
||||||
|
store, "victim", keys.export_public_b64(keys.generate()[1]), secret
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_failed_rekey_eventually_rate_limited(self):
|
||||||
|
store, _ = helpers.fresh("hammer")
|
||||||
|
hit = False
|
||||||
|
for _ in range(30):
|
||||||
|
try:
|
||||||
|
identity.enrol_rekey(
|
||||||
|
store, "hammer", keys.export_public_b64(keys.generate()[1]), "nope"
|
||||||
|
)
|
||||||
|
except errors.RateLimited:
|
||||||
|
hit = True
|
||||||
|
break
|
||||||
|
except errors.EnrolmentFailed:
|
||||||
|
pass
|
||||||
|
self.assertTrue(hit, "junk rekey attempts should eventually RateLimited")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class TestBootstrapRowDeletedOnSpend(unittest.TestCase):
|
class TestBootstrapRowDeletedOnSpend(unittest.TestCase):
|
||||||
"""R8: spent bootstrap secrets are deleted, not retained as hashes."""
|
"""R8: spent bootstrap secrets are deleted, not retained as hashes."""
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user