Review
Login rate limit for HarborLedger marina berth owners
After a spike of credential stuffing against slip-owner accounts, HarborLedger adds a Redis sliding-window limiter on POST /auth/login. Keys by the caller's IP so a single host cannot hammer the password check; ops asked for something that still works when the API sits behind the clubhouse Cloudflare tunnel.
FastAPITier 3securityrate-limitauth
Click a line to flag it, pick one or more labels, then submit. If the change looks correct, approve it.
app/auth/login.py+21-1
| 12 | 12 | from fastapi import APIRouter, Depends, HTTPException, Request | |
| 13 | 13 | from pydantic import BaseModel, EmailStr | |
| 14 | 14 | ||
| 15 | 15 | from app.auth.passwords import verify_password | |
| 16 | 16 | from app.auth.sessions import issue_session | |
| 17 | 17 | from app.db import get_user_by_email | |
| 18 | 18 | from app.redis import redis | |
| 19 | 19 | ||
| 20 | 20 | router = APIRouter(prefix="/auth", tags=["auth"]) | |
| 21 | 21 | ||
| 22 | 22 | class LoginBody(BaseModel): | |
| 23 | 23 | email: EmailStr | |
| 24 | 24 | password: str | |
| 25 | + | ||
| 26 | + | LOGIN_LIMIT = 10 | |
| 27 | + | LOGIN_WINDOW_SEC = 60 | |
| 28 | + | ||
| 29 | + | def client_ip(request: Request) -> str: | |
| 30 | + | # honor tunnel / CDN hop so clubhouse Wi‑Fi isn't one bucket | |
| 31 | + | forwarded = request.headers.get("x-forwarded-for") | |
| 32 | + | if forwarded: | |
| 33 | + | return forwarded.split(",")[0].strip() | |
| 34 | + | return request.client.host if request.client else "unknown" | |
| 35 | + | ||
| 36 | + | async def enforce_login_rate(request: Request) -> None: | |
| 37 | + | ip = client_ip(request) | |
| 38 | + | key = f"login:rl:{ip}" | |
| 39 | + | n = await redis.incr(key) | |
| 40 | + | if n == 1: | |
| 41 | + | await redis.expire(key, LOGIN_WINDOW_SEC) | |
| 42 | + | if n > LOGIN_LIMIT: | |
| 43 | + | raise HTTPException(status_code=429, detail="too many login attempts") | |
| 25 | 44 | ||
| 26 | 45 | @router.post("/login") | |
| 27 | - | async def login(body: LoginBody): | |
| 46 | + | async def login(body: LoginBody, request: Request): | |
| 47 | + | await enforce_login_rate(request) | |
| 28 | 48 | user = await get_user_by_email(body.email) | |
| 29 | 49 | if user is None or not verify_password(body.password, user.password_hash): | |
| 30 | 50 | raise HTTPException(status_code=401, detail="invalid credentials") | |
| 31 | 51 | return {"token": issue_session(user.id)} |