From 5ed52a72391dbf4a2b8184f6840d1c9c291c4418 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sat, 3 Oct 2026 22:01:33 +0400 Subject: [PATCH] =?UTF-8?q?fix:=20JWKS-=D0=BF=D1=80=D0=BE=D0=B2=D0=B5?= =?UTF-8?q?=D1=80=D1=8F=D1=8E=D1=89=D0=B8=D0=B9=20=E2=80=94=20=D0=B0=D0=BB?= =?UTF-8?q?=D0=B3=D0=BE=D1=80=D0=B8=D1=82=D0=BC=20=D0=BF=D0=BE=20=D1=82?= =?UTF-8?q?=D0=B8=D0=BF=D1=83=20=D0=BA=D0=BB=D1=8E=D1=87=D0=B0,=20=D0=B1?= =?UTF-8?q?=D0=B5=D0=B7=20=D0=BA=D1=8D=D1=88=D0=B0=20=D0=BA=D0=BB=D1=8E?= =?UTF-8?q?=D1=87=D0=B5=D0=B9=20=D0=BF=D0=BE=20kid;=20email=20=D0=B2=20?= =?UTF-8?q?=D0=BD=D0=B8=D0=B6=D0=BD=D0=B8=D0=B9=20=D1=80=D0=B5=D0=B3=D0=B8?= =?UTF-8?q?=D1=81=D1=82=D1=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit I-1: в jwt.decode передаётся PyJWK (а не голый ключ), рассогласование alg и типа ключа даёт 401, а не TypeError/500; страховка на (TypeError, ValueError). I-2: cache_keys=False — кэш ключей по kid не имеет срока жизни и держал бы отозванный ключ; остаётся кэш набора JWKS на 300 с. R-17: Claims.email = strip().lower(). R-18: запредельный exp не даёт 500; пустой sub отвергается; kid из заголовка пишется в лог через repr и усекается. Co-Authored-By: Claude Fable 5.1 --- api/src/core/auth.py | 41 +++++--- api/tests/test_auth_core.py | 194 ++++++++++++++++++++++++++++++------ 2 files changed, 190 insertions(+), 45 deletions(-) diff --git a/api/src/core/auth.py b/api/src/core/auth.py index 2f3e53b..fe97015 100644 --- a/api/src/core/auth.py +++ b/api/src/core/auth.py @@ -38,9 +38,12 @@ DEV_ALGORITHM = "HS256" JWKS_LEEWAY_SECONDS = 10 # Срок жизни dev-токена (требование спеки authn-authz) DEV_TOKEN_TTL = timedelta(hours=8) -# Время жизни кэша набора ключей JWKS (PyJWKClient), секунды +# Время жизни кэша набора ключей JWKS (PyJWKClient), секунды. Кэш отдельных ключей по kid (cache_keys) выключен +# намеренно: он без срока жизни и оставил бы отозванный ключ действующим до перезапуска процесса. JWKS_CACHE_LIFESPAN_SECONDS = 300 JWKS_FETCH_TIMEOUT_SECONDS = 10 +# Предел длины подконтрольного клиенту текста (kid и т.п.) в строке лога +_LOG_TEXT_LIMIT = 200 class AuthError(Exception): @@ -75,7 +78,7 @@ class TokenVerifier(Protocol): class _SigningKeyProvider(Protocol): """Источник ключа проверки подписи по токену (интерфейс :class:`jwt.PyJWKClient`).""" - def get_signing_key_from_jwt(self, token: str) -> Any: ... + def get_signing_key_from_jwt(self, token: str) -> jwt.PyJWK: ... # Порядок важен: более конкретные классы исключений PyJWT — раньше общих @@ -99,21 +102,30 @@ def _to_auth_error(exc: jwt.PyJWTError) -> AuthError: return AuthError(f"в токене нет обязательного поля «{exc.claim}»") if isinstance(exc, jwt.PyJWKClientError): # сбой получения ключей (сеть, неизвестный kid) — деталь только в лог, не в ответ - logger.warning("Не удалось получить ключ проверки подписи JWT: {}", exc) + # сообщение может содержать 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`` уже проверены на наличие.""" + """Собирает :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") - return Claims( - subject=str(payload["sub"]), - email=email.strip(), - exp=datetime.fromtimestamp(payload["exp"], tz=UTC), - ) + 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: @@ -159,7 +171,7 @@ def _load_static_key(pem: str) -> tuple[Any, tuple[str, ...]]: class JwksVerifier: - """Проверяющий токены РАИСА: RS256/ES256 по статическому публичному ключу либо по JWKS (кэш ключей). + """Проверяющий токены РАИСА: RS256/ES256 по статическому публичному ключу либо по JWKS (кэш набора ключей). ``iss`` проверяется, только если задан ``issuer``; ``aud`` — только если задан ``audience``. ``exp`` и ``sub`` обязательны всегда. ``jwk_client`` — внедряемый источник ключей (тесты); по умолчанию :class:`jwt.PyJWKClient`. @@ -189,7 +201,7 @@ class JwksVerifier: elif jwks_url: self._client = PyJWKClient( jwks_url, - cache_keys=True, + cache_keys=False, cache_jwk_set=True, lifespan=JWKS_CACHE_LIFESPAN_SECONDS, timeout=JWKS_FETCH_TIMEOUT_SECONDS, @@ -204,7 +216,9 @@ class JwksVerifier: 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) + # Отдаём сам PyJWK, а не signing_key.key: PyJWT тогда сверяет алгоритм из заголовка токена с типом ключа + # (RSA-ключ с ES256 и наоборот — InvalidAlgorithmError, а не TypeError из недр библиотеки) + return signing_key, list(JWKS_ALGORITHMS) async def verify(self, token: str) -> Claims: try: @@ -220,6 +234,9 @@ class JwksVerifier: ) 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) diff --git a/api/tests/test_auth_core.py b/api/tests/test_auth_core.py index ff86646..2e98a96 100644 --- a/api/tests/test_auth_core.py +++ b/api/tests/test_auth_core.py @@ -6,6 +6,8 @@ import hmac import json import os import threading +import time +from collections.abc import Callable, Iterator from datetime import UTC, datetime, timedelta from http.server import BaseHTTPRequestHandler, HTTPServer from types import SimpleNamespace @@ -15,9 +17,11 @@ import jwt import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ec, rsa -from jwt.exceptions import PyJWKClientConnectionError +from jwt import PyJWKClient +from jwt.exceptions import PyJWKClientConnectionError, PyJWKClientError from src.core.auth import ( + JWKS_CACHE_LIFESPAN_SECONDS, AuthError, Claims, DevHs256Verifier, @@ -31,6 +35,8 @@ 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" +SET_CACHE_MAX_SECONDS = 300 # кэш набора ключей JWKS не дольше 5 минут +LOG_LINE_MAX = 400 class _LogSpy: @@ -209,6 +215,35 @@ def test_dev_empty_secret_forbidden() -> None: DevHs256Verifier("") +# --- Нормализация email и защита от запредельных claims (R-17, R-18) ------------------------------------------- + + +async def test_email_is_lowercased_and_stripped(dev: DevHs256Verifier) -> None: + token = jwt.encode(_payload(email=" Ka_RF@ARMKA.Local "), DEV_SECRET, algorithm="HS256") + assert (await dev.verify(token)).email == "ka_rf@armka.local" + + +async def test_email_is_lowercased_in_jwks(asym_pub_priv: tuple[str, str, str]) -> None: + alg, priv, pub = asym_pub_priv + token = jwt.encode(_payload(email="Ivan.Petrov@RSHB.ru"), priv, algorithm=alg) + assert (await JwksVerifier(public_key=pub).verify(token)).email == "ivan.petrov@rshb.ru" + + +@pytest.mark.parametrize("sub", ["", " "]) +async def test_empty_sub_rejected(dev: DevHs256Verifier, sub: str) -> None: + token = jwt.encode(_payload(sub=sub), DEV_SECRET, algorithm="HS256") + with pytest.raises(AuthError, match="sub"): + await dev.verify(token) + + +@pytest.mark.parametrize("exp", ["9999999999", 10**12, 10**30, -(10**12), 1.5e300, True, [1]]) +async def test_garbage_exp_is_auth_error_not_500(dev: DevHs256Verifier, exp: Any) -> None: + """Запредельный или нечисловой exp (даже с верной подписью) — AuthError, а не OverflowError/ValueError/TypeError.""" + token = jwt.encode(_payload(exp=exp), DEV_SECRET, algorithm="HS256") + with pytest.raises(AuthError): + await dev.verify(token) + + # --- JwksVerifier (статический публичный ключ) ------------------------------------------------------------------- @@ -230,6 +265,11 @@ def asym( ) +@pytest.fixture +def asym_pub_priv(asym: SimpleNamespace) -> tuple[str, str, str]: + return asym.alg, asym.priv, asym.pub + + 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) @@ -360,15 +400,27 @@ def test_jwks_rejects_both_sources(rsa_key: rsa.RSAPrivateKey) -> None: # --- 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 _jwk(public_key: Any, kid: str) -> dict[str, Any]: + """JWK (словарь) публичного ключа RSA или EC P-256.""" + algorithm = jwt.algorithms.RSAAlgorithm if isinstance(public_key, rsa.RSAPublicKey) else jwt.algorithms.ECAlgorithm + jwk = json.loads(algorithm.to_jwk(public_key)) + jwk.update(kid=kid, use="sig") + return jwk - def get_signing_key_from_jwt(self, token: str) -> Any: + +class _FakeJwkClient: + """Заглушка PyJWKClient: отдаёт PyJWK (как настоящий клиент), без сети.""" + + def __init__(self, key: Any = None, error: Exception | None = None) -> None: + self._jwk = jwt.PyJWK(_jwk(key, "k1")) if key is not None else None + self._error, self.calls = error, 0 + + def get_signing_key_from_jwt(self, token: str) -> jwt.PyJWK: self.calls += 1 if self._error is not None: raise self._error - return SimpleNamespace(key=self._key) + assert self._jwk is not None + return self._jwk async def test_jwks_client_valid(rsa_key: rsa.RSAPrivateKey) -> None: @@ -409,38 +461,114 @@ async def test_jwks_client_failure_is_auth_error_without_internals(log_messages: assert any("internal-host" in m for m in log_messages) # деталь — только в лог -class _JwksHandler(BaseHTTPRequestHandler): - body = b"" +class _JwksServer: + """Локальный JWKS-эндпоинт (только loopback): набор ключей можно менять между запросами.""" - 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 __init__(self, keys: list[dict[str, Any]]) -> None: + self.keys = keys + outer = self - def log_message(self, format: str, *args: Any) -> None: # noqa: A002 - сигнатура базового класса - return + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps({"keys": outer.keys}).encode()) + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 - сигнатура базового класса + return + + self._server = HTTPServer(("127.0.0.1", 0), Handler) + self.url = f"http://127.0.0.1:{self._server.server_port}/jwks" + threading.Thread(target=self._server.serve_forever, daemon=True).start() + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() -async def test_jwks_real_pyjwkclient_over_loopback(rsa_key: rsa.RSAPrivateKey) -> None: +@pytest.fixture +def jwks_server_factory() -> Iterator[Callable[[list[dict[str, Any]]], _JwksServer]]: + servers: list[_JwksServer] = [] + + def make(keys: list[dict[str, Any]]) -> _JwksServer: + servers.append(_JwksServer(keys)) + return servers[-1] + + yield make + for server in servers: + server.close() + + +def _forge(header: dict[str, Any], payload: dict[str, Any]) -> str: + """Токен с произвольным заголовком и фиктивной подписью (проверка алгоритма идёт раньше подписи).""" + body = json.dumps(payload, default=lambda v: int(v.timestamp())).encode() + return (_b64(json.dumps(header).encode()) + b"." + _b64(body) + b"." + _b64(b"x" * 64)).decode() + + +async def test_jwks_real_pyjwkclient_over_loopback( + rsa_key: rsa.RSAPrivateKey, jwks_server_factory: Callable[[list[dict[str, Any]]], _JwksServer] +) -> 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() + server = jwks_server_factory([_jwk(rsa_key.public_key(), "key-1")]) + verifier = JwksVerifier(jwks_url=server.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) + + +async def test_jwks_cross_typed_alg_and_kid_is_401_not_500( + rsa_key: rsa.RSAPrivateKey, + ec_key: ec.EllipticCurvePrivateKey, + jwks_server_factory: Callable[[list[dict[str, Any]]], _JwksServer], +) -> None: + """JWKS с RSA- и EC-ключом: алгоритм из заголовка не совпадает с типом ключа -> AuthError, а не TypeError.""" + server = jwks_server_factory([_jwk(rsa_key.public_key(), "r1"), _jwk(ec_key.public_key(), "e1")]) + verifier = JwksVerifier(jwks_url=server.url) + # штатные токены обоих типов проходят + assert (await verifier.verify(jwt.encode(_payload(), _pem_private(rsa_key), "RS256", headers={"kid": "r1"}))).email + assert (await verifier.verify(jwt.encode(_payload(), _pem_private(ec_key), "ES256", headers={"kid": "e1"}))).email + for alg, kid in (("ES256", "r1"), ("RS256", "e1")): + with pytest.raises(AuthError): + await verifier.verify(_forge({"alg": alg, "kid": kid, "typ": "JWT"}, _payload())) + + +async def test_jwks_revoked_key_is_rejected_after_set_cache_expiry( + rsa_key: rsa.RSAPrivateKey, + other_rsa_key: rsa.RSAPrivateKey, + jwks_server_factory: Callable[[list[dict[str, Any]]], _JwksServer], +) -> None: + """Ключ, исчезнувший из JWKS, перестаёт приниматься после истечения кэша набора (кэша отдельных ключей нет).""" + keep = _jwk(other_rsa_key.public_key(), "r2") + server = jwks_server_factory([_jwk(rsa_key.public_key(), "r1"), keep]) + verifier = JwksVerifier(jwks_url=server.url) + client = verifier._client + assert isinstance(client, PyJWKClient) + assert client.jwk_set_cache is not None + assert client.jwk_set_cache.lifespan == JWKS_CACHE_LIFESPAN_SECONDS + assert JWKS_CACHE_LIFESPAN_SECONDS <= SET_CACHE_MAX_SECONDS + token = jwt.encode(_payload(), _pem_private(rsa_key), "RS256", headers={"kid": "r1"}) + assert (await verifier.verify(token)).email == EMAIL + server.keys = [keep] # эмитент отозвал r1 + client.jwk_set_cache.lifespan = 0.05 # ускоренное «истечение» кэша набора + time.sleep(0.1) + with pytest.raises(AuthError, match="ключ"): + await verifier.verify(token) + + +async def test_jwks_kid_is_logged_escaped_and_truncated(log_messages: list[str]) -> None: + """kid приходит от клиента: в логе он экранирован (нет переводов строк) и усечён.""" + hostile = "x\nINJECTED " + "A" * 5000 + verifier = JwksVerifier( + jwk_client=_FakeJwkClient(error=PyJWKClientError(f'Unable to find a signing key that matches: "{hostile}"')) + ) + with pytest.raises(AuthError): + await verifier.verify(jwt.encode(_payload(), DEV_SECRET, algorithm="HS256")) + (message,) = log_messages + assert "\n" not in message + assert len(message) < LOG_LINE_MAX # --- Фабрика --------------------------------------------------------------------------------------------------