Ben Stull
9e1b7ce34f
Add RATELIMIT_OTC_REQUEST_MAX / RATELIMIT_VERIFY_MAX / RATELIMIT_CHECK_MAX (default to the existing secure values; non-positive/unparseable → default). Lets a test/PPE stack that drives the auth endpoints repeatedly from one IP lift the budget; production leaves them unset. Mirrors the existing OTC_REQUEST_COOLDOWN_SECONDS knob. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
112 lines
4.6 KiB
Python
112 lines
4.6 KiB
Python
"""In-process per-IP sliding-window rate limiter (security audit 0026, H1).
|
|
|
|
The auth verify endpoints (`/auth/otc/verify`, `/auth/passcode/verify`)
|
|
had no per-IP brake, so an attacker could fan out guesses against a
|
|
target identity bounded only by bcrypt cost. This module is the brake.
|
|
|
|
It is deliberately tiny: §4.2 says the app is a single process with a
|
|
colocated SQLite file, so an in-memory dict of `key -> deque[timestamps]`
|
|
is sufficient and needs no shared store. State resets on restart, which
|
|
fails *open* for a brief window — acceptable because the per-email OTC
|
|
lockout (`otc_verify_state`) and the passcode lockout both persist in the
|
|
database and carry the durable guarantee; this limiter is the
|
|
anti-fan-out layer on top.
|
|
|
|
Chosen over a per-identity lockout *as the primary control* because a
|
|
per-IP window throttles the attacker without letting them grief a victim
|
|
by locking that victim's account (the known downside of identity
|
|
lockouts). Both layers run together.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import threading
|
|
import time
|
|
from collections import defaultdict, deque
|
|
|
|
|
|
def _max_events(env_name: str, default: int) -> int:
|
|
"""Per-limiter budget, overridable via env (e.g. a test/PPE stack that
|
|
drives the auth endpoints repeatedly from one IP). Production leaves these
|
|
unset and gets the secure defaults below. A non-positive / unparseable
|
|
value falls back to the default."""
|
|
raw = os.environ.get(env_name, "").strip()
|
|
if not raw:
|
|
return default
|
|
try:
|
|
n = int(raw)
|
|
except ValueError:
|
|
return default
|
|
return n if n > 0 else default
|
|
|
|
|
|
class SlidingWindowLimiter:
|
|
"""Allow at most `max_events` per `window_seconds` per key.
|
|
|
|
`allow(key)` records an event and returns True if the key is still
|
|
within budget, False if it has exceeded it. Timestamps use a
|
|
monotonic clock so the limiter is immune to wall-clock jumps.
|
|
"""
|
|
|
|
def __init__(self, max_events: int, window_seconds: float) -> None:
|
|
self.max_events = max_events
|
|
self.window_seconds = window_seconds
|
|
self._events: dict[str, deque[float]] = defaultdict(deque)
|
|
self._lock = threading.Lock()
|
|
|
|
def allow(self, key: str) -> bool:
|
|
now = time.monotonic()
|
|
cutoff = now - self.window_seconds
|
|
with self._lock:
|
|
q = self._events[key]
|
|
while q and q[0] < cutoff:
|
|
q.popleft()
|
|
if len(q) >= self.max_events:
|
|
return False
|
|
q.append(now)
|
|
# Opportunistic cleanup so idle keys don't accumulate forever.
|
|
if not q:
|
|
self._events.pop(key, None)
|
|
return True
|
|
|
|
def reset(self, key: str) -> None:
|
|
"""Drop a key's window — e.g. after a successful sign-in so a
|
|
legitimate user who fat-fingered a few times isn't throttled."""
|
|
with self._lock:
|
|
self._events.pop(key, None)
|
|
|
|
|
|
# Module-level limiters shared across requests (one process, so module
|
|
# state is the natural home). Tunables are intentionally generous enough
|
|
# not to bother a human retyping a code, tight enough to kill fan-out:
|
|
# * verify: 10 attempts / 5 min / IP across the auth verify surfaces.
|
|
# * otc request: 5 sends / 5 min / IP (Turnstile is the primary gate;
|
|
# this is defense in depth against a solved-challenge replay loop).
|
|
verify_limiter = SlidingWindowLimiter(
|
|
max_events=_max_events("RATELIMIT_VERIFY_MAX", 10), window_seconds=300)
|
|
otc_request_limiter = SlidingWindowLimiter(
|
|
max_events=_max_events("RATELIMIT_OTC_REQUEST_MAX", 5), window_seconds=300)
|
|
# /auth/passcode/check is an anonymous has-passcode oracle (audit 0026 L3).
|
|
# It's a legitimate Login-flow affordance, so the budget is generous —
|
|
# enough for a human typing emails, tight enough to stop bulk scraping.
|
|
check_limiter = SlidingWindowLimiter(
|
|
max_events=_max_events("RATELIMIT_CHECK_MAX", 30), window_seconds=300)
|
|
|
|
|
|
def _reset_all_for_tests() -> None:
|
|
"""Clear every module-level limiter's window. Test support only — the
|
|
limiters are process-global singletons, so without a per-test reset
|
|
one test's requests bleed into the next and later tests trip the
|
|
budget (429). Not called in production."""
|
|
for lim in (verify_limiter, otc_request_limiter, check_limiter):
|
|
with lim._lock:
|
|
lim._events.clear()
|
|
|
|
|
|
def client_key(request) -> str:
|
|
"""Best-effort client identity for limiting. Behind nginx the app is
|
|
started with `--forwarded-allow-ips 127.0.0.1`, so `request.client.host`
|
|
reflects the real client IP via Uvicorn's ProxyHeaders handling."""
|
|
client = getattr(request, "client", None)
|
|
return client.host if client and client.host else "unknown"
|