PC3 Phase 3.3: rate-limit the auth endpoints

app/rate_limit.py: an in-process sliding-window limiter (single-worker
backend, so no Redis needed -- flagged for if that changes). /register and
/login are capped at 10 attempts / minute / client IP (X-Forwarded-For
aware), returning 429 + Retry-After past that. Blunts credential
brute-force, which the backend had no protection against.

- conftest.py resets the limiter between tests (every TestClient request
  looks like one IP).
- tests/test_rate_limit.py: threshold, shared bucket, per-key window. 77 passed.

CORS reviewed while here: already an explicit origin allowlist
(settings.CORS_ORIGINS), not "*" -- and the Godot client isn't a browser
so it doesn't apply there anyway. Left as-is.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015jtxAiu5dAW4WxCNdGkrkJ

backend/app/rate_limit.py

@@ -0,0 +1,63 @@
+"""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
+
+
+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."""
+    _auth_limiter.check(f"auth:{_client_key(request)}")
+
+
+def reset_all() -> None:
+    """Test hook."""
+    _auth_limiter.reset()

backend/app/routers/auth.py

@@ -16,6 +16,7 @@ from app.database import get_db
 from app.dependencies import CurrentUser
 from app.models.session import RefreshToken
 from app.models.user import User
+from app.rate_limit import auth_rate_limit
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
@@ -153,7 +154,8 @@ def _issue_refresh_token(user_id: int, db: AsyncSession) -> str:
 
 # --- Endpoints ---
 
-@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
+@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED,
+             dependencies=[Depends(auth_rate_limit)])
 async def register(req: RegisterRequest, db: AsyncSession = Depends(get_db)):
     """Register a new user and return access + refresh tokens.
 
@@ -213,7 +215,7 @@ async def register(req: RegisterRequest, db: AsyncSession = Depends(get_db)):
     )
 
 
-@router.post("/login", response_model=TokenResponse)
+@router.post("/login", response_model=TokenResponse, dependencies=[Depends(auth_rate_limit)])
 async def login(form: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
     """Authenticate user and return access + refresh tokens.
 

backend/tests/conftest.py

@@ -132,6 +132,17 @@ def _reset_database():
     yield
 
 
+@pytest.fixture(autouse=True)
+def _reset_rate_limiter():
+    """Every TestClient request looks like the same client IP, so the
+    per-IP auth limiter would trip mid-suite. Clear it between tests; the
+    dedicated rate-limit test disables this to exercise the real behaviour.
+    """
+    from app import rate_limit
+    rate_limit.reset_all()
+    yield
+
+
 @pytest.fixture(autouse=True)
 def _override_db():
     """Override get_db so the app uses a fresh session per test.

backend/tests/test_rate_limit.py

@@ -0,0 +1,51 @@
+"""The auth endpoints are rate-limited per client IP."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app import rate_limit
+from app.main import app
+
+
+@pytest.fixture(autouse=True)
+def _clean_limiter():
+    # This module tests the limiter itself, so start each test from empty
+    # and don't let conftest's between-test reset interfere mid-test.
+    rate_limit.reset_all()
+    yield
+    rate_limit.reset_all()
+
+
+def test_login_is_limited_after_the_threshold():
+    client = TestClient(app)
+    codes = []
+    for i in range(15):
+        r = client.post("/api/auth/login", data={"username": f"nobody{i}", "password": "x"})
+        codes.append(r.status_code)
+    # First 10 get through to the (failing) auth check; the rest are 429.
+    assert codes[:10] == [401] * 10
+    assert 429 in codes[10:]
+    assert codes[-1] == 429
+
+
+def test_register_shares_the_same_bucket():
+    client = TestClient(app)
+    # Burn the budget on login...
+    for i in range(10):
+        client.post("/api/auth/login", data={"username": f"x{i}", "password": "y"})
+    # ...register from the same IP is now blocked too.
+    r = client.post("/api/auth/register", json={
+        "username": "rl_user", "email": "rl@e.com",
+        "password": "secret123", "date_of_birth": "1990-01-01",
+    })
+    assert r.status_code == 429
+    assert "Retry-After" in r.headers
+
+
+def test_limiter_window_is_per_key():
+    limiter = rate_limit.SlidingWindowLimiter(max_hits=2, window_seconds=60)
+    limiter.check("a")
+    limiter.check("a")
+    with pytest.raises(Exception):
+        limiter.check("a")
+    limiter.check("b")  # different key, unaffected