urllib3 redirects are not disabled when retries are disabled on PoolManager instantiation

@@ -1,3 +1,12 @@
+2.5.0 (TBD)
+==================
+
+- Fixed a security issue where restricting the maximum number of followed
+  redirects at the ``urllib3.PoolManager`` level via the ``retries`` parameter
+  did not work.
+- TODO: add other entries in the release PR.
+
+
 2.4.0 (2025-04-10)
 ==================
 

@@ -65,7 +65,7 @@ Features which are usable with Emscripten support are:
 * Timeouts
 * Retries
 * Streaming (with Web Workers and Cross-Origin Isolation)
-* Redirects
+* Redirects (determined by browser/runtime, not restrictable with urllib3)
 * Decompressing response bodies
 
 Features which don't work with Emscripten:

@@ -227,6 +227,7 @@ async def encodingrequest() -> ResponseReturnValue:
 
 
 @hypercorn_app.route("/redirect", methods=["GET", "POST", "PUT"])
+@pyodide_testing_app.route("/redirect", methods=["GET", "POST", "PUT"])
 async def redirect() -> ResponseReturnValue:
     "Perform a redirect to ``target``"
     values = await request.values

@@ -203,6 +203,22 @@ def __init__(
         **connection_pool_kw: typing.Any,
     ) -> None:
         super().__init__(headers)
+        if "retries" in connection_pool_kw:
+            retries = connection_pool_kw["retries"]
+            if not isinstance(retries, Retry):
+                # When Retry is initialized, raise_on_redirect is based
+                # on a redirect boolean value.
+                # But requests made via a pool manager always set
+                # redirect to False, and raise_on_redirect always ends
+                # up being False consequently.
+                # Here we fix the issue by setting raise_on_redirect to
+                # a value needed by the pool manager without considering
+                # the redirect boolean.
+                raise_on_redirect = retries is not False
+                retries = Retry.from_int(retries, redirect=False)
+                retries.raise_on_redirect = raise_on_redirect
+                connection_pool_kw = connection_pool_kw.copy()
+                connection_pool_kw["retries"] = retries
         self.connection_pool_kw = connection_pool_kw
 
         self.pools: RecentlyUsedContainer[PoolKey, HTTPConnectionPool]
