diff --git a/api/requirements.txt b/api/requirements.txt index f26f6d9..ba04acc 100644 --- a/api/requirements.txt +++ b/api/requirements.txt @@ -44,6 +44,7 @@ pydantic-extra-types==2.11.0 pydantic-settings==2.14.2 pydantic_core==2.46.4 Pygments==2.20.0 +PyJWT[crypto]==2.15.1 PyNaCl==1.5.0 python-dotenv==1.2.2 python-multipart==0.0.32 diff --git a/api/src/core/auth.py b/api/src/core/auth.py new file mode 100644 index 0000000..2f3e53b --- /dev/null +++ b/api/src/core/auth.py @@ -0,0 +1,250 @@ +"""Аутентификация: проверка 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-токена (требование спеки authn-authz) +DEV_TOKEN_TTL = timedelta(hours=8) +# Время жизни кэша набора ключей JWKS (PyJWKClient), секунды +JWKS_CACHE_LIFESPAN_SECONDS = 300 +JWKS_FETCH_TIMEOUT_SECONDS = 10 + + +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) -> Any: ... + + +# Порядок важен: более конкретные классы исключений 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) — деталь только в лог, не в ответ + logger.warning("Не удалось получить ключ проверки подписи JWT: {}", exc) + return AuthError("не удалось получить ключ проверки подписи токена") + return AuthError("недействительный токен") + + +def _claims_from_payload(payload: dict[str, Any], email_claim: str) -> Claims: + """Собирает :class:`Claims` из проверенной полезной нагрузки; ``exp`` и ``sub`` уже проверены на наличие.""" + email = payload.get(email_claim) + if not isinstance(email, str) or not email.strip(): + raise AuthError("в токене нет email") + return Claims( + subject=str(payload["sub"]), + email=email.strip(), + exp=datetime.fromtimestamp(payload["exp"], tz=UTC), + ) + + +class DevHs256Verifier: + """Проверяющий и эмитент dev-токенов: HS256 с общим секретом. Только для разработки (``DEBUG=true``).""" + + def __init__(self, secret: str, *, email_claim: str = "email") -> None: + if not secret: + raise ValueError("секрет dev-эмитента не задан") + 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=True, + 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) + return signing_key.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 + 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 diff --git a/api/tests/test_auth_core.py b/api/tests/test_auth_core.py new file mode 100644 index 0000000..ff86646 --- /dev/null +++ b/api/tests/test_auth_core.py @@ -0,0 +1,554 @@ +"""Ядро аутентификации: проверяющие токены и фабрика по настройкам (без БД, без внешней сети).""" + +import base64 +import hashlib +import hmac +import json +import os +import threading +from datetime import UTC, datetime, timedelta +from http.server import BaseHTTPRequestHandler, HTTPServer +from types import SimpleNamespace +from typing import Any + +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec, rsa +from jwt.exceptions import PyJWKClientConnectionError + +from src.core.auth import ( + AuthError, + Claims, + DevHs256Verifier, + JwksVerifier, + TokenExpiredError, + TokenSignatureError, + build_verifier, +) +from src.core.config import Settings + +DEV_SECRET = "dev-secret-for-unit-tests-0123456789abcdef" +OTHER_SECRET = "another-secret-for-unit-tests-0123456789" +EMAIL = "ka_rf@armka.local" + + +class _LogSpy: + """Подмена logger модуля auth: собирает сообщения, не пишет в stderr.""" + + def __init__(self) -> None: + self.messages: list[str] = [] + + def warning(self, message: str, *args: Any) -> None: + self.messages.append(message.format(*args)) + + +@pytest.fixture(autouse=True) +def log_messages(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """Перехватывает логи модуля auth (шум не попадает в вывод тестов) и отдаёт их для проверок.""" + spy = _LogSpy() + monkeypatch.setattr("src.core.auth.logger", spy) + return spy.messages + + +def _payload(**overrides: Any) -> dict[str, Any]: + now = datetime.now(UTC) + data: dict[str, Any] = {"sub": EMAIL, "email": EMAIL, "iat": now, "exp": now + timedelta(hours=1)} + data.update(overrides) + return {k: v for k, v in data.items() if v is not None} + + +def _pem_public(key: Any) -> str: + return ( + key.public_key() + .public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + .decode() + ) + + +def _pem_private(key: Any) -> str: + return key.private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption() + ).decode() + + +def _b64(data: bytes) -> bytes: + return base64.urlsafe_b64encode(data).rstrip(b"=") + + +def _forge_hs(payload: dict[str, Any], secret: bytes, alg: str = "HS256") -> str: + """Собирает HS-токен вручную (PyJWT не даст подписать HMAC асимметричным PEM — атаки подмены ключа).""" + header = _b64(json.dumps({"alg": alg, "typ": "JWT"}).encode()) + body = _b64(json.dumps(payload, default=lambda v: int(v.timestamp())).encode()) + digest = {"HS256": hashlib.sha256, "HS384": hashlib.sha384}[alg] + sig = _b64(hmac.new(secret, header + b"." + body, digest).digest()) + return (header + b"." + body + b"." + sig).decode() + + +def _unsigned(payload: dict[str, Any]) -> str: + header = _b64(json.dumps({"alg": "none", "typ": "JWT"}).encode()) + body = _b64(json.dumps(payload, default=lambda v: int(v.timestamp())).encode()) + return (header + b"." + body + b".").decode() + + +@pytest.fixture(scope="module") +def rsa_key() -> rsa.RSAPrivateKey: + return rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +@pytest.fixture(scope="module") +def other_rsa_key() -> rsa.RSAPrivateKey: + return rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +@pytest.fixture(scope="module") +def ec_key() -> ec.EllipticCurvePrivateKey: + return ec.generate_private_key(ec.SECP256R1()) + + +@pytest.fixture(scope="module") +def other_ec_key() -> ec.EllipticCurvePrivateKey: + return ec.generate_private_key(ec.SECP256R1()) + + +# --- DevHs256Verifier ------------------------------------------------------------------------------------------- + + +@pytest.fixture +def dev() -> DevHs256Verifier: + return DevHs256Verifier(DEV_SECRET) + + +async def test_dev_valid_token(dev: DevHs256Verifier) -> None: + exp = datetime.now(UTC) + timedelta(hours=2) + token = jwt.encode(_payload(exp=exp), DEV_SECRET, algorithm="HS256") + claims = await dev.verify(token) + assert isinstance(claims, Claims) + assert claims.subject == EMAIL + assert claims.email == EMAIL + assert claims.exp == exp.replace(microsecond=0) + assert claims.exp.tzinfo is not None + + +async def test_dev_expired(dev: DevHs256Verifier) -> None: + token = jwt.encode(_payload(exp=datetime.now(UTC) - timedelta(minutes=5)), DEV_SECRET, algorithm="HS256") + with pytest.raises(TokenExpiredError, match="срок действия"): + await dev.verify(token) + + +async def test_dev_foreign_secret(dev: DevHs256Verifier) -> None: + token = jwt.encode(_payload(), OTHER_SECRET, algorithm="HS256") + with pytest.raises(TokenSignatureError, match="подпись"): + await dev.verify(token) + + +async def test_dev_no_email(dev: DevHs256Verifier) -> None: + token = jwt.encode(_payload(email=None), DEV_SECRET, algorithm="HS256") + with pytest.raises(AuthError, match="нет email"): + await dev.verify(token) + + +async def test_dev_empty_email(dev: DevHs256Verifier) -> None: + token = jwt.encode(_payload(email=" "), DEV_SECRET, algorithm="HS256") + with pytest.raises(AuthError, match="нет email"): + await dev.verify(token) + + +async def test_dev_no_exp(dev: DevHs256Verifier) -> None: + token = jwt.encode(_payload(exp=None), DEV_SECRET, algorithm="HS256") + with pytest.raises(AuthError, match="exp"): + await dev.verify(token) + + +async def test_dev_no_sub(dev: DevHs256Verifier) -> None: + token = jwt.encode(_payload(sub=None), DEV_SECRET, algorithm="HS256") + with pytest.raises(AuthError, match="sub"): + await dev.verify(token) + + +async def test_dev_alg_none(dev: DevHs256Verifier) -> None: + with pytest.raises(AuthError): + await dev.verify(_unsigned(_payload())) + + +async def test_dev_alg_hs384(dev: DevHs256Verifier) -> None: + """Токен с корректной подписью, но другим алгоритмом (HS384) не принимается: алгоритм зафиксирован.""" + with pytest.raises(AuthError, match="алгоритм"): + await dev.verify(_forge_hs(_payload(), DEV_SECRET.encode(), alg="HS384")) + + +@pytest.mark.parametrize("token", ["", "abc", "a.b.c", "не-токен"]) +async def test_dev_garbage(dev: DevHs256Verifier, token: str) -> None: + with pytest.raises(AuthError): + await dev.verify(token) + + +async def test_dev_issue_token_roundtrip(dev: DevHs256Verifier) -> None: + before = datetime.now(UTC) + token = dev.issue_token(EMAIL) + claims = await dev.verify(token) + assert claims.subject == EMAIL + assert claims.email == EMAIL + assert timedelta(hours=8) - timedelta(seconds=5) <= claims.exp - before <= timedelta(hours=8, seconds=5) + header = jwt.get_unverified_header(token) + assert header["alg"] == "HS256" + assert "iat" in jwt.decode(token, options={"verify_signature": False}) + + +async def test_dev_issue_token_custom_ttl_and_claim() -> None: + dev = DevHs256Verifier(DEV_SECRET, email_claim="mail") + token = dev.issue_token(EMAIL, ttl=timedelta(minutes=10)) + assert jwt.decode(token, options={"verify_signature": False})["mail"] == EMAIL + claims = await dev.verify(token) + assert claims.email == EMAIL + assert claims.exp - datetime.now(UTC) < timedelta(minutes=11) + + +def test_dev_empty_secret_forbidden() -> None: + with pytest.raises(ValueError, match="секрет"): + DevHs256Verifier("") + + +# --- JwksVerifier (статический публичный ключ) ------------------------------------------------------------------- + + +@pytest.fixture(params=["rs256", "es256"]) +def asym( + request: pytest.FixtureRequest, + rsa_key: rsa.RSAPrivateKey, + other_rsa_key: rsa.RSAPrivateKey, + ec_key: ec.EllipticCurvePrivateKey, + other_ec_key: ec.EllipticCurvePrivateKey, +) -> SimpleNamespace: + """Пара алгоритм/ключи: подписывающий, публичный PEM для проверяющего и чужой подписывающий ключ.""" + if request.param == "rs256": + return SimpleNamespace( + alg="RS256", priv=_pem_private(rsa_key), pub=_pem_public(rsa_key), other=_pem_private(other_rsa_key) + ) + return SimpleNamespace( + alg="ES256", priv=_pem_private(ec_key), pub=_pem_public(ec_key), other=_pem_private(other_ec_key) + ) + + +async def test_jwks_static_valid(asym: SimpleNamespace) -> None: + exp = datetime.now(UTC) + timedelta(hours=1) + token = jwt.encode(_payload(exp=exp), asym.priv, algorithm=asym.alg) + claims = await JwksVerifier(public_key=asym.pub).verify(token) + assert (claims.subject, claims.email) == (EMAIL, EMAIL) + assert claims.exp == exp.replace(microsecond=0) + + +async def test_jwks_static_pem_with_escaped_newlines(asym: SimpleNamespace) -> None: + """В .env PEM удобно хранить одной строкой с литеральными \\n.""" + token = jwt.encode(_payload(), asym.priv, algorithm=asym.alg) + claims = await JwksVerifier(public_key=asym.pub.replace("\n", "\\n")).verify(token) + assert claims.email == EMAIL + + +async def test_jwks_expired(asym: SimpleNamespace) -> None: + token = jwt.encode(_payload(exp=datetime.now(UTC) - timedelta(minutes=5)), asym.priv, algorithm=asym.alg) + with pytest.raises(TokenExpiredError): + await JwksVerifier(public_key=asym.pub).verify(token) + + +async def test_jwks_leeway_tolerates_small_clock_skew(asym: SimpleNamespace) -> None: + token = jwt.encode(_payload(exp=datetime.now(UTC) - timedelta(seconds=3)), asym.priv, algorithm=asym.alg) + claims = await JwksVerifier(public_key=asym.pub).verify(token) + assert claims.email == EMAIL + + +async def test_jwks_foreign_key(asym: SimpleNamespace) -> None: + token = jwt.encode(_payload(), asym.other, algorithm=asym.alg) + with pytest.raises(TokenSignatureError): + await JwksVerifier(public_key=asym.pub).verify(token) + + +async def test_jwks_no_exp(asym: SimpleNamespace) -> None: + token = jwt.encode(_payload(exp=None), asym.priv, algorithm=asym.alg) + with pytest.raises(AuthError, match="exp"): + await JwksVerifier(public_key=asym.pub).verify(token) + + +async def test_jwks_no_sub(asym: SimpleNamespace) -> None: + token = jwt.encode(_payload(sub=None), asym.priv, algorithm=asym.alg) + with pytest.raises(AuthError, match="sub"): + await JwksVerifier(public_key=asym.pub).verify(token) + + +async def test_jwks_issuer_checked_when_configured(asym: SimpleNamespace) -> None: + verifier = JwksVerifier(public_key=asym.pub, issuer="https://sso.raisa") + ok = jwt.encode(_payload(iss="https://sso.raisa"), asym.priv, algorithm=asym.alg) + assert (await verifier.verify(ok)).email == EMAIL + for bad in (_payload(iss="https://evil"), _payload()): + with pytest.raises(AuthError): + await verifier.verify(jwt.encode(bad, asym.priv, algorithm=asym.alg)) + + +async def test_jwks_issuer_ignored_when_not_configured(asym: SimpleNamespace) -> None: + token = jwt.encode(_payload(iss="whatever"), asym.priv, algorithm=asym.alg) + assert (await JwksVerifier(public_key=asym.pub).verify(token)).email == EMAIL + + +async def test_jwks_audience_checked_when_configured(asym: SimpleNamespace) -> None: + verifier = JwksVerifier(public_key=asym.pub, audience="arm-core") + ok = jwt.encode(_payload(aud="arm-core"), asym.priv, algorithm=asym.alg) + assert (await verifier.verify(ok)).email == EMAIL + for bad in (_payload(aud="other"), _payload()): + with pytest.raises(AuthError): + await verifier.verify(jwt.encode(bad, asym.priv, algorithm=asym.alg)) + + +async def test_jwks_audience_in_token_ignored_when_not_configured(asym: SimpleNamespace) -> None: + token = jwt.encode(_payload(aud="anything"), asym.priv, algorithm=asym.alg) + assert (await JwksVerifier(public_key=asym.pub).verify(token)).email == EMAIL + + +async def test_jwks_custom_email_claim(asym: SimpleNamespace) -> None: + verifier = JwksVerifier(public_key=asym.pub, email_claim="preferred_username") + token = jwt.encode(_payload(preferred_username="ivan@rshb.ru"), asym.priv, algorithm=asym.alg) + assert (await verifier.verify(token)).email == "ivan@rshb.ru" + # claim «email» при настроенном другом имени не подходит + with pytest.raises(AuthError, match="нет email"): + await verifier.verify(jwt.encode(_payload(), asym.priv, algorithm=asym.alg)) + + +async def test_jwks_alg_none(asym: SimpleNamespace) -> None: + with pytest.raises(AuthError): + await JwksVerifier(public_key=asym.pub).verify(_unsigned(_payload())) + + +async def test_jwks_key_confusion_hs256_signed_with_public_key(asym: SimpleNamespace) -> None: + """Атака подмены алгоритма: HS256, подписанный байтами публичного ключа как секретом, не принимается.""" + for secret in (asym.pub.encode(), asym.pub.strip().encode()): + with pytest.raises(AuthError): + await JwksVerifier(public_key=asym.pub).verify(_forge_hs(_payload(), secret)) + + +async def test_jwks_rejects_dev_token(asym: SimpleNamespace) -> None: + token = jwt.encode(_payload(), DEV_SECRET, algorithm="HS256") + with pytest.raises(AuthError): + await JwksVerifier(public_key=asym.pub).verify(token) + + +async def test_jwks_rs256_key_does_not_verify_es256_token( + rsa_key: rsa.RSAPrivateKey, ec_key: ec.EllipticCurvePrivateKey +) -> None: + """Алгоритм выводится из типа ключа: RSA-ключ принимает только RS256, EC-ключ — только ES256.""" + es_token = jwt.encode(_payload(), _pem_private(ec_key), algorithm="ES256") + rs_token = jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256") + with pytest.raises(AuthError): + await JwksVerifier(public_key=_pem_public(rsa_key)).verify(es_token) + with pytest.raises(AuthError): + await JwksVerifier(public_key=_pem_public(ec_key)).verify(rs_token) + + +def test_jwks_requires_a_key_source() -> None: + with pytest.raises(ValueError, match="AUTH_JWKS_URL|ключ"): + JwksVerifier() + + +def test_jwks_bad_pem() -> None: + with pytest.raises(ValueError, match="PEM"): + JwksVerifier(public_key="not a pem") + + +def test_jwks_rejects_both_sources(rsa_key: rsa.RSAPrivateKey) -> None: + with pytest.raises(ValueError, match="одновременно"): + JwksVerifier(public_key=_pem_public(rsa_key), jwks_url="http://127.0.0.1:1/jwks") + + +# --- JwksVerifier (JWKS: заглушка клиента и реальный PyJWKClient через локальный HTTP-сервер) -------------------- + + +class _FakeJwkClient: + def __init__(self, key: Any = None, error: Exception | None = None) -> None: + self._key, self._error, self.calls = key, error, 0 + + def get_signing_key_from_jwt(self, token: str) -> Any: + self.calls += 1 + if self._error is not None: + raise self._error + return SimpleNamespace(key=self._key) + + +async def test_jwks_client_valid(rsa_key: rsa.RSAPrivateKey) -> None: + client = _FakeJwkClient(key=rsa_key.public_key()) + verifier = JwksVerifier(jwk_client=client) + token = jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256") + assert (await verifier.verify(token)).email == EMAIL + assert client.calls == 1 + + +async def test_jwks_client_foreign_key(rsa_key: rsa.RSAPrivateKey, other_rsa_key: rsa.RSAPrivateKey) -> None: + verifier = JwksVerifier(jwk_client=_FakeJwkClient(key=rsa_key.public_key())) + token = jwt.encode(_payload(), _pem_private(other_rsa_key), algorithm="RS256") + with pytest.raises(TokenSignatureError): + await verifier.verify(token) + + +async def test_jwks_client_es256(ec_key: ec.EllipticCurvePrivateKey) -> None: + verifier = JwksVerifier(jwk_client=_FakeJwkClient(key=ec_key.public_key())) + token = jwt.encode(_payload(), _pem_private(ec_key), algorithm="ES256") + assert (await verifier.verify(token)).email == EMAIL + + +async def test_jwks_client_key_confusion(rsa_key: rsa.RSAPrivateKey) -> None: + verifier = JwksVerifier(jwk_client=_FakeJwkClient(key=rsa_key.public_key())) + with pytest.raises(AuthError): + await verifier.verify(_forge_hs(_payload(), _pem_public(rsa_key).encode())) + + +async def test_jwks_client_failure_is_auth_error_without_internals(log_messages: list[str]) -> None: + secret_detail = "connection to http://internal-host:8443/keys refused" + verifier = JwksVerifier(jwk_client=_FakeJwkClient(error=PyJWKClientConnectionError(secret_detail))) + token = jwt.encode(_payload(), DEV_SECRET, algorithm="HS256") + with pytest.raises(AuthError) as exc: + await verifier.verify(token) + assert "internal-host" not in str(exc.value) + assert "ключ" in str(exc.value) + assert any("internal-host" in m for m in log_messages) # деталь — только в лог + + +class _JwksHandler(BaseHTTPRequestHandler): + body = b"" + + def do_GET(self) -> None: + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(self.body) + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 - сигнатура базового класса + return + + +async def test_jwks_real_pyjwkclient_over_loopback(rsa_key: rsa.RSAPrivateKey) -> None: + """Реальный PyJWKClient: выбор ключа по kid из JWKS, отдаваемого локальным сервером (только loopback).""" + jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(rsa_key.public_key())) + jwk.update(kid="key-1", use="sig", alg="RS256") + _JwksHandler.body = json.dumps({"keys": [jwk]}).encode() + server = HTTPServer(("127.0.0.1", 0), _JwksHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + url = f"http://127.0.0.1:{server.server_port}/jwks" + verifier = JwksVerifier(jwks_url=url) + good = jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256", headers={"kid": "key-1"}) + assert (await verifier.verify(good)).email == EMAIL + unknown_kid = jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256", headers={"kid": "nope"}) + with pytest.raises(AuthError, match="ключ"): + await verifier.verify(unknown_kid) + finally: + server.shutdown() + server.server_close() + + +# --- Фабрика -------------------------------------------------------------------------------------------------- + + +def _settings(monkeypatch: pytest.MonkeyPatch, **auth: Any) -> Settings: + """Settings без env-файла и без AUTH_*/DEBUG из окружения процесса; обязательные OPENBAO__* — заглушки.""" + for key in list(os.environ): + if key.startswith(("OPENBAO__", "AUTH_")) or key == "DEBUG": + monkeypatch.delenv(key) + required = { + "OPENBAO__DATABASE__DATABASE_URL": "postgresql+asyncpg://u:p@localhost:5432/d", + "OPENBAO__MINIO__MINIO_URL": "localhost:9000", + "OPENBAO__MINIO__MINIO_ADMIN_LOGIN": "login", + "OPENBAO__MINIO__MINIO_ADMIN_PASSWORD": "password", + "OPENBAO__MINIO__MINIO_BUCKET": "bucket", + "OPENBAO__NATS__NATS_URL": "nats://localhost:4222", + "OPENBAO__SMB__SMB_HOST": "host", + "OPENBAO__SMB__SMB_PATH": "path", + "OPENBAO__SMB__SMB_LOGIN": "login", + "OPENBAO__SMB__SMB_PASSWORD": "password", + } + for key, value in required.items(): + monkeypatch.setenv(key, value) + return Settings(_env_file=None, **auth) # type: ignore[call-arg] + + +def test_factory_dev_off_and_no_jwks_gives_none(monkeypatch: pytest.MonkeyPatch) -> None: + s = _settings(monkeypatch, AUTH_DEV_ISSUER=False, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=True) + assert build_verifier(s) is None + + +def test_factory_dev_flag_without_debug_gives_no_dev(monkeypatch: pytest.MonkeyPatch, log_messages: list[str]) -> None: + s = _settings(monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=False) + assert build_verifier(s) is None + assert any("AUTH_DEV_ISSUER=true проигнорирован" in m for m in log_messages) + + +def test_factory_dev_flag_without_debug_falls_back_to_jwks( + monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey +) -> None: + s = _settings( + monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=False, AUTH_PUBLIC_KEY=_pem_public(rsa_key) + ) + verifier = build_verifier(s) + assert isinstance(verifier, JwksVerifier) + assert not isinstance(verifier, DevHs256Verifier) + + +def test_factory_both_flags_gives_dev(monkeypatch: pytest.MonkeyPatch) -> None: + s = _settings(monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=True, AUTH_EMAIL_CLAIM="mail") + verifier = build_verifier(s) + assert isinstance(verifier, DevHs256Verifier) + token = verifier.issue_token(EMAIL) + assert jwt.decode(token, options={"verify_signature": False})["mail"] == EMAIL + + +def test_factory_dev_wins_over_jwks_when_both_flags( + monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey +) -> None: + s = _settings( + monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=True, AUTH_PUBLIC_KEY=_pem_public(rsa_key) + ) + assert isinstance(build_verifier(s), DevHs256Verifier) + + +def test_factory_dev_flags_without_secret_is_config_error(monkeypatch: pytest.MonkeyPatch) -> None: + s = _settings(monkeypatch, AUTH_DEV_ISSUER=True, DEBUG=True) + with pytest.raises(RuntimeError, match="AUTH_DEV_SECRET"): + build_verifier(s) + + +def test_factory_public_key_gives_jwks(monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey) -> None: + s = _settings(monkeypatch, AUTH_PUBLIC_KEY=_pem_public(rsa_key)) + assert isinstance(build_verifier(s), JwksVerifier) + + +def test_factory_jwks_url_gives_jwks(monkeypatch: pytest.MonkeyPatch) -> None: + s = _settings(monkeypatch, AUTH_JWKS_URL="https://sso.example/.well-known/jwks.json") + assert isinstance(build_verifier(s), JwksVerifier) + + +async def test_factory_passes_issuer_audience_and_email_claim( + monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey +) -> None: + s = _settings( + monkeypatch, + AUTH_PUBLIC_KEY=_pem_public(rsa_key), + AUTH_ISSUER="https://sso.raisa", + AUTH_AUDIENCE="arm-core", + AUTH_EMAIL_CLAIM="mail", + ) + verifier = build_verifier(s) + assert verifier is not None + priv = _pem_private(rsa_key) + ok = jwt.encode(_payload(iss="https://sso.raisa", aud="arm-core", mail=EMAIL), priv, algorithm="RS256") + assert (await verifier.verify(ok)).email == EMAIL + with pytest.raises(AuthError): + await verifier.verify(jwt.encode(_payload(mail=EMAIL), priv, algorithm="RS256")) + + +async def test_dev_token_rejected_when_debug_off(monkeypatch: pytest.MonkeyPatch, rsa_key: rsa.RSAPrivateKey) -> None: + """Токен, подписанный dev-секретом, не принимается ни пустой конфигурацией, ни JWKS-проверяющим (DEBUG=false).""" + token = DevHs256Verifier(DEV_SECRET).issue_token(EMAIL) + prod = _settings( + monkeypatch, AUTH_DEV_ISSUER=True, AUTH_DEV_SECRET=DEV_SECRET, DEBUG=False, AUTH_PUBLIC_KEY=_pem_public(rsa_key) + ) + verifier = build_verifier(prod) + assert verifier is not None + with pytest.raises(AuthError): + await verifier.verify(token)