feat: ядро аутентификации (Claims, TokenVerifier, DevHs256Verifier, JwksVerifier, build_verifier)
Dev HS256 включается только при AUTH_DEV_ISSUER и DEBUG одновременно; РАИСА — RS256/ES256 по публичному ключу или JWKS с кэшем. PyJWT[crypto] добавлен в requirements.txt как прод-зависимость. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
7429240f01
commit
70b47a097d
@ -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
|
||||
|
||||
250
api/src/core/auth.py
Normal file
250
api/src/core/auth.py
Normal file
@ -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
|
||||
554
api/tests/test_auth_core.py
Normal file
554
api/tests/test_auth_core.py
Normal file
@ -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)
|
||||
Loading…
x
Reference in New Issue
Block a user