"""محدودسازی تلاش ورود — جلوگیری از حمله brute-force روی /api/auth/login. پیش‌فرض: حداکثر ۵ ورود ناموفق در هر ۱۵ دقیقه برای هر IP؛ ورود موفق شمارنده را پاک می‌کند. شمارنده در حافظه همین پروسه است — با ری‌استارت بک‌اند پاک می‌شود و همین کافی است. """ import time from fastapi import HTTPException, Request, status MAX_FAILED_ATTEMPTS = 5 WINDOW_SECONDS = 15 * 60 _failed: dict[str, list[float]] = {} def client_ip(request: Request) -> str: forwarded = request.headers.get("x-forwarded-for") if forwarded: return forwarded.split(",")[0].strip() return request.client.host if request.client else "unknown" def _prune(key: str, now: float) -> list[float]: recent = [t for t in _failed.get(key, []) if now - t < WINDOW_SECONDS] _failed[key] = recent return recent def ensure_allowed(request: Request) -> None: """اگر این IP سهمیه تلاش ناموفقش پر شده باشد، درخواست با 429 رد می‌شود.""" key = client_ip(request) recent = _prune(key, time.monotonic()) if len(recent) >= MAX_FAILED_ATTEMPTS: retry_after = max(1, int(WINDOW_SECONDS - (time.monotonic() - recent[0]))) raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="تلاش‌های ناموفق زیاد است — بعداً دوباره امتحان کنید", headers={"Retry-After": str(retry_after)}, ) def record_failure(request: Request) -> None: _failed.setdefault(client_ip(request), []).append(time.monotonic()) def clear(request: Request) -> None: """با ورود موفق، تاریخچه تلاش‌های ناموفق IP پاک می‌شود.""" _failed.pop(client_ip(request), None)