@@ -456,7 +472,7 @@ def urlopen(  # type: ignore[override]
             kw["body"] = None
             kw["headers"] = HTTPHeaderDict(kw["headers"])._prepare_for_method_change()
 
-        retries = kw.get("retries")
+        retries = kw.get("retries", response.retries)
         if not isinstance(retries, Retry):
             retries = Retry.from_int(retries, redirect=redirect)
 

@@ -944,6 +944,22 @@ def count_calls(self, *args, **argv):  # type: ignore[no-untyped-def]
     pyodide_test(selenium_coverage, testserver_http.http_host, find_unused_port())
 
 
+def test_redirects(
+    selenium_coverage: typing.Any, testserver_http: PyodideServerInfo
+) -> None:
+    @run_in_pyodide  # type: ignore[misc]
+    def pyodide_test(selenium_coverage: typing.Any, host: str, port: int) -> None:
+        from urllib3 import request
+
+        redirect_url = f"http://{host}:{port}/redirect"
+        response = request("GET", redirect_url)
+        assert response.status == 200
+
+    pyodide_test(
+        selenium_coverage, testserver_http.http_host, testserver_http.http_port
+    )
+
+
 def test_insecure_requests_warning(
     selenium_coverage: typing.Any, testserver_http: PyodideServerInfo
 ) -> None:

@@ -379,9 +379,10 @@ def test_pool_kwargs_socket_options(self) -> None:
 
     def test_merge_pool_kwargs(self) -> None:
         """Assert _merge_pool_kwargs works in the happy case"""
-        p = PoolManager(retries=100)
+        retries = retry.Retry(total=100)
+        p = PoolManager(retries=retries)
         merged = p._merge_pool_kwargs({"new_key": "value"})
-        assert {"retries": 100, "new_key": "value"} == merged
+        assert {"retries": retries, "new_key": "value"} == merged
 
     def test_merge_pool_kwargs_none(self) -> None:
         """Assert false-y values to _merge_pool_kwargs result in defaults"""

@@ -84,6 +84,89 @@ def test_redirect_to_relative_url(self) -> None:
             assert r.status == 200
             assert r.data == b"Dummy server!"
 
+    @pytest.mark.parametrize(
+        "retries",
+        (0, Retry(total=0), Retry(redirect=0), Retry(total=0, redirect=0)),
+    )
+    def test_redirects_disabled_for_pool_manager_with_0(
+        self, retries: typing.Literal[0] | Retry
+    ) -> None:
+        """
+        Check handling redirects when retries is set to 0 on the pool
+        manager.
+        """
+        with PoolManager(retries=retries) as http:
+            with pytest.raises(MaxRetryError):
+                http.request("GET", f"{self.base_url}/redirect")
+
+            # Setting redirect=True should not change the behavior.
+            with pytest.raises(MaxRetryError):
+                http.request("GET", f"{self.base_url}/redirect", redirect=True)
+
+            # Setting redirect=False should not make it follow the redirect,
+            # but MaxRetryError should not be raised.
+            response = http.request("GET", f"{self.base_url}/redirect", redirect=False)
+            assert response.status == 303
+
+    @pytest.mark.parametrize(
+        "retries",
+        (
+            False,
+            Retry(total=False),
+            Retry(redirect=False),
+            Retry(total=False, redirect=False),
+        ),
+    )
+    def test_redirects_disabled_for_pool_manager_with_false(
+        self, retries: typing.Literal[False] | Retry
+    ) -> None:
+        """
+        Check that setting retries set to False on the pool manager disables
+        raising MaxRetryError and redirect=True does not change the
+        behavior.
+        """
+        with PoolManager(retries=retries) as http:
+            response = http.request("GET", f"{self.base_url}/redirect")
+            assert response.status == 303
+
+            response = http.request("GET", f"{self.base_url}/redirect", redirect=True)
+            assert response.status == 303
+
+            response = http.request("GET", f"{self.base_url}/redirect", redirect=False)
+            assert response.status == 303
+
+    def test_redirects_disabled_for_individual_request(self) -> None:
+        """
+        Check handling redirects when they are meant to be disabled
+        on the request level.
+        """
+        with PoolManager() as http:
+            # Check when redirect is not passed.
+            with pytest.raises(MaxRetryError):
+                http.request("GET", f"{self.base_url}/redirect", retries=0)
+            response = http.request("GET", f"{self.base_url}/redirect", retries=False)
+            assert response.status == 303
+
+            # Check when redirect=True.
+            with pytest.raises(MaxRetryError):
+                http.request(
+                    "GET", f"{self.base_url}/redirect", retries=0, redirect=True
+                )
+            response = http.request(
+                "GET", f"{self.base_url}/redirect", retries=False, redirect=True
+            )
+            assert response.status == 303
+
+            # Check when redirect=False.
+            response = http.request(
+                "GET", f"{self.base_url}/redirect", retries=0, redirect=False
+            )
+            assert response.status == 303
+            response = http.request(
+                "GET", f"{self.base_url}/redirect", retries=False, redirect=False
+            )
+            assert response.status == 303
+
     def test_cross_host_redirect(self) -> None:
         with PoolManager() as http:
             cross_host_location = f"{self.base_url_alt}/echo?a=b"
@@ -138,6 +221,24 @@ def test_too_many_redirects(self) -> None:
             pool = http.connection_from_host(self.host, self.port)
             assert pool.num_connections == 1
 
+        # Check when retries are configured for the pool manager.
+        with PoolManager(retries=1) as http:
+            with pytest.raises(MaxRetryError):
+                http.request(
+                    "GET",
+                    f"{self.base_url}/redirect",
+                    fields={"target": f"/redirect?target={self.base_url}/"},
+                )
+
+            # Here we allow more retries for the request.
+            response = http.request(
+                "GET",
+                f"{self.base_url}/redirect",
+                fields={"target": f"/redirect?target={self.base_url}/"},
+                retries=2,
+            )
+            assert response.status == 200
+
     def test_redirect_cross_host_remove_headers(self) -> None:
         with PoolManager() as http:
             r = http.request(