davideandClaude Opus 5
0aac73e557
The registration throttle called record_failure on every attempt, successful ones included. Five legitimate signups from one shared or NAT address locked the sixth real user out for up to 600s, doubling from there — while an attacker sidestepped the limiter entirely through B-54. Failure backoff is the wrong instrument here: nothing about creating an account is a failed guess at a secret, so the only people it reliably punished were the honest ones. RollingQuota says what was actually meant: 5 accounts per IP per hour, in a rolling window. The caller over it waits exactly until the oldest of the five ages out — an accurate Retry-After, and waiting never makes the next wait longer. It is recorded only once an account exists, so attempts that create nothing (a taken username, a validation error) leave the quota untouched, and checked before the Argon2 hash, so an IP out of quota costs nothing to refuse. Bounded like the failure limiter (B-56): the keys are caller-chosen, so the dict gets both a sweep and a hard cap, evicting keys with room left in their quota before full ones. Also fixes the inline comment that cited B-31 (the resubscribe finding) where it meant B-33. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
141 lines
5.2 KiB
Python
141 lines
5.2 KiB
Python
"""B-56: RateLimiter._buckets is keyed by strings the caller chooses — any
|
|
username, and (before B-54) any IP — and used to only ever grow. decay_seconds
|
|
aged a bucket's *counter* but never removed the entry, so hammering login with
|
|
random usernames was an unbounded memory leak. These tests pin both halves of
|
|
the bound: spent entries are swept, and the dict has a hard cap.
|
|
|
|
The throttling behaviour itself is exercised end-to-end in test_auth.py; what
|
|
matters here is that pruning never hands an attacker a free pass.
|
|
"""
|
|
|
|
import time
|
|
|
|
from app.auth.rate_limit import RateLimiter, RollingQuota
|
|
|
|
|
|
def _limiter(**kwargs) -> RateLimiter:
|
|
# sweep_interval_seconds=0 makes every record_failure sweep, so the tests are
|
|
# deterministic instead of depending on wall-clock timing.
|
|
kwargs.setdefault("sweep_interval_seconds", 0.0)
|
|
return RateLimiter(**kwargs)
|
|
|
|
|
|
def test_spent_buckets_are_swept_on_the_next_failure():
|
|
limiter = _limiter(decay_seconds=0.05)
|
|
for i in range(50):
|
|
limiter.record_failure(f"user:{i}")
|
|
assert len(limiter._buckets) == 50
|
|
|
|
time.sleep(0.06) # every bucket is now past decay_seconds and unlocked
|
|
limiter.record_failure("user:fresh")
|
|
|
|
assert list(limiter._buckets) == ["user:fresh"]
|
|
|
|
|
|
def test_a_bucket_still_locking_someone_out_is_never_swept():
|
|
"""The whole point of the entry: evicting it would reset the backoff and let the
|
|
attacker start over from a free attempt."""
|
|
limiter = _limiter(threshold=1, base_delay=300.0, decay_seconds=0.05)
|
|
limiter.record_failure("user:victim")
|
|
assert limiter.retry_after("user:victim") > 0
|
|
|
|
time.sleep(0.06) # past decay_seconds, but the lockout is still running
|
|
limiter.record_failure("user:someone-else")
|
|
|
|
assert "user:victim" in limiter._buckets
|
|
assert limiter.retry_after("user:victim") > 0
|
|
|
|
|
|
def test_the_dict_is_capped_even_within_one_sweep_interval():
|
|
"""The cap is the backstop for a burst faster than the sweep interval, where
|
|
nothing has had time to expire yet."""
|
|
limiter = _limiter(max_buckets=10, sweep_interval_seconds=3600.0, decay_seconds=3600.0)
|
|
for i in range(200):
|
|
limiter.record_failure(f"ip:{i}")
|
|
|
|
assert len(limiter._buckets) <= 11 # the cap, plus the entry recorded after the last prune
|
|
|
|
|
|
def test_the_cap_evicts_the_entries_closest_to_expiry_first():
|
|
"""What gets dropped under pressure must buy an attacker the least. The deepest
|
|
lockout — the one built up over the most failures, and so the one actually
|
|
holding an attack back — has to be the last thing evicted, not collateral of a
|
|
flood of one-failure keys."""
|
|
limiter = _limiter(max_buckets=3, threshold=1, base_delay=1.0, max_delay=600.0)
|
|
for _ in range(10):
|
|
limiter.record_failure("ip:persistent") # backoff doubles: locked for ~512s
|
|
for i in range(20):
|
|
limiter.record_failure(f"ip:filler{i}") # one failure each: locked for ~1s
|
|
|
|
assert "ip:persistent" in limiter._buckets
|
|
assert limiter.retry_after("ip:persistent") > 100
|
|
|
|
|
|
def test_retry_after_drops_a_spent_bucket_it_looks_at():
|
|
limiter = _limiter(decay_seconds=0.05)
|
|
limiter.record_failure("user:probe")
|
|
|
|
time.sleep(0.06)
|
|
assert limiter.retry_after("user:probe") == 0.0
|
|
assert limiter._buckets == {}
|
|
|
|
|
|
def test_pruning_does_not_reset_a_live_failure_count():
|
|
"""A bucket below the lockout threshold still carries state worth keeping: the
|
|
next failure must count as the second, not the first."""
|
|
limiter = _limiter(threshold=2, base_delay=300.0, decay_seconds=3600.0)
|
|
limiter.record_failure("user:a")
|
|
limiter.record_failure("user:b") # triggers a sweep
|
|
|
|
limiter.record_failure("user:a")
|
|
assert limiter.retry_after("user:a") > 0
|
|
|
|
|
|
# --- B-58: registration is a quota, not failure backoff -------------------------
|
|
|
|
|
|
def test_quota_allows_up_to_the_limit_then_asks_for_a_wait():
|
|
quota = RollingQuota(limit=3, window_seconds=60.0)
|
|
for _ in range(3):
|
|
assert quota.retry_after("ip:1") == 0.0
|
|
quota.record("ip:1")
|
|
|
|
wait = quota.retry_after("ip:1")
|
|
assert 0 < wait <= 60.0
|
|
|
|
|
|
def test_quota_is_per_key():
|
|
quota = RollingQuota(limit=1, window_seconds=60.0)
|
|
quota.record("ip:1")
|
|
|
|
assert quota.retry_after("ip:1") > 0
|
|
assert quota.retry_after("ip:2") == 0.0
|
|
|
|
|
|
def test_quota_frees_a_slot_once_the_oldest_event_leaves_the_window():
|
|
"""The point of a rolling window over failure backoff: the caller waits exactly
|
|
until there's room again, and waiting doesn't make the next wait longer."""
|
|
quota = RollingQuota(limit=2, window_seconds=0.05)
|
|
quota.record("ip:1")
|
|
quota.record("ip:1")
|
|
assert quota.retry_after("ip:1") > 0
|
|
|
|
time.sleep(0.06)
|
|
assert quota.retry_after("ip:1") == 0.0
|
|
|
|
|
|
def test_quota_prunes_spent_keys_and_caps_its_dict():
|
|
"""Same bound as the failure limiter (B-56): caller-chosen keys, so both a sweep
|
|
and a hard cap."""
|
|
quota = RollingQuota(limit=1, window_seconds=0.05, sweep_interval_seconds=0.0)
|
|
for i in range(50):
|
|
quota.record(f"ip:{i}")
|
|
time.sleep(0.06)
|
|
quota.record("ip:fresh")
|
|
assert list(quota._events) == ["ip:fresh"]
|
|
|
|
capped = RollingQuota(limit=1, window_seconds=3600.0, max_keys=10, sweep_interval_seconds=0.0)
|
|
for i in range(200):
|
|
capped.record(f"ip:{i}")
|
|
assert len(capped._events) <= 11
|