fix: JWKS-проверяющий — алгоритм по типу ключа, без кэша ключей по kid; email в нижний регистр
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 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
d7469f31bc
commit
5ed52a7239
@ -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)
|
||||
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
# --- Фабрика --------------------------------------------------------------------------------------------------
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user