fix: detect/stop redirect loops

httpxthrottlecache/filecache/transport.py

@@ -28,6 +28,13 @@
 MAX_READABLE_NAME = 120
 
 
+def _points_to_itself(request: httpx.Request, location: str) -> bool:
+    """Redirect target shares the cache key."""
+    # keys omit scheme: http->https collides
+    target = request.url.join(location)
+    return (target.host, target.path, target.query) == (request.url.host, request.url.path, request.url.query)
+
+
 class AlreadyLockedError(Exception):
     pass
 
@@ -333,7 +340,7 @@ def _cache_miss_response(
         tee_factory: Callable[..., Union[httpx.SyncByteStream, httpx.AsyncByteStream]],
     ):
         location = net.headers.get("location")
-        if net.status_code in CACHED_REDIRECTS and location:
+        if net.status_code in CACHED_REDIRECTS and location and not _points_to_itself(req, location):
             self._cache.store_redirect(path, net.status_code, location, net.headers.get("Date"))
             return net
 
@@ -365,6 +372,10 @@ def return_if_fresh(self, request: httpx.Request) -> Tuple[Optional[httpx.Respon
 
         if path:
             if fresh:
+                meta = json.loads(path.with_suffix(path.suffix + ".meta").read_text())
+                if meta.get("status") in CACHED_REDIRECTS and _points_to_itself(request, meta.get("location", "")):
+                    logger.info("Ignoring self-redirect cached for %s", request.url)
+                    return None, None
                 return self._cache_hit_response(request, path), path
             else:
                 lm = json.loads(path.with_suffix(path.suffix + ".meta").read_text()).get("origin_lm")

tests/test_redirect_cache.py

@@ -88,6 +88,41 @@ def handler(req):
     assert len(calls) == 1
 
 
+def test_scheme_upgrade_redirect_does_not_loop(tmp_path):
+    """http->https shares a cache key, so caching it would redirect https to itself."""
+    calls = []
+
+    def handler(req):
+        calls.append(str(req.url))
+        date = email.utils.formatdate(time.time(), usegmt=True)
+        if req.url.scheme == "http":
+            return httpx.Response(301, headers={"Location": str(req.url.copy_with(scheme="https")), "Date": date}, request=req)
+        return httpx.Response(200, headers={"Date": date}, stream=httpx.ByteStream(b"body"), request=req)
+
+    transport = CachingTransport(tmp_path, RULES, transport=httpx.MockTransport(handler))
+    with httpx.Client(transport=transport, max_redirects=5) as c:
+        assert c.get("http://a.test/feed", follow_redirects=True).content == b"body"
+        assert c.get("https://a.test/feed", follow_redirects=True).content == b"body"
+        assert c.get("http://a.test/feed", follow_redirects=True).content == b"body"
+
+
+def test_cached_self_redirect_is_ignored(tmp_path):
+    """A self-pointing redirect already on disk is treated as a miss."""
+    calls = []
+
+    def handler(req):
+        calls.append(str(req.url))
+        date = email.utils.formatdate(time.time(), usegmt=True)
+        return httpx.Response(200, headers={"Date": date}, stream=httpx.ByteStream(b"body"), request=req)
+
+    transport = CachingTransport(tmp_path, RULES, transport=httpx.MockTransport(handler))
+    p = transport._cache.to_path("a.test", "/feed", "")
+    transport._cache.store_redirect(p, 301, "https://a.test/feed", None)
+    with httpx.Client(transport=transport, max_redirects=5) as c:
+        assert c.get("https://a.test/feed", follow_redirects=True).content == b"body"
+    assert calls == ["https://a.test/feed"]
+
+
 @pytest.mark.asyncio
 async def test_301_is_served_from_cache_async(tmp_path):
     calls = []