From 95791b1759b8aa4f2203575d344d5c78564cdc81 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Padilla?= Date: Thu, 21 May 2026 14:11:10 -0400 Subject: [PATCH] Bundle security fixes and hardening into 2.13.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Security: - `HMACAlgorithm.prepare_key` rejects JWK JSON documents passed as raw HMAC secrets to close an algorithm-confusion gap not covered by the existing PEM/SSH guard. Reported by @aradona91 in GHSA-xgmm-8j9v-c9wx. - Bind the JWT header `alg` to `PyJWK.algorithm_name` during verification so the caller's `algorithms` allow-list cannot be bypassed when decoding with a `PyJWK` / `PyJWKClient` key. Reported by @sushi-gif in GHSA-jq35-7prp-9v3f. - Skip the unconditional base64 decode of the compact-form payload segment when `b64=false` is set, and require that segment to be empty (RFC 7515 Appendix F detached form). Closes an unauthenticated DoS amplifier. Reported by @thesmartshadow in GHSA-w7vc-732c-9m39. - `PyJWKClient` rejects any URI whose scheme is not `http` or `https` so attacker-influenced URIs cannot read local files or reach unintended schemes via urllib's default `file://` / `ftp://` / `data:` handlers. Reported by @KEIJOT in GHSA-993g-76c3-p5m4. - Preserve the cached JWK Set on fetch errors in `PyJWKClient.fetch_data`. The previous `finally`-block `put(None)` pattern cleared the cache on any transient outage. Reported by @eddieran in GHSA-fhv5-28vv-h8m8. Fixes: - Reject empty HMAC keys outright in `HMACAlgorithm.prepare_key` with `InvalidKeyError` instead of accepting them with only a warning. Hardening prompted by reports from @SnailSploit and @spartan8806. - Forward per-call `options` (including `enforce_minimum_key_length`) from `PyJWT.decode` through to `PyJWS._verify_signature`. Thanks to @WLUB. - RFC 7797 §3 compliance for `b64=false`: encoder auto-adds `"b64"` to `crit`; decoder rejects tokens that set `b64=false` without listing it in `crit`. Thanks to @MachineLearning-Nerd. --- CHANGELOG.rst | 46 ++++++++++++- jwt/__init__.py | 2 +- jwt/algorithms.py | 23 +++++++ jwt/api_jws.py | 66 ++++++++++++++++-- jwt/api_jwt.py | 3 + jwt/jwks_client.py | 25 +++++-- tests/test_algorithms.py | 68 +++++++++++++++++-- tests/test_api_jws.py | 136 +++++++++++++++++++++++++++++++++++--- tests/test_jwks_client.py | 46 ++++++++++++- 9 files changed, 385 insertions(+), 30 deletions(-) diff --git a/jwt/algorithms.py b/jwt/algorithms.py index 615dcf38..20444bf3 100644 --- a/jwt/algorithms.py +++ b/jwt/algorithms.py @@ -325,12 +325,35 @@ def __init__(self, hash_alg: HashlibHash) -> None: def prepare_key(self, key: str | bytes) -> bytes: key_bytes = force_bytes(key) + if len(key_bytes) == 0: + raise InvalidKeyError("HMAC key must not be empty.") + if is_pem_format(key_bytes) or is_ssh_key(key_bytes): raise InvalidKeyError( "The specified key is an asymmetric key or x509 certificate and" " should not be used as an HMAC secret." ) + # Defense against algorithm-confusion attacks: an attacker with + # control over the token header can force this code path by setting + # alg=HS*, and HMACAlgorithm is the only algorithm that accepts + # arbitrary bytes as a valid secret. Other algorithms reject + # non-key-shaped input naturally. Even a symmetric (kty=oct) JWK + # should be loaded via PyJWK / from_jwk rather than fed as raw JSON + # bytes (whose contents are not the secret material). + stripped = key_bytes.lstrip() + if stripped.startswith(b"{"): + try: + jwk_obj = json.loads(key_bytes) + except ValueError: + jwk_obj = None + if isinstance(jwk_obj, dict) and "kty" in jwk_obj: + raise InvalidKeyError( + "The specified key looks like a JWK and should not be " + "used directly as an HMAC secret. Load it via " + "PyJWK / HMACAlgorithm.from_jwk first." + ) + return key_bytes @overload diff --git a/jwt/api_jws.py b/jwt/api_jws.py index 0ab7e4b4..236cf265 100644 --- a/jwt/api_jws.py +++ b/jwt/api_jws.py @@ -165,6 +165,17 @@ def encode( if is_payload_detached: header["b64"] = False + # RFC 7797 §3: producers MUST list "b64" in "crit" whenever + # "b64" appears in the protected header, so b64-unaware + # verifiers don't silently treat an unencoded payload as + # base64-encoded. + existing_crit = header.get("crit", []) + if not isinstance(existing_crit, list): + raise InvalidTokenError( + "Invalid 'crit' header: must be a list" + ) + if "b64" not in existing_crit: + header["crit"] = [*existing_crit, "b64"] elif "b64" in header: # True is the standard value for b64, so no need for it del header["b64"] @@ -242,6 +253,15 @@ def decode_complete( self._validate_headers(header) if header.get("b64", True) is False: + # RFC 7797 §3: when "b64" is present in the protected header, + # it MUST also appear in "crit". A token that sets b64=false + # without declaring it critical is malformed. + crit = header.get("crit") or [] + if not isinstance(crit, list) or "b64" not in crit: + raise InvalidTokenError( + "The 'b64' header parameter requires 'b64' to be " + "listed in 'crit'." + ) if detached_payload is None: raise DecodeError( 'It is required that you pass in a value for the "detached_payload" argument to decode a message having the b64 header set to false.' @@ -250,7 +270,14 @@ def decode_complete( signing_input = b".".join([signing_input.rsplit(b".", 1)[0], payload]) if verify_signature: - self._verify_signature(signing_input, header, signature, key, algorithms) + self._verify_signature( + signing_input, + header, + signature, + key, + algorithms, + options=merged_options, + ) return { "payload": payload, @@ -317,10 +344,24 @@ def _load(self, jwt: str | bytes) -> tuple[bytes, bytes, dict[str, Any], bytes]: if not isinstance(header, dict): raise DecodeError("Invalid header string: must be a json object") - try: - payload = base64url_decode(payload_segment) - except (TypeError, binascii.Error) as err: - raise DecodeError("Invalid payload padding") from err + if header.get("b64", True) is False: + # Detached payload form (RFC 7515 Appendix F): the compact-form + # payload segment must be empty; the caller supplies the actual + # payload via the `detached_payload` argument in decode_complete. + # Skipping the base64 decode here removes an unauthenticated work + # amplifier — otherwise an attacker can inflate the unused + # segment to force CPU + memory cost before the signature is + # even checked. + if payload_segment: + raise DecodeError( + "Payload segment must be empty when 'b64' is false." + ) + payload = b"" + else: + try: + payload = base64url_decode(payload_segment) + except (TypeError, binascii.Error) as err: + raise DecodeError("Invalid payload padding") from err try: signature = base64url_decode(crypto_segment) @@ -336,7 +377,10 @@ def _verify_signature( signature: bytes, key: AllowedPublicKeys | PyJWK | str | bytes = "", algorithms: Sequence[str] | None = None, + options: SigOptions | None = None, ) -> None: + effective_options = options if options is not None else self.options + if algorithms is None and isinstance(key, PyJWK): algorithms = [key.algorithm_name] try: @@ -348,6 +392,16 @@ def _verify_signature( raise InvalidAlgorithmError("The specified alg value is not allowed") if isinstance(key, PyJWK): + # The PyJWK has a fixed algorithm bound at construction time. + # Verification must use that algorithm, not whatever the token + # header advertises, otherwise the caller's allow-list check + # above degenerates into a string compare with no behavioural + # effect on which algorithm actually verifies the signature. + if alg != key.algorithm_name: + raise InvalidAlgorithmError( + f"Token algorithm {alg!r} does not match the key's " + f"algorithm {key.algorithm_name!r}" + ) alg_obj = key.Algorithm prepared_key = key.key else: @@ -359,7 +413,7 @@ def _verify_signature( key_length_msg = alg_obj.check_key_length(prepared_key) if key_length_msg: - if self.options.get("enforce_minimum_key_length", False): + if effective_options.get("enforce_minimum_key_length", False): raise InvalidKeyError(key_length_msg) else: warnings.warn(key_length_msg, InsecureKeyLengthWarning, stacklevel=4) diff --git a/jwt/api_jwt.py b/jwt/api_jwt.py index 429c2d79..ad65cb8b 100644 --- a/jwt/api_jwt.py +++ b/jwt/api_jwt.py @@ -258,6 +258,9 @@ def decode_complete( sig_options: SigOptions = { "verify_signature": verify_signature, + "enforce_minimum_key_length": merged_options.get( + "enforce_minimum_key_length", False + ), } decoded = self._jws.decode_complete( jwt, diff --git a/jwt/jwks_client.py b/jwt/jwks_client.py index b81e8f4b..41f333a9 100644 --- a/jwt/jwks_client.py +++ b/jwt/jwks_client.py @@ -6,6 +6,7 @@ from ssl import SSLContext from typing import Any from urllib.error import HTTPError, URLError +from urllib.parse import urlparse from .api_jwk import PyJWK, PyJWKSet from .api_jwt import decode_complete as decode_token @@ -69,6 +70,16 @@ def __init__( """ if headers is None: headers = {} + # urllib's default OpenerDirector also handles file://, ftp://, and + # data: URIs. Reject anything that isn't http(s) eagerly so a caller + # passing an attacker-influenced URL (e.g. taken from a `jku` token + # header) can't read local files or reach other unintended schemes. + scheme = urlparse(uri).scheme.lower() + if scheme not in ("http", "https"): + raise PyJWKClientError( + f"Invalid JWKS URI scheme {scheme!r}: only 'http' and 'https' " + f"are supported." + ) self.uri = uri self.jwk_set_cache: JWKSetCache | None = None self.headers = headers @@ -102,7 +113,6 @@ def fetch_data(self) -> Any: :returns: The parsed JWK Set as a dictionary. :raises PyJWKClientConnectionError: If the HTTP request fails. """ - jwk_set: Any = None try: r = urllib.request.Request(url=self.uri, headers=self.headers) with urllib.request.urlopen( @@ -115,11 +125,14 @@ def fetch_data(self) -> Any: raise PyJWKClientConnectionError( f'Fail to fetch data from the url, err: "{e}"' ) from e - else: - return jwk_set - finally: - if self.jwk_set_cache is not None: - self.jwk_set_cache.put(jwk_set) + + # Only update the cache on a successful fetch. Writing in a + # `finally` block with `jwk_set=None` on error clears any + # previously-cached JWKS, turning a transient outage into a cache + # wipe that breaks legitimate auth. + if self.jwk_set_cache is not None: + self.jwk_set_cache.put(jwk_set) + return jwk_set def get_jwk_set(self, refresh: bool = False) -> PyJWKSet: """Return the JWK Set, using the cache when available. diff --git a/tests/test_algorithms.py b/tests/test_algorithms.py index 5449f3c8..a1c902b0 100644 --- a/tests/test_algorithms.py +++ b/tests/test_algorithms.py @@ -122,6 +122,38 @@ def test_hmac_from_jwk_should_raise_exception_if_empty_json(self) -> None: with pytest.raises(InvalidKeyError): algo.from_jwk(keyfile.read()) + @pytest.mark.parametrize("empty_key", ["", b""]) + def test_hmac_prepare_key_rejects_empty_key( + self, empty_key: Union[str, bytes] + ) -> None: + algo = HMACAlgorithm(HMACAlgorithm.SHA256) + + with pytest.raises(InvalidKeyError, match="must not be empty"): + algo.prepare_key(empty_key) + + @pytest.mark.parametrize( + "jwk_file", + [ + "jwk_rsa_pub.json", + "jwk_ec_pub_P-256.json", + "jwk_okp_pub_Ed25519.json", + "jwk_hmac.json", + ], + ) + def test_hmac_prepare_key_rejects_jwk_json(self, jwk_file: str) -> None: + algo = HMACAlgorithm(HMACAlgorithm.SHA256) + + with open(key_path(jwk_file)) as keyfile: + with pytest.raises(InvalidKeyError, match="looks like a JWK"): + algo.prepare_key(keyfile.read()) + + def test_hmac_prepare_key_accepts_json_without_kty(self) -> None: + # JSON that doesn't look like a JWK (no "kty") should not be misclassified. + algo = HMACAlgorithm(HMACAlgorithm.SHA256) + + key = algo.prepare_key('{"this": "is just a json-shaped secret"}') + assert key == b'{"this": "is just a json-shaped secret"}' + @crypto_required def test_rsa_should_parse_pem_public_key(self) -> None: algo = RSAAlgorithm(RSAAlgorithm.SHA256) @@ -1452,11 +1484,10 @@ def test_hmac_short_key_warns_by_default_hs512(self) -> None: assert msg is not None assert "64" in msg - def test_hmac_empty_key_returns_warning_message(self) -> None: + def test_hmac_empty_key_rejected_outright(self) -> None: algo = HMACAlgorithm(HMACAlgorithm.SHA256) - key = algo.prepare_key(b"") - msg = algo.check_key_length(key) - assert msg is not None + with pytest.raises(InvalidKeyError, match="must not be empty"): + algo.prepare_key(b"") def test_hmac_exact_minimum_no_warning(self) -> None: algo = HMACAlgorithm(HMACAlgorithm.SHA256) @@ -1603,6 +1634,35 @@ def test_pyjwt_decode_enforces_short_hmac_key(self) -> None: with pytest.raises(InvalidKeyError): pyjwt_enforce.decode(token, "short", algorithms=["HS256"]) + def test_pyjwt_decode_honors_per_call_enforce_minimum_key_length(self) -> None: + # Regression: per-call options passed to PyJWT.decode() must be + # forwarded to the JWS verification layer. enforce_minimum_key_length + # was previously dropped between PyJWT.decode_complete and + # PyJWS._verify_signature. + import jwt + + adequate_key = "a" * 32 + token = jwt.encode({"hello": "world"}, adequate_key, algorithm="HS256") + + # Module-level singleton path with per-call enforcement. + with pytest.raises(InvalidKeyError, match="below"): + jwt.decode( + token, + "short", + algorithms=["HS256"], + options={"enforce_minimum_key_length": True}, + ) + + # PyJWT() instance path (instance default off, per-call on). + pyjwt = jwt.PyJWT() + with pytest.raises(InvalidKeyError, match="below"): + pyjwt.decode( + token, + "short", + algorithms=["HS256"], + options={"enforce_minimum_key_length": True}, + ) + def test_pyjwt_encode_no_warning_adequate_key(self) -> None: import warnings diff --git a/tests/test_api_jws.py b/tests/test_api_jws.py index 9f7edc04..e55742c4 100644 --- a/tests/test_api_jws.py +++ b/tests/test_api_jws.py @@ -9,10 +9,11 @@ from jwt.exceptions import ( DecodeError, InvalidAlgorithmError, + InvalidKeyError, InvalidSignatureError, InvalidTokenError, ) -from jwt.utils import base64url_decode +from jwt.utils import base64url_decode, base64url_encode from jwt.warnings import RemovedInPyjwt3Warning from .utils import crypto_required, key_path, no_crypto_required @@ -397,6 +398,33 @@ def test_decodes_with_jwk_and_mismatched_algorithm( with pytest.raises(InvalidAlgorithmError): jws.decode(example_jws, jwk) + def test_decodes_with_jwk_rejects_header_alg_outside_jwk_alg( + self, jws: PyJWS + ) -> None: + # Token header says HS256 and the caller's allow-list also accepts + # HS256, but the PyJWK is bound to HS512. Even though the allow-list + # would pass, verification must be locked to the PyJWK's algorithm + # rather than the header's — otherwise an attacker who controls a + # registered key can advertise a disallowed algorithm in the header + # and have it accepted. + jwk = PyJWK( + { + "kty": "oct", + "alg": "HS512", + "k": "c2VjcmV0", # "secret" + } + ) + example_jws = ( + b"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9." + b"aGVsbG8gd29ybGQ." + b"gEW0pdU4kxPthjtehYdhxB9mMOGajt1xCKlGGXDJ8PM" + ) + + with pytest.raises( + InvalidAlgorithmError, match="does not match the key's algorithm" + ): + jws.decode(example_jws, jwk, algorithms=["HS256", "HS512"]) + # 'Control' Elliptic Curve jws created by another library. # Used to test for regressions that could affect both # encoding / decoding operations equally (causing tests @@ -511,18 +539,16 @@ def test_no_secret(self, jws: PyJWS, payload: bytes) -> None: right_secret = "foo" jws_message = jws.encode(payload, right_secret) - with pytest.raises(DecodeError): + with pytest.raises(InvalidKeyError, match="must not be empty"): jws.decode(jws_message, algorithms=["HS256"]) def test_verify_signature_with_no_secret(self, jws: PyJWS, payload: bytes) -> None: right_secret = "foo" jws_message = jws.encode(payload, right_secret) - with pytest.raises(DecodeError) as exc: + with pytest.raises(InvalidKeyError, match="must not be empty"): jws.decode(jws_message, algorithms=["HS256"]) - assert "Signature verification" in str(exc.value) - def test_verify_signature_with_no_algo_header_throws_exception( self, jws: PyJWS, payload: bytes ) -> None: @@ -957,13 +983,103 @@ def test_encode_detached_content_with_b64_header( assert "b64" not in msg_header_obj assert msg_payload - def test_decode_detached_content_without_proper_argument(self, jws: PyJWS) -> None: - example_jws = ( - "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiIsImI2NCI6ZmFsc2V9" - "." - ".65yNkX_ZH4A_6pHaTL_eI84OXOHtfl4K0k5UnlXZ8f4" + def test_encode_b64_false_auto_adds_b64_to_crit( + self, jws: PyJWS, payload: bytes + ) -> None: + # RFC 7797 §3: producers MUST list "b64" in "crit" whenever "b64" + # appears in the protected header. + secret = "secret" + token = jws.encode( + payload, secret, algorithm="HS256", is_payload_detached=True + ) + + msg_header, _, _ = token.split(".") + header_obj = json.loads(base64url_decode(msg_header.encode())) + + assert header_obj["b64"] is False + assert "b64" in header_obj.get("crit", []) + + def test_encode_b64_false_preserves_existing_crit_entries( + self, jws: PyJWS, payload: bytes + ) -> None: + secret = "secret" + # Caller-supplied crit (containing a hypothetical extension that + # PyJWT does support) should be preserved alongside the auto-added + # "b64" marker. + token = jws.encode( + payload, + secret, + algorithm="HS256", + headers={"b64": False, "crit": ["b64"]}, ) + + header_obj = json.loads(base64url_decode(token.split(".")[0].encode())) + assert header_obj["crit"] == ["b64"] + + def test_decode_b64_false_rejects_non_empty_payload_segment( + self, jws: PyJWS, payload: bytes + ) -> None: + # RFC 7515 Appendix F detached form: when b64=false, the compact- + # serialization payload segment must be empty. PyJWT must reject a + # non-empty middle segment without doing any base64-decoding work + # on it — that decode used to be the unauthenticated DoS amplifier. + secret = "secret" + import hmac as _hmac + import hashlib as _hashlib + + header_obj = { + "typ": "JWT", + "alg": "HS256", + "b64": False, + "crit": ["b64"], + } + header_b64 = base64url_encode( + json.dumps(header_obj, separators=(",", ":")).encode() + ) + # Stuff the middle segment with arbitrary attacker-controlled bytes. + # This should be rejected without being base64-decoded. + attacker_segment = b"A" * 1024 + signing_input = b".".join([header_b64, payload]) + sig = _hmac.new(secret.encode(), signing_input, _hashlib.sha256).digest() + token = b".".join( + [header_b64, attacker_segment, base64url_encode(sig)] + ).decode() + + with pytest.raises(DecodeError, match="Payload segment must be empty"): + jws.decode( + token, secret, algorithms=["HS256"], detached_payload=payload + ) + + def test_decode_b64_false_without_crit_b64_is_rejected( + self, jws: PyJWS, payload: bytes + ) -> None: + # Hand-craft a non-compliant token: header has b64=false but no + # crit:["b64"]. Per RFC 7797 §3, such tokens are malformed and must + # be rejected even though PyJWT understands b64. + secret = "secret" + import hmac as _hmac + import hashlib as _hashlib + + header_obj = {"typ": "JWT", "alg": "HS256", "b64": False} + header_b64 = base64url_encode( + json.dumps(header_obj, separators=(",", ":")).encode() + ) + signing_input = b".".join([header_b64, payload]) + sig = _hmac.new(secret.encode(), signing_input, _hashlib.sha256).digest() + token = b".".join([header_b64, b"", base64url_encode(sig)]).decode() + + with pytest.raises(InvalidTokenError, match="b64.*crit"): + jws.decode( + token, secret, algorithms=["HS256"], detached_payload=payload + ) + + def test_decode_detached_content_without_proper_argument( + self, jws: PyJWS, payload: bytes + ) -> None: example_secret = "secret" + example_jws = jws.encode( + payload, example_secret, algorithm="HS256", is_payload_detached=True + ) with pytest.raises(DecodeError) as exc: jws.decode(example_jws, example_secret, algorithms=["HS256"]) diff --git a/tests/test_jwks_client.py b/tests/test_jwks_client.py index ceee672f..da686645 100644 --- a/tests/test_jwks_client.py +++ b/tests/test_jwks_client.py @@ -288,18 +288,30 @@ def test_get_jwt_set_cache_disabled(self) -> None: assert repeated_call.call_count == 1 - def test_get_jwt_set_failed_request_should_clear_cache(self) -> None: + def test_get_jwt_set_failed_refresh_preserves_cached_jwks(self) -> None: + # Regression: a transient fetch failure used to clear the cache via + # the previous `finally: put(jwk_set=None)` pattern, turning one bad + # request from the JWKS endpoint into application-wide auth failure. + # The cache must survive. url = "https://dev-87evx9ru.auth0.com/.well-known/jwks.json" jwks_client = PyJWKClient(url) with mocked_success_response(RESPONSE_DATA_WITH_MATCHING_KID): jwks_client.get_jwk_set() + assert jwks_client.jwk_set_cache is not None + assert jwks_client.jwk_set_cache.get() is not None + with pytest.raises(PyJWKClientError): with mocked_failed_response(): jwks_client.get_jwk_set(refresh=True) - assert jwks_client.jwk_set_cache is None + cached = jwks_client.jwk_set_cache.get() + assert cached is not None + # Subsequent reads still serve from cache without another fetch. + with mocked_success_response(RESPONSE_DATA_WITH_MATCHING_KID) as call: + jwks_client.get_jwk_set() + assert call.call_count == 0 def test_failed_request_should_raise_connection_error(self) -> None: token = "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6Ik5FRTFRVVJCT1RNNE16STVSa0ZETlRZeE9UVTFNRGcyT0Rnd1EwVXpNVGsxUWpZeVJrUkZRdyJ9.eyJpc3MiOiJodHRwczovL2Rldi04N2V2eDlydS5hdXRoMC5jb20vIiwic3ViIjoiYVc0Q2NhNzl4UmVMV1V6MGFFMkg2a0QwTzNjWEJWdENAY2xpZW50cyIsImF1ZCI6Imh0dHBzOi8vZXhwZW5zZXMtYXBpIiwiaWF0IjoxNTcyMDA2OTU0LCJleHAiOjE1NzIwMDY5NjQsImF6cCI6ImFXNENjYTc5eFJlTFdVejBhRTJINmtEME8zY1hCVnRDIiwiZ3R5IjoiY2xpZW50LWNyZWRlbnRpYWxzIn0.PUxE7xn52aTCohGiWoSdMBZGiYAHwE5FYie0Y1qUT68IHSTXwXVd6hn02HTah6epvHHVKA2FqcFZ4GGv5VTHEvYpeggiiZMgbxFrmTEY0csL6VNkX1eaJGcuehwQCRBKRLL3zKmA5IKGy5GeUnIbpPHLHDxr-GXvgFzsdsyWlVQvPX2xjeaQ217r2PtxDeqjlf66UYl6oY6AqNS8DH3iryCvIfCcybRZkc_hdy-6ZMoKT6Piijvk_aXdm7-QQqKJFHLuEqrVSOuBqqiNfVrG27QzAPuPOxvfXTVLXL2jek5meH6n-VWgrBdoMFH93QEszEDowDAEhQPHVs0xj7SIzA" @@ -344,6 +356,36 @@ def test_get_jwt_set_invalid_lifespan(self) -> None: jwks_client = PyJWKClient(url, lifespan=-1) assert jwks_client is None + @pytest.mark.parametrize( + "uri", + [ + "file:///etc/passwd", + "ftp://example.org/keys.json", + 'data:application/json,{"keys":[]}', + "/etc/passwd", # urlparse gives scheme="" — also rejected + "ldap://internal.test/jwks", + ], + ) + def test_pyjwkclient_rejects_non_http_schemes(self, uri: str) -> None: + # urllib's default OpenerDirector handles file://, ftp://, and data: + # URIs. PyJWKClient must reject these so callers can't be tricked + # into reading attacker-controlled local files or other unintended + # schemes via a manipulated URI. + with pytest.raises(PyJWKClientError, match="Invalid JWKS URI scheme"): + PyJWKClient(uri) + + @pytest.mark.parametrize( + "uri", + [ + "http://localhost/jwks.json", + "https://example.test/jwks.json", + "HTTPS://Example.Test/jwks.json", # case-insensitive + ], + ) + def test_pyjwkclient_accepts_http_https_schemes(self, uri: str) -> None: + # Construction succeeds; no fetch is made until get_jwk_set(). + PyJWKClient(uri) + def test_get_jwt_set_timeout(self) -> None: url = "https://dev-87evx9ru.auth0.com/.well-known/jwks.json" jwks_client = PyJWKClient(url, timeout=5)