import time from collections import defaultdict from fastapi import HTTPException, Request, status _WINDOW_SECONDS = 300 _MAX_ATTEMPTS = 5 _attempts: dict[str, list[float]] = defaultdict(list) def _client_key(request: Request, email: str) -> str: forwarded = request.headers.get("x-forwarded-for", "") ip = forwarded.split(",")[0].strip() if forwarded else (request.client.host if request.client else "unknown") return f"{ip}:{email.strip().lower()}" def check_login_rate_limit(request: Request, email: str) -> None: key = _client_key(request, email) now = time.monotonic() window_start = now - _WINDOW_SECONDS recent = [t for t in _attempts[key] if t >= window_start] _attempts[key] = recent if len(recent) >= _MAX_ATTEMPTS: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Zu viele Anmeldeversuche. Bitte in einigen Minuten erneut versuchen.", ) def record_failed_login(request: Request, email: str) -> None: _attempts[_client_key(request, email)].append(time.monotonic()) def clear_login_attempts(request: Request, email: str) -> None: _attempts.pop(_client_key(request, email), None)