davideandClaude Sonnet 5
16802cafb6
POST /auth/login had no rate limiting, no lockout, no delay — a patient distributed attack could brute-force a password against an enumerable username list on a custodial wallet, where a guessed password means withdrawing someone's funds. Add per-username and per-IP throttling with exponential backoff (app/auth/rate_limit.py), keyed on app.state like UserLocks rather than a module global so each app instance gets isolated throttle state. Unknown-user and wrong-password already shared one response path, so no enumeration oracle there. Registration is throttled per-IP too, which also bounds how many accounts one IP can spin up (B-31). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
import time
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass
|
|
class _Bucket:
|
|
failures: int = 0
|
|
locked_until: float = 0.0
|
|
last_failure_at: float = 0.0
|
|
|
|
|
|
class RateLimiter:
|
|
"""In-process failed-attempt throttle with exponential backoff, keyed by an
|
|
arbitrary string (username, IP...). Single-process-only, like UserLocks
|
|
(app/tx/locks.py) — an accepted MVP constraint; a multi-worker deployment
|
|
would need a shared store (Redis) instead (B-33).
|
|
|
|
Brute-forcing a login here isn't a spammy client to be capped at N req/s —
|
|
it's an attempt to withdraw someone else's funds — so failures are
|
|
penalized with a delay that doubles each time past `threshold` free
|
|
attempts, rather than a flat rate cap. `decay_seconds` ages a bucket back
|
|
to zero once failures stop, so a shared/NAT IP isn't punished forever for
|
|
someone else's earlier mistakes.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
threshold: int = 5,
|
|
base_delay: float = 2.0,
|
|
max_delay: float = 300.0,
|
|
decay_seconds: float = 900.0,
|
|
) -> None:
|
|
self._threshold = threshold
|
|
self._base_delay = base_delay
|
|
self._max_delay = max_delay
|
|
self._decay_seconds = decay_seconds
|
|
self._buckets: dict[str, _Bucket] = {}
|
|
|
|
def retry_after(self, key: str) -> float:
|
|
bucket = self._buckets.get(key)
|
|
if bucket is None:
|
|
return 0.0
|
|
remaining = bucket.locked_until - time.monotonic()
|
|
return remaining if remaining > 0 else 0.0
|
|
|
|
def record_failure(self, key: str) -> None:
|
|
now = time.monotonic()
|
|
bucket = self._buckets.setdefault(key, _Bucket())
|
|
if bucket.failures and now - bucket.last_failure_at > self._decay_seconds:
|
|
bucket.failures = 0
|
|
bucket.failures += 1
|
|
bucket.last_failure_at = now
|
|
if bucket.failures >= self._threshold:
|
|
delay = min(self._max_delay, self._base_delay * 2 ** (bucket.failures - self._threshold))
|
|
bucket.locked_until = now + delay
|
|
|
|
def record_success(self, key: str) -> None:
|
|
self._buckets.pop(key, None)
|
|
|
|
|
|
class AuthRateLimiters:
|
|
"""The three throttles B-33 needs, bundled so they can live on `app.state`
|
|
(like `UserLocks`, see app/tx/locks.py) rather than as module globals.
|
|
|
|
A module global would persist for the lifetime of the process — fine in
|
|
production (one app instance), but wrong in the test suite, where every
|
|
test builds its own FastAPI app against a fresh in-memory DB and expects a
|
|
clean slate; a shared global would leak failure counts between unrelated
|
|
tests. Per-`app.state` state gets a fresh instance per app automatically.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.login = RateLimiter(threshold=5, base_delay=2.0, max_delay=300.0)
|
|
self.login_ip = RateLimiter(threshold=20, base_delay=2.0, max_delay=300.0)
|
|
self.register_ip = RateLimiter(threshold=5, base_delay=5.0, max_delay=600.0)
|