273 lines
16 KiB
Python
273 lines
16 KiB
Python
"""Аутентификация: проверка JWT и фабрика проверяющего по настройкам (D-5).
|
||
|
||
Модуль ничего не знает про HTTP и БД: он превращает строку токена в :class:`Claims` или бросает :class:`AuthError`
|
||
(сообщение по-русски, пригодно для ``{"detail": ...}`` ответа 401). Поиск ``app_user`` по email и права — зависимости
|
||
``src/api/dependencies/auth.py`` (задача 17).
|
||
|
||
Проверяющие:
|
||
|
||
* :class:`DevHs256Verifier` — только для разработки: HS256 с общим секретом ``AUTH_DEV_SECRET``; выпускает токены для
|
||
``POST /api/v1/dev/token``. Создаётся фабрикой при ``AUTH_DEV_ISSUER=true`` **и** ``DEBUG=true``.
|
||
* :class:`JwksVerifier` — каркас под токены РАИСА (параметры JWT РАИСА неизвестны, решение #5 design.md): подпись
|
||
RS256/ES256 по публичному ключу ``AUTH_PUBLIC_KEY`` (PEM) либо по ключам из ``AUTH_JWKS_URL`` (с кэшем).
|
||
|
||
Фабрика :func:`build_verifier` возвращает ``None``, если ни dev-режим, ни ключи РАИСА не настроены: в этом случае
|
||
токены не принимаются вовсе (зависимость ``current_user`` отвечает 401 «аутентификация не настроена»).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import TYPE_CHECKING, Any, Protocol
|
||
|
||
import jwt
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric import ec, rsa
|
||
from jwt import PyJWKClient
|
||
from loguru import logger
|
||
|
||
if TYPE_CHECKING:
|
||
from src.core.config import Settings
|
||
|
||
# Алгоритмы РАИСА-проверяющего: асимметричные, явно. ``none`` и HS* не допускаются никогда (атака подмены алгоритма).
|
||
JWKS_ALGORITHMS: tuple[str, ...] = ("RS256", "ES256")
|
||
DEV_ALGORITHM = "HS256"
|
||
# Допуск на расхождение часов с эмитентом токена
|
||
JWKS_LEEWAY_SECONDS = 10
|
||
# Минимальная длина dev-секрета HS256 (RFC 7518 §3.2: не короче выхода SHA-256)
|
||
DEV_SECRET_MIN_BYTES = 32
|
||
# Срок жизни dev-токена (требование спеки authn-authz)
|
||
DEV_TOKEN_TTL = timedelta(hours=8)
|
||
# Время жизни кэша набора ключей JWKS (PyJWKClient), секунды. Кэш отдельных ключей по kid (cache_keys) выключен
|
||
# намеренно: он без срока жизни и оставил бы отозванный ключ действующим до перезапуска процесса.
|
||
JWKS_CACHE_LIFESPAN_SECONDS = 300
|
||
JWKS_FETCH_TIMEOUT_SECONDS = 10
|
||
# Предел длины подконтрольного клиенту текста (kid и т.п.) в строке лога
|
||
_LOG_TEXT_LIMIT = 200
|
||
|
||
|
||
class AuthError(Exception):
|
||
"""Токен не принят. Текст сообщения — по-русски, без внутренностей; годится для тела 401."""
|
||
|
||
|
||
class TokenExpiredError(AuthError):
|
||
"""Срок действия токена истёк."""
|
||
|
||
|
||
class TokenSignatureError(AuthError):
|
||
"""Подпись токена не проходит проверку."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Claims:
|
||
"""Проверенные данные токена: ``sub``, email пользователя и момент окончания срока (UTC)."""
|
||
|
||
subject: str
|
||
email: str
|
||
exp: datetime
|
||
|
||
|
||
class TokenVerifier(Protocol):
|
||
"""Проверяющий токены. ``verify`` асинхронный: JWKS-проверяющий может ходить за ключами по сети."""
|
||
|
||
async def verify(self, token: str) -> Claims:
|
||
"""Возвращает claims проверенного токена или бросает :class:`AuthError`."""
|
||
...
|
||
|
||
|
||
class _SigningKeyProvider(Protocol):
|
||
"""Источник ключа проверки подписи по токену (интерфейс :class:`jwt.PyJWKClient`)."""
|
||
|
||
def get_signing_key_from_jwt(self, token: str) -> jwt.PyJWK: ...
|
||
|
||
|
||
# Порядок важен: более конкретные классы исключений PyJWT — раньше общих
|
||
_ERROR_MESSAGES: tuple[tuple[type[jwt.PyJWTError], type[AuthError], str], ...] = (
|
||
(jwt.ExpiredSignatureError, TokenExpiredError, "срок действия токена истёк"),
|
||
(jwt.InvalidSignatureError, TokenSignatureError, "недействительная подпись токена"),
|
||
(jwt.InvalidAlgorithmError, AuthError, "недопустимый алгоритм подписи токена"),
|
||
(jwt.InvalidIssuerError, AuthError, "токен выпущен неизвестным эмитентом"),
|
||
(jwt.InvalidAudienceError, AuthError, "токен предназначен для другого сервиса"),
|
||
(jwt.ImmatureSignatureError, AuthError, "токен ещё не действителен"),
|
||
(jwt.DecodeError, AuthError, "некорректный токен"),
|
||
)
|
||
|
||
|
||
def _to_auth_error(exc: jwt.PyJWTError) -> AuthError:
|
||
"""Переводит ошибку PyJWT в :class:`AuthError` с русским сообщением; детали библиотеки наружу не отдаются."""
|
||
for exc_type, auth_type, message in _ERROR_MESSAGES:
|
||
if isinstance(exc, exc_type):
|
||
return auth_type(message)
|
||
if isinstance(exc, jwt.MissingRequiredClaimError):
|
||
return AuthError(f"в токене нет обязательного поля «{exc.claim}»")
|
||
if isinstance(exc, jwt.PyJWKClientError):
|
||
# сбой получения ключей (сеть, неизвестный kid) — деталь только в лог, не в ответ
|
||
# сообщение может содержать kid из заголовка токена (его присылает клиент): repr экранирует переводы строк,
|
||
# усечение ограничивает размер
|
||
logger.warning("Не удалось получить ключ проверки подписи JWT: {}", repr(str(exc))[:_LOG_TEXT_LIMIT])
|
||
return AuthError("не удалось получить ключ проверки подписи токена")
|
||
return AuthError("недействительный токен")
|
||
|
||
|
||
def _claims_from_payload(payload: dict[str, Any], email_claim: str) -> Claims:
|
||
"""Собирает :class:`Claims` из проверенной полезной нагрузки; ``exp`` и ``sub`` уже проверены на наличие.
|
||
|
||
Email приводится к нижнему регистру (``strip().lower()``): сопоставление с ``app_user.email`` — без учёта регистра.
|
||
"""
|
||
email = payload.get(email_claim)
|
||
if not isinstance(email, str) or not email.strip():
|
||
raise AuthError("в токене нет email")
|
||
subject = payload["sub"]
|
||
if not isinstance(subject, str) or not subject.strip():
|
||
raise AuthError("в токене нет обязательного поля «sub»")
|
||
try:
|
||
exp = datetime.fromtimestamp(payload["exp"], tz=UTC)
|
||
except (TypeError, ValueError, OverflowError, OSError) as exc:
|
||
# нечисловой или запредельный exp, прошедший проверку подписи, не должен давать 500
|
||
raise AuthError("недействительный токен") from exc
|
||
return Claims(subject=subject, email=email.strip().lower(), exp=exp)
|
||
|
||
|
||
class DevHs256Verifier:
|
||
"""Проверяющий и эмитент dev-токенов: HS256 с общим секретом. Только для разработки (``DEBUG=true``)."""
|
||
|
||
def __init__(self, secret: str, *, email_claim: str = "email") -> None:
|
||
if not secret:
|
||
raise ValueError("секрет dev-эмитента не задан")
|
||
if len(secret.encode()) < DEV_SECRET_MIN_BYTES:
|
||
# HS256 по RFC 7518 §3.2 требует ключ не короче хэша (32 байта); PyJWT иначе шумит предупреждением
|
||
raise RuntimeError(f"AUTH_DEV_SECRET короче {DEV_SECRET_MIN_BYTES} байт")
|
||
self._secret = secret
|
||
self._email_claim = email_claim
|
||
|
||
async def verify(self, token: str) -> Claims:
|
||
try:
|
||
payload = jwt.decode(
|
||
token,
|
||
self._secret,
|
||
algorithms=[DEV_ALGORITHM],
|
||
options={"require": ["exp", "sub"], "verify_aud": False},
|
||
)
|
||
except jwt.PyJWTError as exc:
|
||
raise _to_auth_error(exc) from exc
|
||
return _claims_from_payload(payload, self._email_claim)
|
||
|
||
def issue_token(self, email: str, *, ttl: timedelta = DEV_TOKEN_TTL) -> str:
|
||
"""Выпускает HS256 JWT для ``POST /api/v1/dev/token``: ``sub`` = email, email в настроенном claim."""
|
||
now = datetime.now(UTC)
|
||
payload = {"sub": email, self._email_claim: email, "iat": now, "exp": now + ttl}
|
||
return jwt.encode(payload, self._secret, algorithm=DEV_ALGORITHM)
|
||
|
||
|
||
def _load_static_key(pem: str) -> tuple[Any, tuple[str, ...]]:
|
||
"""Загружает публичный ключ из PEM и выводит допустимый алгоритм из его типа (RSA → RS256, EC P-256 → ES256)."""
|
||
# в .env PEM удобно хранить одной строкой с литеральными «\n»
|
||
try:
|
||
key = serialization.load_pem_public_key(pem.replace("\\n", "\n").strip().encode())
|
||
except (ValueError, TypeError) as exc:
|
||
raise ValueError("AUTH_PUBLIC_KEY не является публичным ключом в формате PEM") from exc
|
||
if isinstance(key, rsa.RSAPublicKey):
|
||
return key, ("RS256",)
|
||
if isinstance(key, ec.EllipticCurvePublicKey) and key.curve.name == "secp256r1":
|
||
return key, ("ES256",)
|
||
raise ValueError("AUTH_PUBLIC_KEY: поддерживаются ключи RSA и EC P-256 (алгоритмы RS256 и ES256)")
|
||
|
||
|
||
class JwksVerifier:
|
||
"""Проверяющий токены РАИСА: RS256/ES256 по статическому публичному ключу либо по JWKS (кэш набора ключей).
|
||
|
||
``iss`` проверяется, только если задан ``issuer``; ``aud`` — только если задан ``audience``. ``exp`` и ``sub``
|
||
обязательны всегда. ``jwk_client`` — внедряемый источник ключей (тесты); по умолчанию :class:`jwt.PyJWKClient`.
|
||
"""
|
||
|
||
def __init__( # noqa: PLR0913 - все параметры именованные и соответствуют настройкам AUTH_*
|
||
self,
|
||
*,
|
||
jwks_url: str = "",
|
||
public_key: str = "",
|
||
issuer: str = "",
|
||
audience: str = "",
|
||
email_claim: str = "email",
|
||
leeway: int = JWKS_LEEWAY_SECONDS,
|
||
jwk_client: _SigningKeyProvider | None = None,
|
||
) -> None:
|
||
sources = [bool(jwks_url), bool(public_key), jwk_client is not None]
|
||
if sum(sources) == 0:
|
||
raise ValueError("не задан ключ проверки подписи: нужен AUTH_JWKS_URL или AUTH_PUBLIC_KEY")
|
||
if sum(sources) > 1:
|
||
raise ValueError("AUTH_JWKS_URL и AUTH_PUBLIC_KEY заданы одновременно: оставьте один источник ключа")
|
||
self._static_key: Any = None
|
||
self._static_algorithms: tuple[str, ...] = ()
|
||
self._client: _SigningKeyProvider | None = jwk_client
|
||
if public_key:
|
||
self._static_key, self._static_algorithms = _load_static_key(public_key)
|
||
elif jwks_url:
|
||
self._client = PyJWKClient(
|
||
jwks_url,
|
||
cache_keys=False,
|
||
cache_jwk_set=True,
|
||
lifespan=JWKS_CACHE_LIFESPAN_SECONDS,
|
||
timeout=JWKS_FETCH_TIMEOUT_SECONDS,
|
||
)
|
||
self._issuer = issuer or None
|
||
self._audience = audience or None
|
||
self._email_claim = email_claim
|
||
self._leeway = leeway
|
||
|
||
async def _signing_key(self, token: str) -> tuple[Any, list[str]]:
|
||
if self._client is None:
|
||
return self._static_key, list(self._static_algorithms)
|
||
# PyJWKClient ходит в сеть блокирующе — выносим из цикла событий
|
||
signing_key = await asyncio.to_thread(self._client.get_signing_key_from_jwt, token)
|
||
# Отдаём сам PyJWK, а не signing_key.key: PyJWT тогда сверяет алгоритм из заголовка токена с типом ключа
|
||
# (RSA-ключ с ES256 и наоборот — InvalidAlgorithmError, а не TypeError из недр библиотеки)
|
||
return signing_key, list(JWKS_ALGORITHMS)
|
||
|
||
async def verify(self, token: str) -> Claims:
|
||
try:
|
||
key, algorithms = await self._signing_key(token)
|
||
payload = jwt.decode(
|
||
token,
|
||
key,
|
||
algorithms=algorithms,
|
||
issuer=self._issuer,
|
||
audience=self._audience,
|
||
leeway=self._leeway,
|
||
options={"require": ["exp", "sub"], "verify_aud": self._audience is not None},
|
||
)
|
||
except jwt.PyJWTError as exc:
|
||
raise _to_auth_error(exc) from exc
|
||
except (TypeError, ValueError) as exc:
|
||
# страховка: несовместимые ключ и алгоритм не должны превращаться в 500
|
||
raise AuthError("недействительный токен") from exc
|
||
return _claims_from_payload(payload, self._email_claim)
|
||
|
||
|
||
def build_verifier(settings: Settings) -> TokenVerifier | None:
|
||
"""Собирает проверяющего токены по настройкам.
|
||
|
||
* ``AUTH_DEV_ISSUER`` **и** ``DEBUG`` → :class:`DevHs256Verifier` (только он; JWKS в этом режиме не используется);
|
||
пустой ``AUTH_DEV_SECRET`` — ошибка конфигурации (``RuntimeError``).
|
||
* иначе, если задан ``AUTH_JWKS_URL`` или ``AUTH_PUBLIC_KEY`` → :class:`JwksVerifier`.
|
||
* иначе ``None``: проверять нечем, токены не принимаются. ``AUTH_DEV_ISSUER`` без ``DEBUG`` dev-проверяющего
|
||
не включает никогда, даже при заданном секрете.
|
||
"""
|
||
if settings.AUTH_DEV_ISSUER and settings.DEBUG:
|
||
if not settings.AUTH_DEV_SECRET:
|
||
raise RuntimeError("AUTH_DEV_ISSUER=true и DEBUG=true, но AUTH_DEV_SECRET не задан")
|
||
logger.warning("Включён dev-эмитент токенов (AUTH_DEV_ISSUER и DEBUG): только для разработки")
|
||
return DevHs256Verifier(settings.AUTH_DEV_SECRET, email_claim=settings.AUTH_EMAIL_CLAIM)
|
||
if settings.AUTH_DEV_ISSUER:
|
||
logger.warning("AUTH_DEV_ISSUER=true проигнорирован: dev-эмитент работает только при DEBUG=true")
|
||
if settings.AUTH_JWKS_URL or settings.AUTH_PUBLIC_KEY:
|
||
return JwksVerifier(
|
||
jwks_url=settings.AUTH_JWKS_URL,
|
||
public_key=settings.AUTH_PUBLIC_KEY,
|
||
issuer=settings.AUTH_ISSUER,
|
||
audience=settings.AUTH_AUDIENCE,
|
||
email_claim=settings.AUTH_EMAIL_CLAIM,
|
||
)
|
||
return None
|