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

# Платёжная нотификация Т-Банка (PR-D2, готовит почву под PR-D3 — путь ещё
# закрыт rbac до того момента). Сервер-к-серверу, без сессии/X-Authenticated-User
# → в общем лимитере попал бы в один и тот же per-IP ключ с любым другим
# анонимным трафиком с той же исходящей сети банка. Мотив НЕ «банк упрётся в
# лимит» — 300/60с и так щедро — а «429 никогда не должен стать причиной, по
# которой денежное состояние разъехалось»: для банка недоставленная нотификация
# = «доставка не удалась», альтернативного канала нет, а очередь ретраев
# растягивается на сутки. Только точный путь notify — НЕ checkout (тот
# инициирует пользователь с сессией/курсором в браузере, абуз там штатно
# лимитируем как любой другой API-путь).
_PAYMENTS_NOTIFY_PATH = "/api/v1/trade-in/payments/notify"


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
        # Платёжная нотификация — мимо ОБЩЕГО (per-user/per-IP shared) лимитера,
        # но НЕ без лимита вовсе: idiom `_notify_limiter` (`SlidingWindowLimiter`,
        # тот же приём, что `support.py:92`/`:319` — узкий per-feature бюджет
        # ВМЕСТО общего, не полное отключение защиты). Порог заведомо выше любого
        # штатного трафика банка (документированное расписание ретраев неизвестно,
        # см. mera-tbank-acquiring-recon.md — берём с кратным запасом), но конечен:
        # полное отключение оставило бы путь без backstop против шторма запросов —
        # подпись отсекает мусор ПОСЛЕ разбора тела (PR-D3), не до.
        if path == _PAYMENTS_NOTIFY_PATH:
            retry_after = _notify_limiter.check(_client_ip(request))
            if retry_after is not None:
                return JSONResponse(
                    status_code=429,
                    content={"detail": "Слишком много запросов. Попробуйте позже."},
                    headers={"Retry-After": str(int(retry_after) + 1)},
                )
            return await call_next(request)
        # Лимитируем только 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) -> int:
        """Регистрирует одну попытку под *key* и возвращает их число в окне ПОСЛЕ неё.

        Счётчик нужен вызывающим, которым мало булева «за лимитом / нет»: login
        (#2571) по нему считает НАСКОЛЬКО перебран порог и растит задержку ответа
        пропорционально. Значение можно игнорировать — `check()` так и делает.
        """
        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]
        return len(bucket)

    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


# Щедрый бюджет для платёжной нотификации (PR-D2): 3000/60с (50 req/s) — на два
# порядка выше любого правдоподобного трафика банка (тест 400/60с проходит с
# запасом в 7.5×), но конечен — backstop против шторма запросов на путь, где
# подпись проверяется уже ПОСЛЕ разбора тела. Ключ — client IP (у сервер-к-
# серверу вызова нет сессии/X-Authenticated-User).
_NOTIFY_RATE_LIMIT = 3000
_NOTIFY_RATE_WINDOW_S = 60.0
_notify_limiter = SlidingWindowLimiter(limit=_NOTIFY_RATE_LIMIT, window_s=_NOTIFY_RATE_WINDOW_S)


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"