From 406d36d22358fc6c1bdeec99aedc36b7515443cc Mon Sep 17 00:00:00 2001
From: pedrooot <pedromarting3@gmail.com>
Date: Thu, 8 Oct 2026 09:04:19 +0200
Subject: [PATCH] fix(image): rebuild request options on a redirect hop

---
 prowler/providers/image/lib/registry/base.py  | 31 +++++++++++-
 .../image/lib/registry/test_oci_adapter.py    | 47 +++++++++++++++++++
 2 files changed, 76 insertions(+), 2 deletions(-)

diff --git a/prowler/providers/image/lib/registry/base.py b/prowler/providers/image/lib/registry/base.py
index b68930ce9c..9af9de2e7b 100644
--- a/prowler/providers/image/lib/registry/base.py
+++ b/prowler/providers/image/lib/registry/base.py
@@ -98,6 +98,30 @@ def _parse_allowed_private_networks(
     return tuple(networks)
 
 
+def _next_hop_kwargs(kwargs: dict, old_url: str, new_url: str) -> dict:
+    """Request options for a redirect hop, rebuilt the way ``requests`` would.
+
+    Following redirects by hand loses what ``Session.resolve_redirects`` does for
+    free, so both of its rules are reapplied here.
+    """
+    hop = dict(kwargs)
+    # the Location carries its own query; reapplying params can invalidate a
+    # signed URL a registry redirects to
+    hop.pop("params", None)
+    # borrowed rather than restated: the port and scheme cases are subtle, and a
+    # hop that keeps credentials hands the registry token to an unrelated host
+    with requests.Session() as redirect_rules:
+        if not redirect_rules.should_strip_auth(old_url, new_url):
+            return hop
+    hop.pop("auth", None)
+    hop["headers"] = {
+        name: value
+        for name, value in (hop.get("headers") or {}).items()
+        if name.lower() != "authorization"
+    }
+    return hop
+
+
 class RegistryAdapter(ABC):
     """Abstract base class for registry adapters."""
 
@@ -310,16 +334,19 @@ class RegistryAdapter(ABC):
         ``requests`` follows redirects itself, which would send the request to a
         host the guard never saw.
         """
+        hop_kwargs = dict(kwargs)
         for _ in range(_MAX_REDIRECTS + 1):
-            resp = requests.request(method, url, allow_redirects=False, **kwargs)
+            resp = requests.request(method, url, allow_redirects=False, **hop_kwargs)
             if resp.status_code not in _REDIRECT_STATUSES:
                 return resp
             location = resp.headers.get("Location")
             if not location:
                 return resp
-            url = self._validate_outbound_url(
+            target = self._validate_outbound_url(
                 urljoin(url, location), enforce_origin=False
             )
+            hop_kwargs = _next_hop_kwargs(hop_kwargs, url, target)
+            url = target
         raise ImageRegistryNetworkError(
             file=__file__,
             message=f"More than {_MAX_REDIRECTS} redirects from {url}.",
diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py
index 4023cf0d02..8bf085732d 100644
--- a/tests/providers/image/lib/registry/test_oci_adapter.py
+++ b/tests/providers/image/lib/registry/test_oci_adapter.py
@@ -1485,3 +1485,50 @@ class TestValidatedRedirects:
             self._adapter()._request_with_retry("GET", "https://reg.io/v2/")
 
         assert mock_request.call_count == _MAX_REDIRECTS + 1
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_drops_credentials_on_a_cross_origin_redirect(self, mock_request):
+        """Following redirects by hand loses what requests' rebuild_auth does."""
+        mock_request.side_effect = [
+            _redirect("https://cdn.example.com/signed?sig=abc"),
+            MagicMock(status_code=200, headers={}),
+        ]
+
+        self._adapter()._request_with_retry(
+            "GET",
+            "https://reg.io/v2/blob",
+            headers={"Authorization": "Bearer a-token"},
+            auth=("user", "pass"),
+        )
+
+        hop_kwargs = mock_request.call_args_list[1].kwargs
+        assert "Authorization" not in hop_kwargs["headers"]
+        assert "auth" not in hop_kwargs
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_keeps_credentials_on_a_same_origin_redirect(self, mock_request):
+        mock_request.side_effect = [
+            _redirect("https://reg.io/v2/token"),
+            MagicMock(status_code=200, headers={}),
+        ]
+
+        self._adapter()._request_with_retry(
+            "GET", "https://reg.io/v2/", headers={"Authorization": "Bearer a-token"}
+        )
+
+        hop_headers = mock_request.call_args_list[1].kwargs["headers"]
+        assert hop_headers["Authorization"] == "Bearer a-token"
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_does_not_reapply_params_to_the_redirect_destination(self, mock_request):
+        """A signed URL a registry redirects to carries its own query."""
+        mock_request.side_effect = [
+            _redirect("https://reg.io/signed?sig=abc"),
+            MagicMock(status_code=200, headers={}),
+        ]
+
+        self._adapter()._request_with_retry(
+            "GET", "https://reg.io/v2/_catalog", params={"n": 200}
+        )
+
+        assert "params" not in mock_request.call_args_list[1].kwargs