fix(image): rebuild request options on a redirect hop

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}.",

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