Max Ronzhin
84b7f807f7
На нагрузочном тесте 31.07.2026 около 70 человек заходили одновременно. Вход развалился: p95 `/api/v1/auth/token` — 7.28 с, p95 `guest-join` — 7.06 с, в БД 33 соединения `idle in transaction` при ОДНОМ активном запросе. Люди попадали внутрь с пятой-десятой попытки, часть не попала вовсе. Медиа при этом работало штатно: 30 участников с 27 камерами в следующем окне прошли без единого лага. Причина — argon2 считался синхронно внутри async-обработчика. Замер на боевом сервере: 95–155 мс на одну проверку, и всё это время event loop процесса стоит целиком. Транзакция БД к тому моменту уже открыта (`get_by_email` сделал SELECT), поэтому соединение висело без работы, пул из 40 выбирался, и отказы получали совершенно посторонние ручки — включая вход в конференцию, где никакого пароля не проверялось. Что изменилось: - `hash_password`/`verify_password` стали асинхронными и считаются в пуле потоков (`asyncio.to_thread`). argon2-cffi освобождает GIL, поэтому проверки идут по-настоящему параллельно; - параметры argon2id заменены с дефолтов библиотеки (t=3, m=64 МБ, p=4) на рекомендацию OWASP (t=2, m=19 МБ, p=1): 95 мс → 42 мс. Отдельно важен `parallelism`: при p=4 одна проверка пароля занимала все четыре ядра сервера — те же, на которых работает LiveKit; - добавлен `needs_rehash`: существующие хэши проверяются как прежде (параметры зашиты в саму строку) и лениво перевыпускаются при первом успешном входе. Расчёт по замерам: пачка из 70 логинов — 6.7–10.9 с блокировки против ~0.36 с без неё. Тесты: event loop продолжает тикать во время проверки; 8 параллельных проверок укладываются заметно быстрее восьми последовательных; хэш со старыми параметрами принимается и перевыпускается при входе.
254 lines
12 KiB
Python
254 lines
12 KiB
Python
"""Бизнес-логика аутентификации: регистрация, подтверждение email, JWT access/refresh.
|
||
|
||
Refresh-токены хранятся server-side в Redis (`refresh:{jti}` -> user_id) с
|
||
TTL, равным сроку жизни refresh-токена. Каждое успешное использование
|
||
refresh-токена ротирует его: старый `jti` немедленно удаляется, выдаётся
|
||
новый; повторное использование уже потраченного refresh-токена (reuse)
|
||
обнаруживается по отсутствию ключа в Redis.
|
||
"""
|
||
|
||
import hashlib
|
||
import secrets
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime, timedelta
|
||
|
||
import jwt
|
||
from redis.asyncio import Redis
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from core.config import get_settings
|
||
from core.security import (
|
||
create_access_token,
|
||
create_refresh_token,
|
||
decode_token,
|
||
hash_password,
|
||
needs_rehash,
|
||
verify_password,
|
||
)
|
||
from models.email_verification import EmailVerificationToken
|
||
from models.user import User
|
||
from repositories.admin import TeamRepository
|
||
from repositories.users import UserRepository
|
||
from services.email import EmailBackend
|
||
from services.instance_settings import InstanceSettingsService
|
||
|
||
REFRESH_KEY_PREFIX = "refresh:"
|
||
|
||
|
||
class EmailAlreadyRegisteredError(Exception):
|
||
"""Пользователь с таким email уже зарегистрирован."""
|
||
|
||
|
||
class InvalidTeamSelectionError(Exception):
|
||
"""Выбор команды при регистрации недоступен или команда не существует.
|
||
|
||
Публичный эндпоинт `/auth/register` не должен различать эти две причины
|
||
в ответе (не раскрываем администраторскую настройку/список команд
|
||
перебором id) — единая ошибка для обоих случаев.
|
||
"""
|
||
|
||
|
||
class InvalidEmailDomainError(Exception):
|
||
"""Домен email регистрирующегося не совпадает с эталонным доменом инстанса.
|
||
|
||
Поднимается только при включённой настройке инстанса
|
||
`registration_email_domain_enabled` (см. `InstanceSettingsService`).
|
||
"""
|
||
|
||
|
||
class InvalidVerificationTokenError(Exception):
|
||
"""Токен подтверждения email не найден, просрочен или уже использован."""
|
||
|
||
|
||
class InvalidCredentialsError(Exception):
|
||
"""Неверный email или пароль."""
|
||
|
||
|
||
class EmailNotVerifiedError(Exception):
|
||
"""Email пользователя ещё не подтверждён."""
|
||
|
||
|
||
class InvalidRefreshTokenError(Exception):
|
||
"""Refresh-токен невалиден, просрочен, отозван или уже был использован (reuse)."""
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class TokenPair:
|
||
"""Пара выданных JWT-токенов (access — в теле ответа, refresh — в cookie)."""
|
||
|
||
access_token: str
|
||
refresh_token: str
|
||
|
||
|
||
class AuthService:
|
||
"""Инкапсулирует сценарии регистрации, входа, обновления и отзыва токенов."""
|
||
|
||
def __init__(self, session: AsyncSession, redis: Redis, email_backend: EmailBackend) -> None:
|
||
self._session = session
|
||
self._redis = redis
|
||
self._email_backend = email_backend
|
||
self._users = UserRepository(session)
|
||
self._settings = get_settings()
|
||
|
||
async def register(
|
||
self,
|
||
*,
|
||
email: str,
|
||
name_user: str,
|
||
password: str,
|
||
team_id: uuid.UUID | None = None,
|
||
) -> User:
|
||
"""Зарегистрировать пользователя и отправить письмо для подтверждения email.
|
||
|
||
`team_id` допустим, только если в настройках инстанса включён выбор
|
||
команды при регистрации (`registration_team_choice`) и команда
|
||
существует — иначе `InvalidTeamSelectionError` (публичный
|
||
эндпоинт, деталей не раскрываем). Если включена верификация домена
|
||
email (`registration_email_domain_enabled`), домен `email` (часть
|
||
после `@`, без учёта регистра) должен совпадать с одним из
|
||
эталонных доменов (`registration_email_domains`) — иначе
|
||
`InvalidEmailDomainError`. Обе проверки — до создания пользователя.
|
||
"""
|
||
existing = await self._users.get_by_email(email)
|
||
if existing is not None:
|
||
raise EmailAlreadyRegisteredError(email)
|
||
|
||
cfg = await InstanceSettingsService(self._session).get()
|
||
|
||
if cfg.registration_email_domain_enabled:
|
||
email_domain = email.rsplit("@", 1)[-1].lower()
|
||
if email_domain not in cfg.registration_email_domains:
|
||
raise InvalidEmailDomainError(email)
|
||
|
||
if team_id is not None:
|
||
if not cfg.registration_team_choice:
|
||
raise InvalidTeamSelectionError(team_id)
|
||
team = await TeamRepository(self._session).get(team_id)
|
||
if team is None:
|
||
raise InvalidTeamSelectionError(team_id)
|
||
|
||
user = await self._users.create(
|
||
email=email,
|
||
name_user=name_user,
|
||
password_hash=await hash_password(password),
|
||
team_id=team_id,
|
||
)
|
||
reply_to = cfg.contact_email if cfg.contact_email_enabled else None
|
||
await self._issue_verification_email(user, reply_to=reply_to)
|
||
await self._session.commit()
|
||
return user
|
||
|
||
async def verify_email(self, token: str) -> None:
|
||
"""Подтвердить email пользователя по токену из письма."""
|
||
token_hash = _hash_token(token)
|
||
result = await self._session.execute(
|
||
select(EmailVerificationToken).where(EmailVerificationToken.token_hash == token_hash)
|
||
)
|
||
record = result.scalar_one_or_none()
|
||
now = datetime.now(UTC)
|
||
if record is None or record.used_at is not None or record.expires_at < now:
|
||
raise InvalidVerificationTokenError
|
||
|
||
user = await self._users.get_by_id(record.user_id)
|
||
if user is None:
|
||
raise InvalidVerificationTokenError
|
||
|
||
record.used_at = now
|
||
user.email_verified = True
|
||
await self._session.commit()
|
||
|
||
async def login(self, *, email: str, password: str) -> TokenPair:
|
||
"""Проверить учётные данные и выдать пару access/refresh токенов."""
|
||
user = await self._users.get_by_email(email)
|
||
if user is None or not await verify_password(password, user.password_hash):
|
||
raise InvalidCredentialsError
|
||
if not user.email_verified:
|
||
raise EmailNotVerifiedError
|
||
|
||
# Постепенная миграция на актуальные параметры argon2 (см. core/security.py):
|
||
# параметры зашиты в саму строку хэша, поэтому старые записи так и
|
||
# проверялись бы вдвое дольше. Открытый пароль есть только здесь и
|
||
# только сейчас — другого места для перевыпуска не будет.
|
||
if needs_rehash(user.password_hash):
|
||
user.password_hash = await hash_password(password)
|
||
# Явный commit: выдача токенов идёт через Redis и БД не трогает,
|
||
# поэтому без него перевыпущенный хэш откатился бы вместе с сессией.
|
||
await self._session.commit()
|
||
|
||
return await self._issue_token_pair(user.id, user.role)
|
||
|
||
async def refresh(self, refresh_token: str) -> TokenPair:
|
||
"""Провалидировать refresh-токен, ротировать его и выдать новую пару токенов."""
|
||
user_id = await self._validate_and_consume(refresh_token)
|
||
user = await self._users.get_by_id(user_id)
|
||
if user is None:
|
||
raise InvalidRefreshTokenError
|
||
return await self._issue_token_pair(user.id, user.role)
|
||
|
||
async def logout(self, refresh_token: str) -> None:
|
||
"""Отозвать refresh-токен (удалить его из Redis), если он вообще декодируется."""
|
||
try:
|
||
payload = decode_token(refresh_token)
|
||
except jwt.PyJWTError:
|
||
return
|
||
jti = payload.get("jti")
|
||
if jti:
|
||
await self._redis.delete(f"{REFRESH_KEY_PREFIX}{jti}")
|
||
|
||
async def _validate_and_consume(self, refresh_token: str) -> uuid.UUID:
|
||
"""Проверить refresh JWT и его наличие в Redis, затем сразу удалить (ротация)."""
|
||
try:
|
||
payload = decode_token(refresh_token)
|
||
except jwt.PyJWTError as exc:
|
||
raise InvalidRefreshTokenError from exc
|
||
|
||
if payload.get("type") != "refresh":
|
||
raise InvalidRefreshTokenError
|
||
|
||
jti = payload.get("jti")
|
||
sub = payload.get("sub")
|
||
if not jti or not sub:
|
||
raise InvalidRefreshTokenError
|
||
|
||
redis_key = f"{REFRESH_KEY_PREFIX}{jti}"
|
||
stored_user_id = await self._redis.get(redis_key)
|
||
if stored_user_id is None or stored_user_id != sub:
|
||
raise InvalidRefreshTokenError
|
||
|
||
# Немедленное удаление использованного jti: повторное предъявление
|
||
# того же refresh-токена (reuse) после этой точки всегда даст 401.
|
||
await self._redis.delete(redis_key)
|
||
return uuid.UUID(sub)
|
||
|
||
async def _issue_token_pair(self, user_id: uuid.UUID, role: str) -> TokenPair:
|
||
access_token = create_access_token(user_id, role)
|
||
refresh_token, jti = create_refresh_token(user_id)
|
||
ttl_seconds = self._settings.refresh_token_ttl_days * 24 * 3600
|
||
await self._redis.set(f"{REFRESH_KEY_PREFIX}{jti}", str(user_id), ex=ttl_seconds)
|
||
return TokenPair(access_token=access_token, refresh_token=refresh_token)
|
||
|
||
async def _issue_verification_email(self, user: User, *, reply_to: str | None = None) -> None:
|
||
token = secrets.token_urlsafe(32) # 256 бит случайности
|
||
expires_at = datetime.now(UTC) + timedelta(
|
||
hours=self._settings.email_verification_ttl_hours
|
||
)
|
||
self._session.add(
|
||
EmailVerificationToken(
|
||
user_id=user.id, token_hash=_hash_token(token), expires_at=expires_at
|
||
)
|
||
)
|
||
link = f"{self._settings.frontend_url}/verify-email?token={token}"
|
||
await self._email_backend.send(
|
||
to=user.email,
|
||
subject="Подтверждение регистрации VidConf",
|
||
body=f"Для подтверждения email перейдите по ссылке: {link}",
|
||
reply_to=reply_to,
|
||
)
|
||
|
||
|
||
def _hash_token(token: str) -> str:
|
||
"""Захэшировать токен подтверждения email алгоритмом sha256 (hex-строка)."""
|
||
return hashlib.sha256(token.encode()).hexdigest()
|