Detect infinite redirection loops in URLFetcher

liZe · liZe · commit 4c414fe2144d · 2026-02-13T10:11:24.000+01:00
Fix #2686.
diff --git a/tests/test_api.py b/tests/test_api.py
@@ -139,6 +139,8 @@ def http_server():
             (b'<html test=accept-encoding-header-fail>', {})
         ),
         '/redirect': lambda env: (b'', {'Location': '/gzip'}),
+        '/redirect-loop': lambda env: (b'', {'Location': '/redirect-loop-2'}),
+        '/redirect-loop-2': lambda env: (b'', {'Location': '/redirect-loop'}),
     }
 
     def wsgi_app(environ, start_response):
@@ -729,6 +731,13 @@ def test_disallowed_protocols(command):
     assert 'URI uses disallowed protocol' in logs[0]
 
 
+@assert_no_logs
+def test_redirect_loop():
+    with http_server() as root_url:
+        with pytest.raises(URLFetchingError, match='infinite loop'):
+            _run(f'{root_url}/redirect-loop -')
+
+
 @pytest.mark.parametrize(('html', 'fields'), [
     ('<input>', ['/Tx', '/V ()']),
     ('<input value="">', ['/Tx', '/V ()']),
diff --git a/weasyprint/urls.py b/weasyprint/urls.py
@@ -312,6 +312,7 @@ def __init__(self, timeout=10, ssl_context=None, http_headers=None,
         self._http_headers = {**HTTP_HEADERS, **(http_headers or {})}
         self._allowed_protocols = allowed_protocols
         self._fail_on_errors = fail_on_errors
+        self._request = None
 
     def fetch(self, url, headers=None):
         """Fetch a given URL.
@@ -342,7 +343,8 @@ def fetch(self, url, headers=None):
 
         # Open URL.
         headers = {**self._http_headers, **(headers or {})}
-        http_request = request.Request(url, headers=headers)
+        http_request = self._request or request.Request(url, headers=headers)
+        self._request = None
         response = super().open(http_request, timeout=self._timeout)
 
         # Decompress response.
@@ -364,6 +366,7 @@ def fetch(self, url, headers=None):
 
     def open(self, url, data=None, timeout=None):
         if isinstance(url, request.Request):
+            self._request = url
             return self.fetch(url.full_url, url.headers)
         return self.fetch(url)
 
