CVE-2026-48522

@@ -4,9 +4,53 @@ Changelog
 All notable changes to this project will be documented in this file.
 This project adheres to `Semantic Versioning <https://semver.org/>`__.
 
-`Unreleased <https://github.com/jpadilla/pyjwt/compare/2.12.1...HEAD>`__
+`Unreleased <https://github.com/jpadilla/pyjwt/compare/2.13.0...HEAD>`__
 ------------------------------------------------------------------------
 
+`v2.13.0 <https://github.com/jpadilla/pyjwt/compare/2.12.1...2.13.0>`__
+-----------------------------------------------------------------------
+
+Security
+~~~~~~~~
+
+- Reject JWK JSON documents passed as raw HMAC secrets in
+  ``HMACAlgorithm.prepare_key`` to close an algorithm-confusion gap that
+  the existing PEM/SSH guard did not cover. Reported by @aradona91 in
+  `GHSA-xgmm-8j9v-c9wx <https://github.com/jpadilla/pyjwt/security/advisories/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 <https://github.com/jpadilla/pyjwt/security/advisories/GHSA-jq35-7prp-9v3f>`__.
+- Reject non-``http(s)`` URI schemes in ``PyJWKClient`` 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 <https://github.com/jpadilla/pyjwt/security/advisories/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, turning one bad JWKS request into application-
+  wide auth failure. Reported by @eddieran in `GHSA-fhv5-28vv-h8m8 <https://github.com/jpadilla/pyjwt/security/advisories/GHSA-fhv5-28vv-h8m8>`__.
+- Skip the unconditional base64 decode of the compact-form payload segment
+  when ``b64=false`` is set in the protected header, 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 <https://github.com/jpadilla/pyjwt/security/advisories/GHSA-w7vc-732c-9m39>`__.
+
+Fixed
+~~~~~
+
+- Reject empty HMAC keys outright in ``HMACAlgorithm.prepare_key`` with
+  ``InvalidKeyError`` instead of accepting them with only a warning.
+  Thanks to @SnailSploit and @spartan8806 for independently flagging the
+  footgun.
+- Forward per-call ``options`` (including ``enforce_minimum_key_length``)
+  from ``PyJWT.decode`` through to ``PyJWS._verify_signature`` so the
+  option actually takes effect when set at the call site rather than only
+  on the ``PyJWT`` instance. Thanks to @WLUB for the report.
+- RFC 7797 §3 compliance for ``b64=false``: the encoder now auto-adds
+  ``"b64"`` to the ``crit`` header parameter, and the decoder rejects
+  tokens that set ``b64=false`` without listing it in ``crit``. Thanks to
+  @MachineLearning-Nerd for the report.
+
 Changed
 ~~~~~~~
 

@@ -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

@@ -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)

@@ -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,

@@ -28,7 +28,7 @@
 from .jwks_client import PyJWKClient
 from .warnings import InsecureKeyLengthWarning
 
-__version__ = "2.12.1"
+__version__ = "2.13.0"
 
 __title__ = "PyJWT"
 __description__ = "JSON Web Token implementation in Python"

@@ -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.

@@ -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
 

@@ -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"])

@@ -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)