gendesign/tradein-mvp/backend/app/core/ratelimit.py

"""Простой in-memory rate-limiter для публичного API.

Sliding window. Достаточно для одного backend-инстанса (MVP).
Защищает `/api/v1/*` от абуза — estimate/geocode/suggest публичны.

Ключ лимита (#2213):
  - authenticated username (заголовок X-Authenticated-User от Caddy basic_auth) —
    per-user лимит = rate_limit × rate_limit_authenticated_multiplier;
  - для анонимов — client IP, лимит = rate_limit.

Раньше весь authenticated-трафик освобождался целиком (#655). Это было плацебо:
заголовок X-Authenticated-User клиент-контролируем (его ставит Caddy, но на общей
docker-сети запрос мог прийти и мимо Caddy). Теперь лимит применяется всегда, но
per-user порог щедрый — живой пилот его не достигнет.

Допущение по IP (#2213): перед backend ровно ОДИН доверенный прокси (Caddy).
Значит честный клиентский IP = ПОСЛЕДНИЙ (rightmost) элемент X-Forwarded-For,
добавленный самим Caddy. Leftmost-элементы клиент может подделать (X-Forwarded-For:
"1.2.3.4" в исходном запросе), поэтому брать leftmost — дыра (тривиальный обход
per-IP лимита сменой фейкового первого хопа). Если XFF пуст — remote_addr
соединения.
"""

from __future__ import annotations

import time
from collections import defaultdict, deque

from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

from app.core.config import settings


class RateLimitMiddleware(BaseHTTPMiddleware):
    """Sliding-window rate limit на /api/v1/*. Health и статика — без лимита."""

    def __init__(self, app) -> None:  # type: ignore[no-untyped-def]
        super().__init__(app)
        self._hits: dict[str, deque[float]] = defaultdict(deque)

    async def dispatch(self, request: Request, call_next):  # type: ignore[no-untyped-def]
        path = request.url.path
        # Лимитируем только API; health и прочее — пропускаем.
        if not path.startswith("/api/"):
            return await call_next(request)

        # Ключ и порог: per-user для аутентифицированных, per-IP для анонимов.
        username = request.headers.get("x-authenticated-user")
        if username:
            key = f"user:{username}"
            limit = settings.rate_limit * settings.rate_limit_authenticated_multiplier
        else:
            key = f"ip:{_client_ip(request)}"
            limit = settings.rate_limit

        now = time.monotonic()
        bucket = self._hits[key]

        # Выкидываем устаревшие отметки за пределами окна.
        cutoff = now - settings.rate_limit_window_s
        while bucket and bucket[0] < cutoff:
            bucket.popleft()

        if len(bucket) >= limit:
            retry = int(settings.rate_limit_window_s - (now - bucket[0])) + 1
            return JSONResponse(
                status_code=429,
                content={"detail": "Слишком много запросов. Попробуйте позже."},
                headers={"Retry-After": str(retry)},
            )

        bucket.append(now)
        # Лёгкая защита от утечки памяти — чистим пустые корзины изредка.
        if len(self._hits) > 10000:
            for k in [k for k, v in self._hits.items() if not v]:
                del self._hits[k]

        return await call_next(request)


class SlidingWindowLimiter:
    """Reusable in-process sliding-window limiter — тот же алгоритм, что
    `RateLimitMiddleware.dispatch` (deque per key, отбрасываем протухшие метки),
    вынесенный для feature-специфичных лимитов, которые нужны ЖЁСТЧЕ общего
    per-user порога `/api/*` (напр. отправка сообщений в веб-чат поддержки,
    #tgsupport-web — общий лимит 300/60с не спасёт support-топик от заливки
    одним флудящим клиентом, т.к. Telegram Bot API токен общий на всех).

    Не заменяет `RateLimitMiddleware` (тот остаётся общим гейтом на `/api/*`),
    а даёт отдельный, более узкий бюджет для конкретного эндпоинта/действия.
    """

    def __init__(self, limit: int, window_s: float) -> None:
        self._limit = limit
        self._window_s = window_s
        self._hits: dict[str, deque[float]] = defaultdict(deque)

    def _prune(self, bucket: deque[float], now: float) -> None:
        cutoff = now - self._window_s
        while bucket and bucket[0] < cutoff:
            bucket.popleft()

    def retry_after(self, key: str) -> float | None:
        """Non-destructive проверка: сколько секунд ждать, если *key* СЕЙЧАС за
        лимитом, иначе None. НЕ регистрирует попытку — вызывающая сторона решает
        сама, когда звать `record()` (обычно — только на успех действия, #tgsupport-web
        review L3: неудачная попытка не должна съедать бюджет)."""
        now = time.monotonic()
        bucket = self._hits[key]
        self._prune(bucket, now)
        if len(bucket) >= self._limit:
            return self._window_s - (now - bucket[0])
        return None

    def record(self, key: str) -> None:
        """Регистрирует одну успешную попытку под *key*."""
        now = time.monotonic()
        bucket = self._hits[key]
        self._prune(bucket, now)
        bucket.append(now)
        # Лёгкая защита от утечки памяти — чистим пустые корзины изредка (тот же
        # паттерн, что RateLimitMiddleware.dispatch).
        if len(self._hits) > 10000:
            for k in [k for k, v in self._hits.items() if not v]:
                del self._hits[k]

    def check(self, key: str) -> float | None:
        """Комбинированная проверка+регистрация (peek+record за один вызов) —
        для вызывающих, которым не нужно различать "попытка"/"успех" (см.
        `retry_after`/`record` для раздельного варианта)."""
        retry_after = self.retry_after(key)
        if retry_after is not None:
            return retry_after
        self.record(key)
        return None


def _client_ip(request: Request) -> str:
    """Честный клиентский IP при РОВНО ОДНОМ доверенном прокси (Caddy) перед нами.

    Caddy добавляет свой хоп в конец X-Forwarded-For, поэтому берём ПОСЛЕДНИЙ
    (rightmost) элемент — его нельзя подделать с клиента. Leftmost элементы
    клиент-контролируемы и для ключа лимита НЕ используются. Пустой XFF (прямое
    соединение) → remote_addr.
    """
    xff = request.headers.get("x-forwarded-for")
    if xff:
        parts = [p.strip() for p in xff.split(",") if p.strip()]
        if parts:
            return parts[-1]
    return request.client.host if request.client else "unknown"