From 743a06a5816f1a38f65bf9e99f27d9fcd0c1fc16 Mon Sep 17 00:00:00 2001 From: pedrooot Date: Thu, 8 Oct 2026 09:07:58 +0200 Subject: [PATCH] fix(image): rebuild method and body on a redirect hop --- prowler/providers/image/lib/registry/base.py | 51 ++++++++++++++++--- .../image/lib/registry/test_oci_adapter.py | 40 +++++++++++++++ 2 files changed, 83 insertions(+), 8 deletions(-) diff --git a/prowler/providers/image/lib/registry/base.py b/prowler/providers/image/lib/registry/base.py index 9af9de2e7b..a54934c150 100644 --- a/prowler/providers/image/lib/registry/base.py +++ b/prowler/providers/image/lib/registry/base.py @@ -98,28 +98,61 @@ 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. +# 307 and 308 are the two that must replay the request unchanged +_BODY_PRESERVING_REDIRECTS = frozenset({307, 308}) +_BODY_OPTIONS = ("data", "json", "files") +_BODY_HEADERS = ("content-length", "content-type", "transfer-encoding") - Following redirects by hand loses what ``Session.resolve_redirects`` does for - free, so both of its rules are reapplied here. + +def _rebuilt_method(method: str, status_code: int) -> str: + """How ``requests`` rewrites the method on a redirect (RFC 7231 and history).""" + if status_code in (302, 303) and method != "HEAD": + return "GET" + if status_code == 301 and method == "POST": + return "GET" + return method + + +def _next_hop( + method: str, kwargs: dict, old_url: str, new_url: str, status_code: int +) -> tuple[str, dict]: + """Method and options for a redirect hop, rebuilt the way ``requests`` would. + + Following redirects by hand loses everything ``Session.resolve_redirects`` + does, so its rules are reapplied here rather than only the ones that first + came to mind. """ hop = dict(kwargs) # the Location carries its own query; reapplying params can invalidate a # signed URL a registry redirects to hop.pop("params", None) + + if status_code not in _BODY_PRESERVING_REDIRECTS: + # a redirected login would otherwise re-send its credentials in the body + for option in _BODY_OPTIONS: + hop.pop(option, None) + hop["headers"] = { + name: value + for name, value in (hop.get("headers") or {}).items() + if name.lower() not in _BODY_HEADERS + } + method = _rebuilt_method(method, status_code) + # 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 + # hop that keeps credentials hands the registry token to an unrelated host. + # `should_strip_auth` ignores `self`, but calling it off a live session reads + # better than passing None. with requests.Session() as redirect_rules: if not redirect_rules.should_strip_auth(old_url, new_url): - return hop + return method, hop + hop.pop("auth", None) hop["headers"] = { name: value for name, value in (hop.get("headers") or {}).items() if name.lower() != "authorization" } - return hop + return method, hop class RegistryAdapter(ABC): @@ -345,7 +378,9 @@ class RegistryAdapter(ABC): target = self._validate_outbound_url( urljoin(url, location), enforce_origin=False ) - hop_kwargs = _next_hop_kwargs(hop_kwargs, url, target) + method, hop_kwargs = _next_hop( + method, hop_kwargs, url, target, resp.status_code + ) url = target raise ImageRegistryNetworkError( file=__file__, diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py index 8bf085732d..90b94dcde5 100644 --- a/tests/providers/image/lib/registry/test_oci_adapter.py +++ b/tests/providers/image/lib/registry/test_oci_adapter.py @@ -1532,3 +1532,43 @@ class TestValidatedRedirects: ) assert "params" not in mock_request.call_args_list[1].kwargs + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_a_redirected_post_becomes_a_get_without_its_body(self, mock_request): + """A redirected login would otherwise re-send its credentials.""" + mock_request.side_effect = [ + _redirect("https://cdn.example.com/elsewhere"), + MagicMock(status_code=200, headers={}), + ] + + self._adapter()._request_with_retry( + "POST", "https://reg.io/v2/users/login", json={"password": "a-password"} + ) + + hop = mock_request.call_args_list[1] + assert hop[0][0] == "GET" + assert "json" not in hop.kwargs + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_a_temporary_redirect_replays_the_request(self, mock_request): + mock_request.side_effect = [ + MagicMock(status_code=307, headers={"Location": "https://reg.io/v2/moved"}), + MagicMock(status_code=200, headers={}), + ] + + self._adapter()._request_with_retry("POST", "https://reg.io/v2/", json={"a": 1}) + + hop = mock_request.call_args_list[1] + assert hop[0][0] == "POST" + assert hop.kwargs["json"] == {"a": 1} + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_a_301_only_rewrites_a_post(self, mock_request): + mock_request.side_effect = [ + MagicMock(status_code=301, headers={"Location": "https://reg.io/v2/moved"}), + MagicMock(status_code=200, headers={}), + ] + + self._adapter()._request_with_retry("PUT", "https://reg.io/v2/") + + assert mock_request.call_args_list[1][0][0] == "PUT"