aramesh-backend/backend/app/rate_limit.py

"""محدودسازی نرخ درخواست — پنجرهٔ لغزان درون‌حافظه‌ای.

برای ورود ادمین (بر پایه تلاش ناموفق) و فرم‌های عمومی (بر پایه هر ارسال)
استفاده می‌شود. شمارنده‌ها در حافظهٔ همین پروسه‌اند — با ری‌استارت پاک می‌شوند؛
به همین دلیل uvicorn تک‌کارگر نگه داشته شده است.
"""

import time

from fastapi import HTTPException, Request, status


def client_ip(request: Request) -> str:
    """IP واقعی کلاینت — پشت Traefik از X-Forwarded-For خوانده می‌شود
    (uvicorn با --proxy-headers هم client.host را درست می‌کند)."""
    forwarded = request.headers.get("x-forwarded-for")
    if forwarded:
        return forwarded.split(",")[0].strip()
    return request.client.host if request.client else "unknown"


class RateLimiter:
    """پنجرهٔ لغزان بر پایه کلید (عملاً IP) — رویدادها پس از پنجره منقضی می‌شوند."""

    def __init__(self, max_events: int, window_seconds: int, message: str):
        self.max_events = max_events
        self.window_seconds = window_seconds
        self.message = message
        self._events: dict[str, list[float]] = {}

    def _recent(self, key: str, now: float) -> list[float]:
        recent = [t for t in self._events.get(key, []) if now - t < self.window_seconds]
        self._events[key] = recent
        return recent

    def _reject(self, recent: list[float]) -> None:
        now = time.monotonic()
        retry_after = max(1, int(self.window_seconds - (now - recent[0])))
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail=self.message,
            headers={"Retry-After": str(retry_after)},
        )

    def hit(self, request: Request) -> None:
        """یک رویداد ثبت می‌کند؛ اگر سهمیه کلید پر باشد 429 برمی‌گرداند."""
        key = client_ip(request)
        now = time.monotonic()
        recent = self._recent(key, now)
        if len(recent) >= self.max_events:
            self._reject(recent)
        self._events.setdefault(key, []).append(now)

    def clear(self, request: Request) -> None:
        self._events.pop(client_ip(request), None)


# ورود ادمین: ۵ تلاش ناموفق در ۱۵ دقیقه — فقط شکست‌ها شمرده می‌شوند
_failed: dict[str, list[float]] = {}
LOGIN_MAX_FAILED_ATTEMPTS = 5
LOGIN_WINDOW_SECONDS = 15 * 60


def ensure_allowed(request: Request) -> None:
    """اگر این IP سهمیه تلاش ناموفقش پر شده باشد، درخواست با 429 رد می‌شود."""
    key = client_ip(request)
    now = time.monotonic()
    recent = [t for t in _failed.get(key, []) if now - t < LOGIN_WINDOW_SECONDS]
    _failed[key] = recent
    if len(recent) >= LOGIN_MAX_FAILED_ATTEMPTS:
        retry_after = max(1, int(LOGIN_WINDOW_SECONDS - (now - 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)


# فرم تماس: ۵ ارسال در ۱۵ دقیقه برای هر IP
contact_limiter = RateLimiter(
    max_events=5,
    window_seconds=15 * 60,
    message="تعداد پیام‌های ارسالی زیاد است — کمی بعد دوباره امتحان کنید",
)

# خبرنامه: ۳ عضویت در ساعت برای هر IP
newsletter_limiter = RateLimiter(
    max_events=3,
    window_seconds=60 * 60,
    message="تعداد درخواست‌ها زیاد است — کمی بعد دوباره امتحان کنید",
)