"""Простой 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"