From ce005850b742304cc65da30ac93537d6c14a1cbd Mon Sep 17 00:00:00 2001 From: pedrooot Date: Thu, 8 Oct 2026 01:14:22 +0200 Subject: [PATCH] fix(image): validate every redirect destination --- .../image-registry-url-ssrf.security.md | 2 +- prowler/providers/image/lib/registry/base.py | 33 +++++++++- .../image/lib/registry/test_oci_adapter.py | 62 +++++++++++++++++++ 3 files changed, 94 insertions(+), 3 deletions(-) diff --git a/prowler/changelog.d/image-registry-url-ssrf.security.md b/prowler/changelog.d/image-registry-url-ssrf.security.md index 2cee8035db..bab387df3e 100644 --- a/prowler/changelog.d/image-registry-url-ssrf.security.md +++ b/prowler/changelog.d/image-registry-url-ssrf.security.md @@ -1 +1 @@ -Image registry URLs resolving to loopback, private, shared or otherwise non-public addresses rejected before any request leaves the adapter +Image registry URLs resolving to loopback, private, shared or otherwise non-public addresses rejected before any request leaves the adapter, including every destination a redirect points to diff --git a/prowler/providers/image/lib/registry/base.py b/prowler/providers/image/lib/registry/base.py index f33504edc3..b68930ce9c 100644 --- a/prowler/providers/image/lib/registry/base.py +++ b/prowler/providers/image/lib/registry/base.py @@ -8,7 +8,7 @@ import re import socket import time from abc import ABC, abstractmethod -from urllib.parse import urlparse +from urllib.parse import urljoin, urlparse import requests import tldextract @@ -22,6 +22,10 @@ from prowler.providers.image.exceptions.exceptions import ( ) _MAX_RETRIES = 3 +_MAX_REDIRECTS = 5 +# a registry legitimately redirects: Docker Hub sends blob pulls to a CDN and a +# renamed repository answers 301, so each hop is validated rather than refused +_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308}) _BACKOFF_BASE = 1 _USER_AGENT = f"Prowler/{prowler_version} (registry-adapter)" @@ -298,6 +302,29 @@ class RegistryAdapter(ABC): return canonical_url + def _request_following_validated_redirects( + self, method: str, url: str, **kwargs + ) -> requests.Response: + """Issue a request, validating the destination of each redirect it takes. + + ``requests`` follows redirects itself, which would send the request to a + host the guard never saw. + """ + for _ in range(_MAX_REDIRECTS + 1): + resp = requests.request(method, url, allow_redirects=False, **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( + urljoin(url, location), enforce_origin=False + ) + raise ImageRegistryNetworkError( + file=__file__, + message=f"More than {_MAX_REDIRECTS} redirects from {url}.", + ) + def _request_with_retry(self, method: str, url: str, **kwargs) -> requests.Response: context_label = kwargs.pop("context_label", None) or self.registry_url # the only chokepoint every outbound URL passes through, including the @@ -313,7 +340,9 @@ class RegistryAdapter(ABC): last_body = None for attempt in range(1, _MAX_RETRIES + 1): try: - resp = requests.request(method, url, **kwargs) + resp = self._request_following_validated_redirects( + method, url, **kwargs + ) if resp.status_code == 429: last_status = 429 wait = _BACKOFF_BASE * (2 ** (attempt - 1)) diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py index b0fcdd5285..4023cf0d02 100644 --- a/tests/providers/image/lib/registry/test_oci_adapter.py +++ b/tests/providers/image/lib/registry/test_oci_adapter.py @@ -11,6 +11,7 @@ from prowler.providers.image.exceptions.exceptions import ( ImageRegistryCatalogError, ImageRegistryNetworkError, ) +from prowler.providers.image.lib.registry.base import _MAX_REDIRECTS from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter @@ -1423,3 +1424,64 @@ class TestOciAdapterArtifactIndexAndCaseInsensitivity: {"artifactType": "Application/vnd.CNCF.Helm.Config.v1+json"}, ) assert self._adapter().is_container_image("charts/app", "1.0") is False + + +def _redirect(location: str): + return MagicMock(status_code=302, headers={"Location": location}) + + +class TestValidatedRedirects: + """A registry may redirect, but the guard must see each destination. + + ``requests`` follows redirects itself, so without this the final host is + never validated. + """ + + def _adapter(self): + return OciRegistryAdapter("https://reg.io", token="t") + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_never_lets_requests_follow_on_its_own(self, mock_request): + mock_request.return_value = MagicMock(status_code=200, headers={}) + + self._adapter()._request_with_retry("GET", "https://reg.io/v2/") + + assert mock_request.call_args.kwargs["allow_redirects"] is False + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_rejects_a_redirect_to_a_non_public_address(self, mock_request): + mock_request.return_value = _redirect("http://169.254.169.254/latest/meta-data") + + with pytest.raises(ImageRegistryAuthError, match="non-public"): + self._adapter()._request_with_retry("GET", "https://reg.io/v2/") + + assert mock_request.call_count == 1 + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_follows_a_redirect_to_a_public_address(self, mock_request): + # Docker Hub sends blob pulls to a CDN on an unrelated host + final = MagicMock(status_code=200, headers={}) + mock_request.side_effect = [_redirect("https://cdn.example.com/blob"), final] + + resp = self._adapter()._request_with_retry("GET", "https://reg.io/v2/blob") + + assert resp is final + assert mock_request.call_args_list[1][0][1] == "https://cdn.example.com/blob" + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_resolves_a_relative_location_against_the_current_url(self, mock_request): + final = MagicMock(status_code=200, headers={}) + mock_request.side_effect = [_redirect("/v2/token"), final] + + self._adapter()._request_with_retry("GET", "https://reg.io/v2/") + + assert mock_request.call_args_list[1][0][1] == "https://reg.io/v2/token" + + @patch("prowler.providers.image.lib.registry.base.requests.request") + def test_bounds_a_redirect_chain(self, mock_request): + mock_request.return_value = _redirect("https://reg.io/v2/again") + + with pytest.raises(ImageRegistryNetworkError, match="redirects"): + self._adapter()._request_with_retry("GET", "https://reg.io/v2/") + + assert mock_request.call_count == _MAX_REDIRECTS + 1