fix(image): validate the registry URL before any request

prowler/changelog.d/image-registry-url-ssrf.security.md

@@ -0,0 +1 @@
+Image registry URLs resolving to loopback, private, shared or otherwise non-public addresses rejected before any request leaves the adapter

prowler/providers/image/lib/registry/base.py

@@ -40,6 +40,10 @@ def _ip_is_non_public(ip_str: str) -> bool:
         addr = ipaddress.ip_address(ip_str)
     except ValueError:
         return False
+    # is_global is the broad check; the properties stay because some multicast
+    # ranges report is_global and would otherwise slip through
+    if not addr.is_global:
+        return True
     return any(getattr(addr, prop) for prop in _NON_PUBLIC_IP_PROPERTIES)
 
 
@@ -139,6 +143,7 @@ class RegistryAdapter(ABC):
         signatures, SBOMs...) override this; by default everything is assumed
         to be an image.
         """
+        del repository, tag  # the default inspects neither
         return True
 
     def _origin_url(self) -> str:
@@ -226,7 +231,7 @@ class RegistryAdapter(ABC):
             )
 
         try:
-            addr = ipaddress.ip_address(host)
+            ipaddress.ip_address(host)
         except ValueError:
             try:
                 infos = socket.getaddrinfo(host, None)
@@ -245,9 +250,7 @@ class RegistryAdapter(ABC):
                         ),
                     )
         else:
-            if any(
-                getattr(addr, prop) for prop in _NON_PUBLIC_IP_PROPERTIES
-            ) and not self._ip_is_allowed(host):
+            if _ip_is_non_public(host) and not self._ip_is_allowed(host):
                 raise ImageRegistryAuthError(
                     file=__file__,
                     message=(
@@ -277,6 +280,9 @@ class RegistryAdapter(ABC):
 
     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
+        # tenant-supplied registry URL that no caller validates
+        url = self._validate_outbound_url(url, enforce_origin=False)
         kwargs.setdefault("timeout", 30)
         kwargs.setdefault("verify", self.verify_ssl)
         headers = kwargs.get("headers", {})

tests/providers/image/lib/registry/test_oci_adapter.py

@@ -547,6 +547,44 @@ class TestOutboundUrlValidator:
         assert canonical == "https://ghcr.io/token"
 
 
+class TestTenantSuppliedRegistryUrlValidator:
+    @pytest.mark.parametrize(
+        "registry_url",
+        [
+            "http://169.254.169.254",
+            "http://10.0.0.5:5000",
+            "http://127.0.0.1:5000",
+            "http://100.100.100.200",
+            "http://100.64.0.1",
+        ],
+    )
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_rejects_a_non_public_registry_url_before_any_request(
+        self, mock_request, registry_url
+    ):
+        adapter = OciRegistryAdapter(registry_url=registry_url)
+
+        with pytest.raises(ImageRegistryAuthError):
+            adapter.list_repositories()
+
+        mock_request.assert_not_called()
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_allows_a_registry_url_inside_an_allowlisted_network(
+        self, mock_request, monkeypatch
+    ):
+        monkeypatch.setenv(
+            "PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS", "10.0.0.0/8"
+        )
+        resp = MagicMock(status_code=200, headers={}, ok=True)
+        resp.json.return_value = {"repositories": ["internal-app"]}
+        mock_request.return_value = resp
+        adapter = OciRegistryAdapter(registry_url="http://10.0.0.5:5000")
+
+        assert adapter.list_repositories() == ["internal-app"]
+        assert mock_request.called
+
+
 class TestObtainBearerTokenAppliesValidator:
     """Integration: malicious Www-Authenticate realm must be rejected before the second call."""