Redis:
- Singleton ConnectionPool (redis.asyncio), 50 connections — не создаём
новое TCP-соединение на каждый HTTP-запрос
Rate limiter:
- Полностью переписан на async/await
- Lua-скрипт _LUA_CHECK_AND_INCR — атомарная проверка+инкремент без race condition
- Lua-скрипт _LUA_ACQUIRE_CONCURRENT — атомарный захват слота задачи
- Старый паттерн INCR→check→DECR удалён (race condition при конкурентных запросах)
Security:
- get_current_user кэширует пользователя в Redis на 5 минут (TTL)
Раньше: SELECT users на каждый HTTP-запрос
Теперь: Redis GET (кэш) → SELECT users (только при промахе)
- hashed_password НЕ кладётся в кэш
- invalidate_user_cache() для сброса при смене тарифа/пароля
- get_ws_user() для WebSocket через ?token=JWT (браузеры не могут
передавать Authorization header при WS-handshake)
WebSocket:
- Добавлена аутентификация (Depends(get_ws_user))
- Проверка ownership задачи ДО accept() соединения
- Чужой task_id → закрытие с кодом 4004
URL obfuscation:
- Task.public_id = secrets.token_urlsafe(16) = 22 случайных base64url символа
- Клиент работает только с public_id, внутренний UUID не раскрывается
- Все роутеры переключены на public_id в WHERE условиях
- TaskResponse больше не возвращает input_data (там minio_key и т.д.)
- Миграция 002_add_task_public_id.py
MinIO:
- Singleton клиент (не создаём новый на каждый upload)
- ensure_bucket() вызывается один раз при старте (lifespan), не на каждый запрос
- Путь uploads/{doc_uuid}{ext} — user_id убран из пути
CORS:
- Убраны wildcard allow_methods/allow_headers (несовместимы с credentials=True)
- Явный список: methods=[GET,POST,DELETE,OPTIONS], headers=[Authorization,Content-Type,Accept]
- Swagger/OpenAPI доступны только в ENVIRONMENT=development
Documents:
- Content-Length проверяется ДО чтения тела (ранняя отбивка больших файлов)
- Повторная проверка реального размера после чтения (защита от поддельного заголовка)
- Используем get_current_verified_user вместо get_current_user (требуем подтверждённый email)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
169 lines
5.8 KiB
Python
169 lines
5.8 KiB
Python
"""Redis-based rate limiter с атомарными Lua-скриптами.
|
||
|
||
Проблема наивного подхода (GET → проверка → INCR):
|
||
- Race condition: 10 конкурентных запросов могут одновременно пройти GET,
|
||
увидеть значение ниже лимита и все инкрементировать.
|
||
|
||
Решение: один Lua-скрипт выполняется атомарно на стороне Redis.
|
||
Redis гарантирует, что между командами внутри скрипта нет других операций.
|
||
"""
|
||
|
||
import logging
|
||
from datetime import datetime, timezone
|
||
|
||
from app.core.redis_client import get_redis
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ─── Лимиты по тарифам ────────────────────────────────────────────────────────
|
||
|
||
PLAN_LIMITS: dict[str, dict[str, int | None]] = {
|
||
"free": {
|
||
"search_per_day": 10,
|
||
"summarize_per_month": 3,
|
||
"plagiarism_per_month": 1,
|
||
"concurrent": 1,
|
||
},
|
||
"student": {
|
||
"search_per_day": None, # None = безлимит
|
||
"summarize_per_month": 30,
|
||
"plagiarism_per_month": 10,
|
||
"concurrent": 2,
|
||
},
|
||
"premium": {
|
||
"search_per_day": None,
|
||
"summarize_per_month": None,
|
||
"plagiarism_per_month": 50,
|
||
"concurrent": 5,
|
||
},
|
||
"science": {
|
||
"search_per_day": None,
|
||
"summarize_per_month": None,
|
||
"plagiarism_per_month": None,
|
||
"concurrent": 10,
|
||
},
|
||
}
|
||
|
||
ACTION_TO_LIMIT: dict[str, tuple[str, str]] = {
|
||
"search": ("search_per_day", "day"),
|
||
"summarize": ("summarize_per_month", "month"),
|
||
"plagiarism":("plagiarism_per_month", "month"),
|
||
# gost не ограничен — не в маппинге
|
||
}
|
||
|
||
# ─── Lua-скрипты ──────────────────────────────────────────────────────────────
|
||
|
||
# Атомарная проверка + инкремент лимита.
|
||
# Возвращает [current_value, allowed] где allowed = 1 если OK, 0 если превышен.
|
||
_LUA_CHECK_AND_INCR = """
|
||
local key = KEYS[1]
|
||
local limit = tonumber(ARGV[1])
|
||
local ttl = tonumber(ARGV[2])
|
||
|
||
local current = tonumber(redis.call('GET', key) or '0')
|
||
|
||
if current >= limit then
|
||
return {current, 0}
|
||
end
|
||
|
||
local new_val = redis.call('INCR', key)
|
||
-- Устанавливаем TTL только при первом инкременте (когда ключ только что создан)
|
||
if new_val == 1 then
|
||
redis.call('EXPIRE', key, ttl)
|
||
end
|
||
return {new_val, 1}
|
||
"""
|
||
|
||
# Атомарная проверка + инкремент счётчика одновременных задач.
|
||
# Возвращает 1 если слот получен, 0 если превышен лимит.
|
||
_LUA_ACQUIRE_CONCURRENT = """
|
||
local key = KEYS[1]
|
||
local limit = tonumber(ARGV[1])
|
||
local ttl = tonumber(ARGV[2])
|
||
|
||
local current = tonumber(redis.call('GET', key) or '0')
|
||
if current >= limit then
|
||
return 0
|
||
end
|
||
|
||
redis.call('INCR', key)
|
||
redis.call('EXPIRE', key, ttl)
|
||
return 1
|
||
"""
|
||
|
||
|
||
def _period_suffix(period: str) -> str:
|
||
now = datetime.now(timezone.utc)
|
||
return now.strftime("%Y-%m-%d") if period == "day" else now.strftime("%Y-%m")
|
||
|
||
|
||
async def check_and_increment_limit(
|
||
user_id: int,
|
||
action: str,
|
||
plan: str,
|
||
) -> dict:
|
||
"""
|
||
Атомарно проверить лимит и инкрементировать счётчик.
|
||
|
||
Returns:
|
||
{allowed, current, limit, remaining, reset_at}
|
||
"""
|
||
limits = PLAN_LIMITS.get(plan, PLAN_LIMITS["free"])
|
||
|
||
if action not in ACTION_TO_LIMIT:
|
||
return {"allowed": True, "current": 0, "limit": None, "remaining": None}
|
||
|
||
limit_key, period = ACTION_TO_LIMIT[action]
|
||
limit_value = limits.get(limit_key)
|
||
|
||
if limit_value is None: # безлимит
|
||
return {"allowed": True, "current": 0, "limit": None, "remaining": None}
|
||
|
||
period_str = _period_suffix(period)
|
||
redis_key = f"rl:{user_id}:{action}:{period_str}"
|
||
ttl = 86_400 if period == "day" else 86_400 * 32
|
||
|
||
r = get_redis()
|
||
result = await r.eval(_LUA_CHECK_AND_INCR, 1, redis_key, limit_value, ttl)
|
||
current, allowed = int(result[0]), bool(result[1])
|
||
|
||
return {
|
||
"allowed": allowed,
|
||
"current": current,
|
||
"limit": limit_value,
|
||
"remaining": max(0, limit_value - current) if allowed else 0,
|
||
"reset_at": period_str,
|
||
}
|
||
|
||
|
||
async def acquire_concurrent_slot(user_id: int, plan: str) -> bool:
|
||
"""
|
||
Атомарно захватить слот одновременной задачи.
|
||
|
||
Returns:
|
||
True — слот получен (задачу можно создавать).
|
||
False — все слоты заняты.
|
||
"""
|
||
limits = PLAN_LIMITS.get(plan, PLAN_LIMITS["free"])
|
||
max_concurrent = limits.get("concurrent", 1)
|
||
|
||
r = get_redis()
|
||
result = await r.eval(
|
||
_LUA_ACQUIRE_CONCURRENT,
|
||
1,
|
||
f"concurrent:{user_id}",
|
||
max_concurrent,
|
||
3_600, # TTL 1 час — автосброс если воркер упал не освободив слот
|
||
)
|
||
return bool(result)
|
||
|
||
|
||
async def release_concurrent_slot(user_id: int) -> None:
|
||
"""Освободить слот одновременной задачи после завершения."""
|
||
r = get_redis()
|
||
key = f"concurrent:{user_id}"
|
||
# DECR безопасен: Redis не уходит в отрицательные значения если мы контролируем acquire
|
||
current = await r.get(key)
|
||
if current and int(current) > 0:
|
||
await r.decr(key)
|