daddy
fb52c4fd08
Every checkpoint smoke test registers users against one backend from one client IP, so the per-IP auth limit tripped mid-suite (main_menu_full_flow -> 2/12). Add RATE_LIMIT_ENABLED (default true); run_checkpoint_tests.py sets it false for its backend. test_rate_limit.py forces it on in-fixture so the limiter is still really tested. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015jtxAiu5dAW4WxCNdGkrkJ
74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
"""Lightweight in-process rate limiting.
|
|
|
|
The backend runs single-worker (required by the game-server supervisor), so a
|
|
process-local sliding-window counter is sufficient and needs no Redis. Keyed
|
|
by client IP. Used to blunt credential brute-force on the auth endpoints.
|
|
|
|
If the backend is ever scaled to multiple workers/hosts this must move to a
|
|
shared store (Redis) -- flagged here rather than silently under-enforcing.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import defaultdict, deque
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
|
|
from app.config import settings
|
|
|
|
|
|
class SlidingWindowLimiter:
|
|
def __init__(self, max_hits: int, window_seconds: float) -> None:
|
|
self.max_hits = max_hits
|
|
self.window = window_seconds
|
|
self._hits: dict[str, deque[float]] = defaultdict(deque)
|
|
|
|
def check(self, key: str) -> None:
|
|
now = time.monotonic()
|
|
q = self._hits[key]
|
|
cutoff = now - self.window
|
|
while q and q[0] < cutoff:
|
|
q.popleft()
|
|
if len(q) >= self.max_hits:
|
|
retry = max(1, int(q[0] + self.window - now))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail="Too many attempts. Try again shortly.",
|
|
headers={"Retry-After": str(retry)},
|
|
)
|
|
q.append(now)
|
|
|
|
def reset(self) -> None:
|
|
self._hits.clear()
|
|
|
|
|
|
def _client_key(request: Request) -> str:
|
|
# X-Forwarded-For first hop when behind the reverse proxy, else the socket.
|
|
fwd = request.headers.get("x-forwarded-for", "")
|
|
if fwd:
|
|
return fwd.split(",")[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
# 10 auth attempts / minute / IP -- generous for a human, painful for a
|
|
# brute-forcer.
|
|
_auth_limiter = SlidingWindowLimiter(max_hits=10, window_seconds=60)
|
|
|
|
|
|
async def auth_rate_limit(request: Request) -> None:
|
|
"""FastAPI dependency: rate-limit auth attempts per client IP.
|
|
|
|
A no-op when ``RATE_LIMIT_ENABLED`` is false -- the checkpoint smoke
|
|
suite runs every test against one backend from one client IP, so the
|
|
per-IP cap would trip mid-suite. The dedicated ``test_rate_limit``
|
|
module exercises the real behaviour regardless of this flag.
|
|
"""
|
|
if not settings.RATE_LIMIT_ENABLED:
|
|
return
|
|
_auth_limiter.check(f"auth:{_client_key(request)}")
|
|
|
|
|
|
def reset_all() -> None:
|
|
"""Test hook."""
|
|
_auth_limiter.reset()
|