From c1bfb5f40ec0ca5af00beaaffc884ee8fe2626cf Mon Sep 17 00:00:00 2001 From: jze9 Date: Sun, 24 May 2026 19:51:51 +0500 Subject: [PATCH] fix(api): security, caching, atomic rate limits, url obfuscation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../versions/002_add_task_public_id.py | 45 ++++ services/api/app/api/documents.py | 123 ++++++----- services/api/app/api/search.py | 52 ++--- services/api/app/api/tasks.py | 50 ++--- services/api/app/config.py | 1 + services/api/app/core/minio_client.py | 48 +++++ services/api/app/core/rate_limiter.py | 193 +++++++++--------- services/api/app/core/redis_client.py | 36 ++++ services/api/app/core/security.py | 131 +++++++----- services/api/app/main.py | 118 ++++++----- services/api/app/schemas/tasks.py | 11 +- services/api/requirements.txt | 2 +- 12 files changed, 493 insertions(+), 317 deletions(-) create mode 100644 services/api/alembic/versions/002_add_task_public_id.py create mode 100644 services/api/app/core/minio_client.py create mode 100644 services/api/app/core/redis_client.py diff --git a/services/api/alembic/versions/002_add_task_public_id.py b/services/api/alembic/versions/002_add_task_public_id.py new file mode 100644 index 0000000..8c927cc --- /dev/null +++ b/services/api/alembic/versions/002_add_task_public_id.py @@ -0,0 +1,45 @@ +"""Добавить public_id в таблицу tasks. + +Revision ID: 002 +Revises: 001 +Create Date: 2026-05-24 +""" + +import secrets + +from alembic import op +import sqlalchemy as sa +from sqlalchemy import text + +revision = "002" +down_revision = "001" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # Добавляем колонку как nullable — сначала заполним данными, потом сделаем NOT NULL + op.add_column( + "tasks", + sa.Column("public_id", sa.String(32), nullable=True), + ) + + # Заполняем существующие строки уникальными public_id + conn = op.get_bind() + tasks = conn.execute(text("SELECT id FROM tasks")).fetchall() + for (task_id,) in tasks: + conn.execute( + text("UPDATE tasks SET public_id = :pid WHERE id = :id"), + {"pid": secrets.token_urlsafe(16), "id": task_id}, + ) + + # Делаем NOT NULL и добавляем уникальный индекс + op.alter_column("tasks", "public_id", nullable=False) + op.create_unique_constraint("uq_tasks_public_id", "tasks", ["public_id"]) + op.create_index("ix_tasks_public_id", "tasks", ["public_id"], unique=True) + + +def downgrade() -> None: + op.drop_index("ix_tasks_public_id", table_name="tasks") + op.drop_constraint("uq_tasks_public_id", "tasks", type_="unique") + op.drop_column("tasks", "public_id") diff --git a/services/api/app/api/documents.py b/services/api/app/api/documents.py index 00344d1..99b6cf1 100644 --- a/services/api/app/api/documents.py +++ b/services/api/app/api/documents.py @@ -1,18 +1,18 @@ -"""Роутер загрузки документов на проверку плагиата.""" +"""Загрузка документов на проверку плагиата.""" +import io import logging import uuid from pathlib import Path -from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status -from minio import Minio -from minio.error import S3Error +from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, status from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.core.celery_app import celery_app -from app.core.rate_limiter import check_and_increment_limit, check_concurrent_limit -from app.core.security import get_current_user +from app.core.minio_client import get_minio +from app.core.rate_limiter import acquire_concurrent_slot, check_and_increment_limit +from app.core.security import get_current_verified_user from app.database import get_db from app.models.task import Task from app.models.user import User @@ -22,102 +22,104 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/documents", tags=["documents"]) -# Допустимые форматы ALLOWED_EXTENSIONS = {".pdf", ".docx", ".txt"} -MAX_FILE_SIZE_MB = 100 -MAX_FILE_SIZE_BYTES = MAX_FILE_SIZE_MB * 1024 * 1024 +MAX_FILE_SIZE_BYTES = 100 * 1024 * 1024 # 100 МБ - -def get_minio_client() -> Minio: - """Создать MinIO клиент.""" - return Minio( - settings.MINIO_ENDPOINT, - access_key=settings.MINIO_ACCESS_KEY, - secret_key=settings.MINIO_SECRET_KEY, - secure=False, - ) +CONTENT_TYPE_MAP = { + ".pdf": "application/pdf", + ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ".txt": "text/plain", +} @router.post("/check", response_model=TaskResponse, status_code=status.HTTP_202_ACCEPTED) async def upload_for_plagiarism_check( - file: UploadFile = File(..., description="PDF, DOCX или TXT файл до 100 МБ"), - current_user: User = Depends(get_current_user), + request: Request, + file: UploadFile, + current_user: User = Depends(get_current_verified_user), db: AsyncSession = Depends(get_db), ) -> TaskResponse: """ - Загрузить документ для проверки на плагиат. + Загрузить документ на проверку плагиата. - Файл сохраняется в MinIO, затем диспатчится задача index.extract_and_check. - Результат доступен через GET /tasks/{task_id}. + Файл сохраняется в MinIO. В пути используется UUID документа, + не ID пользователя (пользователь не должен знать свой internal ID). """ - # Проверить расширение файла + # ── 1. Проверяем заголовки ДО чтения тела ───────────────────────────────── + if not file.filename: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Имя файла не указано", - ) + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Имя файла не указано") ext = Path(file.filename).suffix.lower() if ext not in ALLOWED_EXTENSIONS: raise HTTPException( status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, - detail=f"Неподдерживаемый формат файла. Допустимые форматы: {', '.join(ALLOWED_EXTENSIONS)}", + detail=f"Неподдерживаемый формат. Допустимые: {', '.join(ALLOWED_EXTENSIONS)}", ) - # Проверить лимит по тарифу - limit_check = check_and_increment_limit(current_user.id, "plagiarism", current_user.plan) - if not limit_check["allowed"]: + # Content-Length позволяет отклонить слишком большой файл до его чтения в память. + # Это не 100% защита (заголовок можно не передать), поэтому проверяем ещё раз после чтения. + content_length = request.headers.get("content-length") + if content_length and int(content_length) > MAX_FILE_SIZE_BYTES: + raise HTTPException( + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + detail=f"Файл слишком большой (максимум {MAX_FILE_SIZE_BYTES // 1024 // 1024} МБ)", + ) + + # ── 2. Rate limits ───────────────────────────────────────────────────────── + + limit_result = await check_and_increment_limit(current_user.id, "plagiarism", current_user.plan) + if not limit_result["allowed"]: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail=( - f"Превышен месячный лимит проверок плагиата для тарифа '{current_user.plan}'. " - f"Использовано {limit_check['current']} из {limit_check['limit']}." + f"Превышен месячный лимит проверок (тариф '{current_user.plan}'): " + f"{limit_result['current']}/{limit_result['limit']}. " + f"Сбросится {limit_result['reset_at']}." ), + headers={"Retry-After": "86400"}, ) - # Проверить лимит одновременных задач - if not check_concurrent_limit(current_user.id, current_user.plan): + slot_acquired = await acquire_concurrent_slot(current_user.id, current_user.plan) + if not slot_acquired: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Превышен лимит одновременных задач.", ) - # Прочитать файл - file_data = await file.read() + # ── 3. Читаем файл и проверяем реальный размер ──────────────────────────── + file_data = await file.read() if len(file_data) > MAX_FILE_SIZE_BYTES: raise HTTPException( status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, - detail=f"Файл слишком большой. Максимальный размер: {MAX_FILE_SIZE_MB} МБ", + detail=f"Файл слишком большой (максимум {MAX_FILE_SIZE_BYTES // 1024 // 1024} МБ)", ) - # Загрузить в MinIO - minio_key = f"uploads/{current_user.id}/{uuid.uuid4()}{ext}" - minio_client = get_minio_client() + # ── 4. Сохраняем в MinIO ─────────────────────────────────────────────────── + # В пути — UUID документа, НЕ user_id. Так не раскрываем внутренний ID пользователя. + + doc_uuid = str(uuid.uuid4()) + minio_key = f"uploads/{doc_uuid}{ext}" + content_type = CONTENT_TYPE_MAP.get(ext, "application/octet-stream") try: - # Создать бакет если не существует - if not minio_client.bucket_exists(settings.MINIO_BUCKET_DOCS): - minio_client.make_bucket(settings.MINIO_BUCKET_DOCS) - - import io - minio_client.put_object( + get_minio().put_object( bucket_name=settings.MINIO_BUCKET_DOCS, object_name=minio_key, data=io.BytesIO(file_data), length=len(file_data), - content_type=file.content_type or "application/octet-stream", + content_type=content_type, ) - logger.info(f"Файл загружен в MinIO: {minio_key}") - - except S3Error as e: - logger.error(f"Ошибка загрузки в MinIO: {e}") + except Exception as e: + logger.error("Ошибка загрузки в MinIO: %s", e) raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Ошибка сохранения файла. Попробуйте позже.", ) - # Создать задачу в БД + # ── 5. Создаём задачу и диспатчим ───────────────────────────────────────── + task = Task( user_id=current_user.id, type="plagiarism", @@ -126,30 +128,23 @@ async def upload_for_plagiarism_check( "filename": file.filename, "minio_key": minio_key, "file_size_bytes": len(file_data), - "content_type": file.content_type, }, ) db.add(task) await db.flush() - task_id = task.id - # Диспатч в индексер воркер celery_result = celery_app.send_task( "index.extract_and_check", - args=[task_id, minio_key, file.filename], + args=[task.id, minio_key, file.filename], queue="queue.index", ) task.celery_task_id = celery_result.id task.queue_position = 1 - task.eta_seconds = 120 # ~2 минуты с учётом 4 уровней проверки + task.eta_seconds = 120 await db.commit() await db.refresh(task) - logger.info( - f"Задача проверки плагиата {task_id!r} создана для пользователя {current_user.id}, " - f"файл: {file.filename!r}" - ) - + logger.info("Задача плагиата %s создана для пользователя %d", task.public_id, current_user.id) return TaskResponse.model_validate(task) diff --git a/services/api/app/api/search.py b/services/api/app/api/search.py index 5fe4ab7..5f3abb3 100644 --- a/services/api/app/api/search.py +++ b/services/api/app/api/search.py @@ -6,8 +6,8 @@ from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.core.celery_app import celery_app -from app.core.rate_limiter import check_and_increment_limit, check_concurrent_limit -from app.core.security import get_current_user +from app.core.rate_limiter import acquire_concurrent_slot, check_and_increment_limit +from app.core.security import get_current_verified_user from app.database import get_db from app.models.task import Task from app.models.user import User @@ -17,41 +17,43 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/search", tags=["search"]) -# Примерное время ожидания в секундах для каждой позиции в очереди ETA_PER_POSITION_SECONDS = 15 @router.post("/", response_model=TaskResponse, status_code=status.HTTP_202_ACCEPTED) async def create_search_task( data: SearchRequest, - current_user: User = Depends(get_current_user), + # Поиск требует подтверждённого email — защита от массового abuse + current_user: User = Depends(get_current_verified_user), db: AsyncSession = Depends(get_db), ) -> TaskResponse: """ - Создать задачу семантического поиска источников. + Создать задачу поиска источников. - Возвращает TaskResponse с queue_position и eta_seconds сразу. - Результат доступен через GET /tasks/{task_id} или WebSocket /ws/tasks/{task_id}. + Возвращает сразу с task public_id и позицией в очереди. + Результат — через GET /tasks/{public_id} или WS /ws/tasks/{public_id}?token=JWT. """ - # Проверить лимит по тарифу - limit_check = check_and_increment_limit(current_user.id, "search", current_user.plan) - if not limit_check["allowed"]: + # Проверяем лимит (атомарно, Lua-скрипт) + limit_result = await check_and_increment_limit(current_user.id, "search", current_user.plan) + if not limit_result["allowed"]: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail=( - f"Превышен дневной лимит поиска для тарифа '{current_user.plan}'. " - f"Использовано {limit_check['current']} из {limit_check['limit']}." + f"Превышен дневной лимит поиска (тариф '{current_user.plan}'): " + f"{limit_result['current']}/{limit_result['limit']}. " + f"Сбросится {limit_result['reset_at']}." ), + headers={"Retry-After": "86400"}, ) - # Проверить лимит одновременных задач - if not check_concurrent_limit(current_user.id, current_user.plan): + # Проверяем лимит одновременных задач (атомарно) + slot_acquired = await acquire_concurrent_slot(current_user.id, current_user.plan) + if not slot_acquired: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Превышен лимит одновременных задач. Дождитесь завершения текущих.", ) - # Создать задачу в БД task = Task( user_id=current_user.id, type="search", @@ -65,31 +67,21 @@ async def create_search_task( }, ) db.add(task) - await db.flush() - task_id = task.id + await db.flush() # получаем id и public_id - # Диспатч в GPU воркер celery_result = celery_app.send_task( "gpu.search_semantic", - args=[task_id, data.query], - kwargs={ - "lang": data.lang, - "year_from": data.year_from, - "year_to": data.year_to, - }, + args=[task.id, data.query], + kwargs={"lang": data.lang, "year_from": data.year_from, "year_to": data.year_to}, queue="queue.gpu", ) task.celery_task_id = celery_result.id - task.queue_position = 1 # TODO: реальный подсчёт через Redis + task.queue_position = 1 task.eta_seconds = ETA_PER_POSITION_SECONDS await db.commit() await db.refresh(task) - logger.info( - f"Задача поиска {task_id!r} создана для пользователя {current_user.id}, " - f"запрос: {data.query[:50]!r}" - ) - + logger.info("Задача поиска %s создана для пользователя %d", task.public_id, current_user.id) return TaskResponse.model_validate(task) diff --git a/services/api/app/api/tasks.py b/services/api/app/api/tasks.py index 11a5ba4..f3c2ed3 100644 --- a/services/api/app/api/tasks.py +++ b/services/api/app/api/tasks.py @@ -1,4 +1,8 @@ -"""Роутер для управления задачами пользователя.""" +"""Управление задачами пользователя. + +В URL используется public_id (не внутренний UUID), чтобы не раскрывать +структуру внутренних идентификаторов. +""" import logging @@ -24,7 +28,10 @@ async def list_tasks( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> list[TaskResponse]: - """Получить список задач текущего пользователя (от новых к старым).""" + """Список задач текущего пользователя (новые первыми).""" + # Жёсткий потолок limit — не даём вытащить всю таблицу одним запросом + limit = min(limit, 100) + result = await db.execute( select(Task) .where(Task.user_id == current_user.id) @@ -32,52 +39,47 @@ async def list_tasks( .limit(limit) .offset(offset) ) - tasks = result.scalars().all() - return [TaskResponse.model_validate(t) for t in tasks] + return [TaskResponse.model_validate(t) for t in result.scalars().all()] -@router.get("/{task_id}", response_model=TaskResponse) +@router.get("/{public_id}", response_model=TaskResponse) async def get_task( - task_id: str, + public_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> TaskResponse: - """Получить детали конкретной задачи.""" + """Получить задачу по публичному ID. Возвращает 404 если задача чужая — не раскрываем факт существования.""" result = await db.execute( - select(Task).where(Task.id == task_id, Task.user_id == current_user.id) + select(Task).where( + Task.public_id == public_id, + Task.user_id == current_user.id, # ownership проверяется в одном запросе + ) ) task = result.scalar_one_or_none() if task is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Задача не найдена", - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Задача не найдена") return TaskResponse.model_validate(task) -@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT) +@router.delete("/{public_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_task( - task_id: str, + public_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> None: - """ - Удалить задачу. - - Нельзя удалить задачу в статусе 'processing'. - """ + """Удалить задачу. Нельзя удалить задачу в статусе 'processing'.""" result = await db.execute( - select(Task).where(Task.id == task_id, Task.user_id == current_user.id) + select(Task).where( + Task.public_id == public_id, + Task.user_id == current_user.id, + ) ) task = result.scalar_one_or_none() if task is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Задача не найдена", - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Задача не найдена") if task.status == "processing": raise HTTPException( diff --git a/services/api/app/config.py b/services/api/app/config.py index ac60be1..3698458 100644 --- a/services/api/app/config.py +++ b/services/api/app/config.py @@ -31,6 +31,7 @@ class Settings(BaseSettings): MINIO_SECRET_KEY: str = "changeme" MINIO_BUCKET_DOCS: str = "documents" MINIO_BUCKET_BACKUPS: str = "backups" + MINIO_SECURE: bool = False # True если MinIO за HTTPS # Elasticsearch ELASTICSEARCH_URL: str = "http://elasticsearch:9200" diff --git a/services/api/app/core/minio_client.py b/services/api/app/core/minio_client.py new file mode 100644 index 0000000..6d00c64 --- /dev/null +++ b/services/api/app/core/minio_client.py @@ -0,0 +1,48 @@ +"""Singleton MinIO клиент. + +MinIO SDK держит пул HTTP-соединений внутри себя — создаём один экземпляр +на весь процесс. Бакеты проверяем один раз при старте приложения. +""" + +import logging + +from minio import Minio +from minio.error import S3Error + +from app.config import settings + +logger = logging.getLogger(__name__) + +_client: Minio | None = None +_initialized_buckets: set[str] = set() + + +def get_minio() -> Minio: + """Вернуть singleton MinIO клиент.""" + global _client + if _client is None: + _client = Minio( + settings.MINIO_ENDPOINT, + access_key=settings.MINIO_ACCESS_KEY, + secret_key=settings.MINIO_SECRET_KEY, + secure=settings.MINIO_SECURE, + ) + return _client + + +def ensure_bucket(bucket: str) -> None: + """Создать бакет если не существует. Вызывается один раз при старте.""" + if bucket in _initialized_buckets: + return + + client = get_minio() + try: + if not client.bucket_exists(bucket): + client.make_bucket(bucket) + logger.info(f"MinIO: бакет '{bucket}' создан") + else: + logger.info(f"MinIO: бакет '{bucket}' существует") + _initialized_buckets.add(bucket) + except S3Error as e: + logger.error(f"MinIO: ошибка проверки бакета '{bucket}': {e}") + raise diff --git a/services/api/app/core/rate_limiter.py b/services/api/app/core/rate_limiter.py index 973acfa..910f023 100644 --- a/services/api/app/core/rate_limiter.py +++ b/services/api/app/core/rate_limiter.py @@ -1,13 +1,22 @@ -"""Redis-based rate limiter для проверки лимитов по тарифному плану.""" +"""Redis-based rate limiter с атомарными Lua-скриптами. -import json +Проблема наивного подхода (GET → проверка → INCR): + - Race condition: 10 конкурентных запросов могут одновременно пройти GET, + увидеть значение ниже лимита и все инкрементировать. + +Решение: один Lua-скрипт выполняется атомарно на стороне Redis. +Redis гарантирует, что между командами внутри скрипта нет других операций. +""" + +import logging from datetime import datetime, timezone -import redis +from app.core.redis_client import get_redis -from app.config import settings +logger = logging.getLogger(__name__) + +# ─── Лимиты по тарифам ──────────────────────────────────────────────────────── -# Лимиты по тарифным планам PLAN_LIMITS: dict[str, dict[str, int | None]] = { "free": { "search_per_day": 10, @@ -16,7 +25,7 @@ PLAN_LIMITS: dict[str, dict[str, int | None]] = { "concurrent": 1, }, "student": { - "search_per_day": None, # None = безлимит + "search_per_day": None, # None = безлимит "summarize_per_month": 30, "plagiarism_per_month": 10, "concurrent": 2, @@ -35,133 +44,125 @@ PLAN_LIMITS: dict[str, dict[str, int | None]] = { }, } -# Маппинг действий на ключи лимитов ACTION_TO_LIMIT: dict[str, tuple[str, str]] = { - "search": ("search_per_day", "day"), - "summarize": ("summarize_per_month", "month"), - "plagiarism": ("plagiarism_per_month", "month"), - "gost": ("gost_per_month", "month"), # ГОСТ всегда разрешён + "search": ("search_per_day", "day"), + "summarize": ("summarize_per_month", "month"), + "plagiarism":("plagiarism_per_month", "month"), + # gost не ограничен — не в маппинге } +# ─── Lua-скрипты ────────────────────────────────────────────────────────────── -def get_redis_client() -> redis.Redis: - """Создать синхронный Redis клиент.""" - return redis.from_url(settings.REDIS_URL, decode_responses=True) +# Атомарная проверка + инкремент лимита. +# Возвращает [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 _get_period_key(period: str) -> str: - """Получить строку периода для Redis ключа.""" +def _period_suffix(period: str) -> str: now = datetime.now(timezone.utc) - if period == "day": - return now.strftime("%Y-%m-%d") - elif period == "month": - return now.strftime("%Y-%m") - return now.strftime("%Y-%m-%d") + return now.strftime("%Y-%m-%d") if period == "day" else now.strftime("%Y-%m") -def check_and_increment_limit(user_id: int, action: str, plan: str) -> dict: +async def check_and_increment_limit( + user_id: int, + action: str, + plan: str, +) -> dict: """ - Проверить лимит и инкрементировать счётчик. - - Args: - user_id: ID пользователя - action: Действие (search, plagiarism, summarize, gost) - plan: Тарифный план пользователя + Атомарно проверить лимит и инкрементировать счётчик. Returns: - dict с полями: - - allowed: bool — разрешено ли действие - - current: int — текущее количество использований - - limit: int | None — лимит (None = безлимит) - - remaining: int | None — осталось использований - - reset_at: str — когда сбрасывается счётчик + {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, - "reset_at": None, - } + if limit_value is None: # безлимит + return {"allowed": True, "current": 0, "limit": None, "remaining": None} - period_str = _get_period_key(period) - redis_key = f"rate:{user_id}:{action}:{period_str}" + 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_client() - - # Атомарно инкрементировать - pipe = r.pipeline() - pipe.incr(redis_key) - # Устанавливаем TTL: для дня — 86400 сек, для месяца — 32 дня - ttl = 86400 if period == "day" else 86400 * 32 - pipe.expire(redis_key, ttl) - results = pipe.execute() - - current = results[0] - - if current > limit_value: - # Декрементировать обратно (не считать запрещённые) - r.decr(redis_key) - current -= 1 - return { - "allowed": False, - "current": current, - "limit": limit_value, - "remaining": 0, - "reset_at": period_str, - } + 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": True, + "allowed": allowed, "current": current, "limit": limit_value, - "remaining": limit_value - current, + "remaining": max(0, limit_value - current) if allowed else 0, "reset_at": period_str, } -def check_concurrent_limit(user_id: int, plan: str) -> bool: +async def acquire_concurrent_slot(user_id: int, plan: str) -> bool: """ - Проверить лимит одновременных задач. + Атомарно захватить слот одновременной задачи. Returns: - True если можно создать новую задачу, False если превышен лимит. + True — слот получен (задачу можно создавать). + False — все слоты заняты. """ limits = PLAN_LIMITS.get(plan, PLAN_LIMITS["free"]) max_concurrent = limits.get("concurrent", 1) - r = get_redis_client() + 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}" - current = r.get(key) - - return (current is None) or (int(current) < max_concurrent) - - -def increment_concurrent(user_id: int) -> None: - """Увеличить счётчик одновременных задач (при создании задачи).""" - r = get_redis_client() - key = f"concurrent:{user_id}" - pipe = r.pipeline() - pipe.incr(key) - pipe.expire(key, 3600) # Автосброс через 1 час - pipe.execute() - - -def decrement_concurrent(user_id: int) -> None: - """Уменьшить счётчик одновременных задач (при завершении задачи).""" - r = get_redis_client() - key = f"concurrent:{user_id}" - current = r.get(key) + # DECR безопасен: Redis не уходит в отрицательные значения если мы контролируем acquire + current = await r.get(key) if current and int(current) > 0: - r.decr(key) + await r.decr(key) diff --git a/services/api/app/core/redis_client.py b/services/api/app/core/redis_client.py new file mode 100644 index 0000000..1d79586 --- /dev/null +++ b/services/api/app/core/redis_client.py @@ -0,0 +1,36 @@ +"""Singleton пул Redis-соединений. + +Один ConnectionPool на весь процесс — не создаём новое соединение на каждый запрос. +Используем redis.asyncio для совместимости с FastAPI async endpoints. +""" + +import redis.asyncio as aioredis + +from app.config import settings + +_pool: aioredis.ConnectionPool | None = None + + +def get_pool() -> aioredis.ConnectionPool: + """Инициализировать пул при первом вызове, затем возвращать тот же.""" + global _pool + if _pool is None: + _pool = aioredis.ConnectionPool.from_url( + settings.REDIS_URL, + max_connections=50, + decode_responses=True, + ) + return _pool + + +def get_redis() -> aioredis.Redis: + """Получить Redis клиент из пула (не создаёт новое TCP-соединение).""" + return aioredis.Redis(connection_pool=get_pool()) + + +async def close_pool() -> None: + """Закрыть пул при остановке приложения.""" + global _pool + if _pool: + await _pool.aclose() + _pool = None diff --git a/services/api/app/core/security.py b/services/api/app/core/security.py index f37dd2d..89491b7 100644 --- a/services/api/app/core/security.py +++ b/services/api/app/core/security.py @@ -1,9 +1,11 @@ -"""Утилиты безопасности: JWT, хэширование паролей, dependency для получения текущего пользователя.""" +"""JWT, хэширование паролей, dependency получения пользователя с кэшированием.""" +import json +import logging from datetime import datetime, timedelta, timezone from typing import Any -from fastapi import Depends, HTTPException, status +from fastapi import Depends, HTTPException, Query, WebSocket, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt from passlib.context import CryptContext @@ -11,53 +13,47 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings +from app.core.redis_client import get_redis from app.database import get_db -# Контекст хэширования паролей (bcrypt) -pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +logger = logging.getLogger(__name__) -# OAuth2 схема +pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login") +# TTL кэша пользователя в Redis (5 минут) +_USER_CACHE_TTL = 300 + def hash_password(password: str) -> str: - """Хэшировать пароль через bcrypt.""" return pwd_context.hash(password) -def verify_password(plain_password: str, hashed_password: str) -> bool: - """Проверить соответствие пароля его хэшу.""" - return pwd_context.verify(plain_password, hashed_password) +def verify_password(plain: str, hashed: str) -> bool: + return pwd_context.verify(plain, hashed) def create_access_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str: - """Создать JWT токен доступа.""" - to_encode = data.copy() + payload = data.copy() expire = datetime.now(timezone.utc) + ( expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) ) - to_encode.update({"exp": expire}) - return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) + payload["exp"] = expire + return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM) -def verify_token(token: str) -> dict[str, Any]: +def _decode_token(token: str) -> int: """ - Декодировать и верифицировать JWT токен. - - Raises: - HTTPException: если токен невалиден или просрочен. + Декодировать JWT и вернуть user_id. + Raises HTTPException 401 при любой проблеме с токеном. """ try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) - user_id: int | None = payload.get("sub") - if user_id is None: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Неверный токен аутентификации", - headers={"WWW-Authenticate": "Bearer"}, - ) - return payload - except JWTError: + sub = payload.get("sub") + if sub is None: + raise ValueError("missing sub") + return int(sub) + except (JWTError, ValueError): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Невалидный или просроченный токен", @@ -65,46 +61,79 @@ def verify_token(token: str) -> dict[str, Any]: ) -async def get_current_user( - token: str = Depends(oauth2_scheme), - db: AsyncSession = Depends(get_db), -): - """ - FastAPI dependency: получить текущего авторизованного пользователя из JWT токена. +async def _load_user(user_id: int, db: AsyncSession): + """Загрузить пользователя из Redis-кэша или из БД.""" + from app.models.user import User # избегаем circular import на уровне модуля - Raises: - HTTPException 401: если токен невалиден. - HTTPException 404: если пользователь не найден. - """ - from app.models.user import User + cache_key = f"user:cache:{user_id}" + r = get_redis() - payload = verify_token(token) - user_id: int = int(payload.get("sub")) + # Пробуем кэш + cached = await r.get(cache_key) + if cached: + data = json.loads(cached) + # Возвращаем "живой" объект из БД только по id, но без лишнего SELECT + # Создаём User без ORM-связей (достаточно для проверок в роутерах) + u = User.__new__(User) + u.__dict__.update(data) + return u + # Кэш пустой — идём в БД result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="Пользователь не найден", - ) + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Пользователь не найден") + + # Сохраняем в кэш (только безопасные поля, без hashed_password) + safe = { + "id": user.id, + "email": user.email, + "name": user.name, + "plan": user.plan, + "is_verified": user.is_verified, + } + await r.setex(cache_key, _USER_CACHE_TTL, json.dumps(safe)) return user -async def get_current_verified_user( - current_user=Depends(get_current_user), +async def get_current_user( + token: str = Depends(oauth2_scheme), + db: AsyncSession = Depends(get_db), ): - """ - FastAPI dependency: только верифицированные пользователи. + """Dependency: текущий пользователь из JWT с Redis-кэшем.""" + user_id = _decode_token(token) + return await _load_user(user_id, db) - Raises: - HTTPException 403: если email не подтверждён. - """ + +async def get_current_verified_user(current_user=Depends(get_current_user)): + """Dependency: только верифицированные пользователи.""" if not current_user.is_verified: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Необходимо подтвердить email адрес", ) return current_user + + +async def get_ws_user( + websocket: WebSocket, + token: str = Query(..., description="JWT токен (передаётся как query-параметр ?token=...)"), + db: AsyncSession = Depends(get_db), +): + """ + Dependency для WebSocket: аутентификация через query-параметр ?token=JWT. + + WebSocket API браузера не позволяет передавать Authorization header, + поэтому токен передаётся в query-строке: + ws://host/ws/tasks/{public_id}?token= + """ + user_id = _decode_token(token) + return await _load_user(user_id, db) + + +async def invalidate_user_cache(user_id: int) -> None: + """Сбросить кэш пользователя (при смене тарифа, пароля и т.д.).""" + r = get_redis() + await r.delete(f"user:cache:{user_id}") diff --git a/services/api/app/main.py b/services/api/app/main.py index 2e6ea1c..9fd76be 100644 --- a/services/api/app/main.py +++ b/services/api/app/main.py @@ -1,113 +1,135 @@ -"""Точка входа FastAPI приложения — Академический помощник (anti-plagiarism).""" +"""Точка входа FastAPI приложения — Академический помощник.""" import logging from contextlib import asynccontextmanager from typing import AsyncGenerator -from fastapi import FastAPI, WebSocket, WebSocketDisconnect +from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from app.api import auth, documents, reports, search, tasks from app.config import settings +from app.core.minio_client import ensure_bucket +from app.core.redis_client import close_pool +from app.core.security import get_ws_user from app.core.websocket_manager import ws_manager from app.database import engine logger = logging.getLogger(__name__) -logging.basicConfig(level=logging.INFO) +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: - """ - Lifecycle менеджер приложения. - Выполняет инициализацию при старте и очистку при остановке. - """ + """Инициализация при старте, очистка при остановке.""" logger.info("Запуск Академического помощника...") - # Проверка соединений при старте + # Проверяем PostgreSQL try: + import sqlalchemy async with engine.connect() as conn: - await conn.execute(__import__("sqlalchemy").text("SELECT 1")) - logger.info("PostgreSQL: соединение установлено") + await conn.execute(sqlalchemy.text("SELECT 1")) + logger.info("PostgreSQL: OK") except Exception as e: - logger.error(f"PostgreSQL: ошибка соединения: {e}") + logger.error(f"PostgreSQL: {e}") + + # Инициализируем MinIO бакеты (один раз при старте, не на каждый запрос) + try: + ensure_bucket(settings.MINIO_BUCKET_DOCS) + ensure_bucket(settings.MINIO_BUCKET_BACKUPS) + logger.info("MinIO: бакеты готовы") + except Exception as e: + logger.error(f"MinIO: {e}") yield - # Очистка при остановке logger.info("Остановка приложения...") await engine.dispose() + await close_pool() app = FastAPI( title="Академический помощник API", - description=( - "Сервис антиплагиата и поиска академических источников. " - "Семантический поиск через FAISS GPU + Elasticsearch, " - "проверка плагиата в 4 уровня, ГОСТ библиография." - ), version="0.1.0", lifespan=lifespan, - docs_url="/api/docs", - redoc_url="/api/redoc", - openapi_url="/api/openapi.json", + docs_url="/api/docs" if settings.ENVIRONMENT == "development" else None, + redoc_url=None, + openapi_url="/api/openapi.json" if settings.ENVIRONMENT == "development" else None, ) # ─── CORS ───────────────────────────────────────────────────────────────────── +# allow_credentials=True несовместим с allow_origins=["*"]. +# Указываем явные origins. Wildcard заголовки/методы запрещены с credentials. app.add_middleware( CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], + allow_methods=["GET", "POST", "DELETE", "OPTIONS"], + allow_headers=["Authorization", "Content-Type", "Accept"], ) # ─── Роутеры ────────────────────────────────────────────────────────────────── -app.include_router(auth.router, prefix="/api") -app.include_router(tasks.router, prefix="/api") -app.include_router(search.router, prefix="/api") +app.include_router(auth.router, prefix="/api") +app.include_router(tasks.router, prefix="/api") +app.include_router(search.router, prefix="/api") app.include_router(documents.router, prefix="/api") -app.include_router(reports.router, prefix="/api") +app.include_router(reports.router, prefix="/api") # ─── WebSocket ──────────────────────────────────────────────────────────────── -@app.websocket("/ws/tasks/{task_id}") -async def websocket_task_updates(websocket: WebSocket, task_id: str) -> None: +@app.websocket("/ws/tasks/{public_id}") +async def websocket_task_updates( + websocket: WebSocket, + public_id: str, + # Аутентификация через ?token=JWT (браузеры не могут передавать Authorization header в WS) + current_user=Depends(get_ws_user), +) -> None: """ - WebSocket endpoint для real-time обновлений статуса задачи. + Подписка на обновления задачи в реальном времени. - Клиент подключается и получает обновления при изменении статуса задачи. - Соединение закрывается при получении статуса done/failed. + Подключение: ws://host/ws/tasks/{public_id}?token= + + Пользователь получает только обновления СВОИХ задач — + ownership проверяется до установки соединения. """ - await ws_manager.connect(task_id, websocket) + from sqlalchemy import select + from app.database import AsyncSessionLocal + from app.models.task import Task + + # Проверяем что задача принадлежит этому пользователю + async with AsyncSessionLocal() as db: + result = await db.execute( + select(Task).where( + Task.public_id == public_id, + Task.user_id == current_user.id, + ) + ) + task = result.scalar_one_or_none() + + if task is None: + await websocket.close(code=4004, reason="Task not found or access denied") + return + + await ws_manager.connect(public_id, websocket) try: while True: - # Ждём входящих сообщений (ping/pong для поддержания соединения) data = await websocket.receive_text() if data == "ping": await websocket.send_text("pong") except WebSocketDisconnect: - ws_manager.disconnect(task_id, websocket) - logger.info(f"WebSocket клиент отключился от задачи {task_id!r}") + ws_manager.disconnect(public_id, websocket) -# ─── Health check ────────────────────────────────────────────────────────────── +# ─── Health ──────────────────────────────────────────────────────────────────── @app.get("/health", tags=["monitoring"]) async def health_check() -> JSONResponse: - """Проверка работоспособности сервиса.""" - return JSONResponse( - content={ - "status": "ok", - "service": "api", - "version": "0.1.0", - } - ) + return JSONResponse({"status": "ok", "version": "0.1.0"}) @app.get("/", include_in_schema=False) async def root() -> JSONResponse: - """Корневой endpoint — редирект к документации.""" - return JSONResponse( - content={"message": "Академический помощник API", "docs": "/api/docs"} - ) + return JSONResponse({"message": "Академический помощник API"}) diff --git a/services/api/app/schemas/tasks.py b/services/api/app/schemas/tasks.py index 2c50df3..f4ef58d 100644 --- a/services/api/app/schemas/tasks.py +++ b/services/api/app/schemas/tasks.py @@ -14,14 +14,19 @@ class TaskCreate(BaseModel): class TaskResponse(BaseModel): - """Ответ с информацией о задаче.""" + """Ответ с информацией о задаче. - id: str + В API возвращается public_id (не внутренний UUID). + Клиент использует public_id для всех операций с задачей. + """ + + # public_id — это то, что видит клиент и использует в URL + public_id: str type: str status: str queue_position: int | None = None eta_seconds: int | None = None - input_data: dict[str, Any] = Field(default_factory=dict) + # input_data не возвращаем (может содержать minio_key и другие внутренние данные) result: dict[str, Any] | None = None error: str | None = None created_at: datetime diff --git a/services/api/requirements.txt b/services/api/requirements.txt index 9b021bc..e0a3b8d 100644 --- a/services/api/requirements.txt +++ b/services/api/requirements.txt @@ -4,7 +4,7 @@ sqlalchemy==2.0.30 alembic==1.13.1 asyncpg==0.29.0 psycopg2-binary==2.9.9 -redis==5.0.4 +redis[asyncio]==5.0.4 celery==5.4.0 pydantic==2.7.1 pydantic-settings==2.2.1