From 0a795b8e1f6ef08f634aa7086fc41cc6d5ce3e56 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Padilla?= Date: Wed, 9 Sep 2026 20:20:46 -0400 Subject: [PATCH] Reject redirects in PyJWKClient fetches --- jwt/jwks_client.py | 21 +++++++-- tests/test_jwks_client.py | 91 ++++++++++++++++++++++++++++++++++----- 2 files changed, 99 insertions(+), 13 deletions(-) Index: pyjwt-2.12.1/jwt/jwks_client.py =================================================================== --- pyjwt-2.12.1.orig/jwt/jwks_client.py +++ pyjwt-2.12.1/jwt/jwks_client.py @@ -17,6 +17,19 @@ from .exceptions import PyJWKClientConne from .jwk_set_cache import JWKSetCache +class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + def redirect_request( + self, + req: urllib.request.Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> None: + return None + + class PyJWKClient: def __init__( self, @@ -142,9 +155,11 @@ class PyJWKClient: """ try: r = urllib.request.Request(url=self.uri, headers=self.headers) - with urllib.request.urlopen( - r, timeout=self.timeout, context=self.ssl_context - ) as response: + handlers: list[Any] = [_NoRedirectHandler()] + if self.ssl_context is not None: + handlers.append(urllib.request.HTTPSHandler(context=self.ssl_context)) + opener = urllib.request.build_opener(*handlers) + with opener.open(r, timeout=self.timeout) as response: jwk_set = json.load(response) except (URLError, TimeoutError) as e: if isinstance(e, HTTPError): Index: pyjwt-2.12.1/tests/test_jwks_client.py =================================================================== --- pyjwt-2.12.1.orig/tests/test_jwks_client.py +++ pyjwt-2.12.1/tests/test_jwks_client.py @@ -2,8 +2,10 @@ import contextlib import io import json import ssl +import threading import time from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from unittest import mock from urllib.error import HTTPError, URLError from email.message import Message @@ -12,6 +14,7 @@ import pytest import jwt from jwt import PyJWKClient +from jwt.jwks_client import _NoRedirectHandler from jwt.api_jwk import PyJWK from jwt.exceptions import DecodeError, PyJWKClientConnectionError, PyJWKClientError @@ -52,7 +55,10 @@ RESPONSE_DATA_NO_MATCHING_KID = { @contextlib.contextmanager def mocked_success_response(data: object) -> Iterator[mock.Mock]: - with mock.patch("urllib.request.urlopen") as urlopen_mock: + with mock.patch("urllib.request.build_opener") as build_opener_mock: + opener = mock.Mock() + build_opener_mock.return_value = opener + urlopen_mock = opener.open response = mock.Mock() response.__enter__ = mock.Mock(return_value=response) response.__exit__ = mock.Mock() @@ -63,7 +69,10 @@ def mocked_success_response(data: object @contextlib.contextmanager def mocked_failed_response() -> Iterator[mock.Mock]: - with mock.patch("urllib.request.urlopen") as urlopen_mock: + with mock.patch("urllib.request.build_opener") as build_opener_mock: + opener = mock.Mock() + build_opener_mock.return_value = opener + urlopen_mock = opener.open urlopen_mock.side_effect = URLError("Fail to process the request.") yield urlopen_mock @@ -72,7 +81,10 @@ def mocked_failed_response() -> Iterator def mocked_first_call_wrong_kid_second_call_correct_kid( response_data_one: object, response_data_two: object ) -> Iterator[mock.Mock]: - with mock.patch("urllib.request.urlopen") as urlopen_mock: + with mock.patch("urllib.request.build_opener") as build_opener_mock: + opener = mock.Mock() + build_opener_mock.return_value = opener + urlopen_mock = opener.open response = mock.Mock() response.__enter__ = mock.Mock(return_value=response) response.__exit__ = mock.Mock() @@ -86,14 +98,20 @@ def mocked_first_call_wrong_kid_second_c @contextlib.contextmanager def mocked_timeout() -> Iterator[mock.Mock]: - with mock.patch("urllib.request.urlopen") as urlopen_mock: + with mock.patch("urllib.request.build_opener") as build_opener_mock: + opener = mock.Mock() + build_opener_mock.return_value = opener + urlopen_mock = opener.open urlopen_mock.side_effect = TimeoutError("timed out") yield urlopen_mock @contextlib.contextmanager def mocked_http_error_response() -> Iterator[tuple[mock.Mock, HTTPError]]: - with mock.patch("urllib.request.urlopen") as urlopen_mock: + with mock.patch("urllib.request.build_opener") as build_opener_mock: + opener = mock.Mock() + build_opener_mock.return_value = opener + urlopen_mock = opener.open http_error = HTTPError( url="https://example.com", code=401, @@ -107,6 +125,50 @@ def mocked_http_error_response() -> Iter @crypto_required class TestPyJWKClient: + def test_fetch_data_rejects_redirect_without_contacting_destination(self) -> None: + destination_requests = 0 + + class RedirectHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - stdlib handler API + nonlocal destination_requests + if self.path == "/start": + self.send_response(302) + self.send_header("Location", f"{base_url}/destination") + self.end_headers() + else: + destination_requests += 1 + self.send_response(200) + self.end_headers() + + def log_message(self, _format: str, *args: object) -> None: + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), RedirectHandler) + base_url = f"http://127.0.0.1:{server.server_port}" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + client = PyJWKClient( + f"{base_url}/start", headers={"Authorization": "secret"} + ) + with pytest.raises(PyJWKClientConnectionError): + client.fetch_data() + assert destination_requests == 0 + finally: + server.shutdown() + server.server_close() + thread.join() + + def test_redirect_handler_does_not_follow_redirects(self) -> None: + handler = _NoRedirectHandler() + + assert ( + handler.redirect_request( + mock.Mock(), mock.Mock(), 302, "Found", {}, "https://other.test" + ) + is None + ) + def test_fetch_data_forwards_headers_to_correct_url(self) -> None: url = "https://dev-87evx9ru.auth0.com/.well-known/jwks.json" @@ -561,10 +623,19 @@ class TestPyJWKClient: ssl_ctx = ssl.create_default_context() jwks_client = PyJWKClient(url, ssl_context=ssl_ctx) - with mocked_success_response(RESPONSE_DATA_WITH_MATCHING_KID) as mock_request: + with mock.patch("urllib.request.build_opener") as build_opener_mock: + opener = mock.Mock() + build_opener_mock.return_value = opener + response = mock.Mock() + response.__enter__ = mock.Mock(return_value=response) + response.__exit__ = mock.Mock() + response.read.return_value = json.dumps(RESPONSE_DATA_WITH_MATCHING_KID) + opener.open.return_value = response jwk_set = jwks_client.get_jwk_set() - request_call = mock_request.call_args - assert request_call[1].get("context") is ssl_ctx + handlers = build_opener_mock.call_args.args + assert any( + getattr(handler, "_context", None) is ssl_ctx for handler in handlers + ) assert jwk_set is not None @@ -574,8 +645,8 @@ class TestPyJWKClient: url, ssl_context=ssl.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT) ) - with mock.patch("urllib.request.urlopen") as urlopen_mock: - urlopen_mock.side_effect = URLError( + with mock.patch("urllib.request.build_opener") as build_opener_mock: + build_opener_mock.return_value.open.side_effect = URLError( ssl.SSLCertVerificationError("certificate verify failed") ) with pytest.raises(PyJWKClientError):