Merge pull request 'Зависимости под контур: PyJWT 2.8.0, cryptography 43.0.3 (#43)' (#50) from feat/43-pyjwt-2-8 into main
Reviewed-on: #50
This commit is contained in:
commit
fbbf3efc37
@ -7,7 +7,7 @@ pytest>=8.3
|
||||
pytest-asyncio>=0.24
|
||||
pytest-env>=1.1
|
||||
pytest-cov>=6.0
|
||||
PyJWT[crypto]>=2.9
|
||||
PyJWT[crypto]==2.8.0
|
||||
pydbml>=1.1
|
||||
jsonschema>=4.23
|
||||
types-PyYAML>=6.0
|
||||
|
||||
@ -13,7 +13,7 @@ attrs==26.1.0
|
||||
certifi==2025.11.12
|
||||
cffi==2.1.0
|
||||
click==8.4.2
|
||||
cryptography==50.0.2
|
||||
cryptography==43.0.3
|
||||
dnspython==2.8.0
|
||||
email-validator==2.3.0
|
||||
et_xmlfile==2.0.0
|
||||
@ -47,7 +47,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
|
||||
PyJWT[crypto]==2.8.0
|
||||
PyNaCl==1.5.0
|
||||
python-dotenv==1.2.2
|
||||
python-multipart==0.0.32
|
||||
|
||||
@ -31,8 +31,6 @@ 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
|
||||
@ -152,6 +150,9 @@ class DevHs256Verifier:
|
||||
)
|
||||
except jwt.PyJWTError as exc:
|
||||
raise _to_auth_error(exc) from exc
|
||||
except (TypeError, ValueError) as exc:
|
||||
# PyJWT 2.8 не оборачивает нечисловой exp (например, список) в InvalidTokenError: страховка от 500
|
||||
raise AuthError("недействительный токен") from exc
|
||||
return _claims_from_payload(payload, self._email_claim)
|
||||
|
||||
def issue_token(self, email: str, *, ttl: timedelta = DEV_TOKEN_TTL) -> str:
|
||||
@ -168,11 +169,19 @@ def _load_static_key(pem: str) -> tuple[Any, tuple[str, ...]]:
|
||||
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
|
||||
algorithms = _algorithms_for_key(key)
|
||||
if algorithms is None:
|
||||
raise ValueError("AUTH_PUBLIC_KEY: поддерживаются ключи RSA и EC P-256 (алгоритмы RS256 и ES256)")
|
||||
return key, algorithms
|
||||
|
||||
|
||||
def _algorithms_for_key(key: Any) -> tuple[str, ...] | None:
|
||||
"""Допустимые алгоритмы подписи по типу открытого ключа: RSA → RS256, EC P-256 → ES256; иначе ``None``."""
|
||||
if isinstance(key, rsa.RSAPublicKey):
|
||||
return key, ("RS256",)
|
||||
return ("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)")
|
||||
return ("ES256",)
|
||||
return None
|
||||
|
||||
|
||||
class JwksVerifier:
|
||||
@ -221,9 +230,14 @@ 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)
|
||||
# Отдаём сам PyJWK, а не signing_key.key: PyJWT тогда сверяет алгоритм из заголовка токена с типом ключа
|
||||
# (RSA-ключ с ES256 и наоборот — InvalidAlgorithmError, а не TypeError из недр библиотеки)
|
||||
return signing_key, list(JWKS_ALGORITHMS)
|
||||
# Передаём signing_key.key, а не сам PyJWK: PyJWT < 2.10 (в контуре 2.8.0) не принимает PyJWK в decode
|
||||
# (TypeError «Expecting a PEM-formatted key»). Допустимый алгоритм выводим из типа ключа сами —
|
||||
# RSA-ключ не принимает ES256-токен и наоборот (InvalidAlgorithmError → 401, а не 500)
|
||||
key = signing_key.key
|
||||
algorithms = _algorithms_for_key(key)
|
||||
if algorithms is None:
|
||||
raise jwt.InvalidAlgorithmError("тип ключа JWKS не поддерживается")
|
||||
return key, list(algorithms)
|
||||
|
||||
async def verify(self, token: str) -> Claims:
|
||||
try:
|
||||
|
||||
@ -455,6 +455,18 @@ async def test_jwks_client_es256(ec_key: ec.EllipticCurvePrivateKey) -> None:
|
||||
assert (await verifier.verify(token)).email == EMAIL
|
||||
|
||||
|
||||
async def test_jwks_client_rsa_key_rejects_es256_token(
|
||||
rsa_key: rsa.RSAPrivateKey, ec_key: ec.EllipticCurvePrivateKey
|
||||
) -> None:
|
||||
"""RSA-ключ из JWKS и ES256-токен — AuthError (401), а не TypeError/500; и наоборот."""
|
||||
verifier = JwksVerifier(jwk_client=_FakeJwkClient(key=rsa_key.public_key()))
|
||||
with pytest.raises(AuthError, match="недопустимый алгоритм"):
|
||||
await verifier.verify(jwt.encode(_payload(), _pem_private(ec_key), algorithm="ES256"))
|
||||
verifier = JwksVerifier(jwk_client=_FakeJwkClient(key=ec_key.public_key()))
|
||||
with pytest.raises(AuthError, match="недопустимый алгоритм"):
|
||||
await verifier.verify(jwt.encode(_payload(), _pem_private(rsa_key), algorithm="RS256"))
|
||||
|
||||
|
||||
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):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user