fix(image): validate every redirect destination

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

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))

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