fix(image): rebuild method and body on a redirect hop

docs/user-guide/providers/image/getting-started-image.mdx

@@ -336,13 +336,19 @@ Skipping TLS verification disables certificate validation for registry connectio
 
 <VersionBadge version="5.41.0" />
 
-By default, Prowler rejects registry-provided URLs (token endpoints, pagination links) that resolve to non-public addresses, as an SSRF defense. On-premises registries live on private networks by definition, so to scan them declare the trusted ranges explicitly:
+On-premises registries live on private networks by definition.
+
+**Prowler CLI scans them with no extra configuration.** On the CLI the operator supplies the registry themselves, so there is nothing to defend against and the check below does not apply.
+
+**Prowler App rejects a registry URL that resolves to a non-public address**, as an SSRF defense. The check covers the registry URL supplied for the scan as well as the URLs the registry itself returns — token endpoints and pagination links. There the registry arrives from a tenant and the worker must not be used to reach the internal network it runs in. To scan an on-premises registry from Prowler App, declare the trusted ranges explicitly:
 
 ```bash
-export PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS="192.168.65.254/32,10.20.0.0/16"
+PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS="192.168.65.254/32,10.20.0.0/16"
 ```
 
-The value is a comma-separated list of IPs and CIDRs. A resolved address inside an allowlisted range is permitted; every other non-public address stays blocked, so link-local (`169.254.169.254`), loopback, and the rest of the internal network remain protected. The variable applies to registry enumeration and to the connection test. Malformed entries fail at startup, and a non-empty allowlist is logged as a relaxed security control. When unset, behavior is unchanged: only public addresses are followed.
+The value is a comma-separated list of IPs and CIDRs. A resolved address inside an allowlisted range is permitted; every other non-public address stays blocked, so link-local (`169.254.169.254`), loopback, and the rest of the internal network remain protected. The variable applies to registry enumeration and to the connection test. Malformed entries fail at startup, and a non-empty allowlist is logged as a relaxed security control. The URL scheme is always restricted to `http` and `https`, and the same-origin rule on registry-provided URLs always applies, neither of which the allowlist relaxes.
+
+The scan runs in the worker, not the API, so the variable has to reach the worker. With Docker Compose set it in the root `.env`; with Helm add it under `api.djangoConfig`, which is rendered into the API ConfigMap that the worker inherits through `envFrom`. The other providers use `PROWLER_ALLOWED_PRIVATE_NETWORKS`.
 
 #### Supported Registries
 

prowler/__main__.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
+import os
 import sys
 import tempfile
 from os import environ
@@ -236,6 +237,12 @@ def _send_ocsf_to_cloud(file_path: str) -> dict | None:
 
 
 def prowler():
+    # On the CLI the operator supplies the target themselves, so the outbound host
+    # check defends nothing and would refuse their own private network. setdefault,
+    # so an operator who set the variable explicitly keeps their choice. The name is
+    # the one the registry guard reads, in `image/lib/registry/base.py`.
+    os.environ.setdefault("PROWLER_SKIP_OUTBOUND_HOST_CHECK", "true")
+
     # Parse Arguments
     # Refactor(CLI)
     parser = ProwlerArgumentParser()

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, 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,9 +22,25 @@ 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)"
 
+_ALLOWLIST_HINT = (
+    "To scan a registry on a private network, list the trusted ranges in "
+    "PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS."
+)
+
+# Same variable the shared guard reads; `prowler/__main__.py` sets it so the CLI,
+# where the operator supplies the registry themselves, is not refused its own
+# private network. Unset means the check runs, so a missing setting fails safe.
+SKIP_OUTBOUND_CHECK_ENV = "PROWLER_SKIP_OUTBOUND_HOST_CHECK"
+
+_TRUTHY = {"1", "true", "yes", "on"}
+
 _NON_PUBLIC_IP_PROPERTIES = (
     "is_private",
     "is_loopback",
@@ -35,11 +51,19 @@ _NON_PUBLIC_IP_PROPERTIES = (
 )
 
 
+def _outbound_check_skipped() -> bool:
+    return os.environ.get(SKIP_OUTBOUND_CHECK_ENV, "").strip().lower() in _TRUTHY
+
+
 def _ip_is_non_public(ip_str: str) -> bool:
     try:
         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)
 
 
@@ -74,6 +98,63 @@ def _parse_allowed_private_networks(
     return tuple(networks)
 
 
+# 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")
+
+
+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.
+    # `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 method, hop
+
+    hop.pop("auth", None)
+    hop["headers"] = {
+        name: value
+        for name, value in (hop.get("headers") or {}).items()
+        if name.lower() != "authorization"
+    }
+    return method, hop
+
+
 class RegistryAdapter(ABC):
     """Abstract base class for registry adapters."""
 
@@ -139,6 +220,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:
@@ -225,36 +307,38 @@ class RegistryAdapter(ABC):
                 message=f"URL has no host: {canonical_url}",
             )
 
-        try:
-            addr = ipaddress.ip_address(host)
-        except ValueError:
+        # Only the address classification is skipped. The origin rule below still
+        # applies: a registry-supplied URL pointing at an unrelated host is a
+        # different problem from deliberately reaching a private network.
+        if not _outbound_check_skipped():
             try:
-                infos = socket.getaddrinfo(host, None)
-            except socket.gaierror:
-                infos = []
-            for *_, sockaddr in infos:
-                resolved_ip = sockaddr[0]
-                if _ip_is_non_public(resolved_ip) and not self._ip_is_allowed(
-                    resolved_ip
-                ):
+                ipaddress.ip_address(host)
+            except ValueError:
+                try:
+                    infos = socket.getaddrinfo(host, None)
+                except socket.gaierror:
+                    infos = []
+                for *_, sockaddr in infos:
+                    resolved_ip = sockaddr[0]
+                    if _ip_is_non_public(resolved_ip) and not self._ip_is_allowed(
+                        resolved_ip
+                    ):
+                        raise ImageRegistryAuthError(
+                            file=__file__,
+                            message=(
+                                f"Host {host!r} resolves to non-public address "
+                                f"{resolved_ip}. {_ALLOWLIST_HINT}"
+                            ),
+                        )
+            else:
+                if _ip_is_non_public(host) and not self._ip_is_allowed(host):
                     raise ImageRegistryAuthError(
                         file=__file__,
                         message=(
-                            f"Host {host!r} resolves to non-public address {resolved_ip}. "
-                            "This may indicate an SSRF attempt."
+                            f"URL targets a non-public address: {host}. "
+                            f"{_ALLOWLIST_HINT}"
                         ),
                     )
-        else:
-            if any(
-                getattr(addr, prop) for prop in _NON_PUBLIC_IP_PROPERTIES
-            ) and not self._ip_is_allowed(host):
-                raise ImageRegistryAuthError(
-                    file=__file__,
-                    message=(
-                        f"URL targets a non-public address: {host}. "
-                        "This may indicate an SSRF attempt."
-                    ),
-                )
 
         if enforce_origin:
             registry_host = urlparse(origin_url or self._origin_url()).hostname or ""
@@ -275,8 +359,39 @@ 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.
+        """
+        hop_kwargs = dict(kwargs)
+        for _ in range(_MAX_REDIRECTS + 1):
+            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
+            target = self._validate_outbound_url(
+                urljoin(url, location), enforce_origin=False
+            )
+            method, hop_kwargs = _next_hop(
+                method, hop_kwargs, url, target, resp.status_code
+            )
+            url = target
+        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
+        # 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", {})
@@ -287,7 +402,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
 
 
@@ -547,6 +548,104 @@ 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 TestOutboundCheckOptOut:
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_private_registry_is_reachable_when_the_check_is_skipped(
+        self, mock_request, monkeypatch
+    ):
+        monkeypatch.setenv("PROWLER_SKIP_OUTBOUND_HOST_CHECK", "true")
+        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"]
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_private_registry_is_refused_when_the_variable_is_unset(
+        self, mock_request, monkeypatch
+    ):
+        monkeypatch.delenv("PROWLER_SKIP_OUTBOUND_HOST_CHECK", raising=False)
+        adapter = OciRegistryAdapter(registry_url="http://10.0.0.5:5000")
+
+        with pytest.raises(ImageRegistryAuthError):
+            adapter.list_repositories()
+
+        mock_request.assert_not_called()
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_the_origin_rule_still_applies_when_skipped(
+        self, mock_request, monkeypatch
+    ):
+        """Skipping the address check must not let a registry redirect us elsewhere."""
+        monkeypatch.setenv("PROWLER_SKIP_OUTBOUND_HOST_CHECK", "true")
+        adapter = OciRegistryAdapter(registry_url="https://registry.example.com")
+
+        with pytest.raises(ImageRegistryAuthError, match="unrelated to registry host"):
+            adapter._validate_outbound_url("https://attacker.example.net/token")
+
+        mock_request.assert_not_called()
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_a_same_origin_url_is_allowed_when_skipped(self, mock_request, monkeypatch):
+        monkeypatch.setenv("PROWLER_SKIP_OUTBOUND_HOST_CHECK", "true")
+        adapter = OciRegistryAdapter(registry_url="https://registry.example.com")
+
+        assert adapter._validate_outbound_url(
+            "https://registry.example.com/v2/token"
+        ).startswith("https://registry.example.com/")
+
+    @patch("prowler.providers.image.lib.registry.base.requests.request")
+    def test_the_scheme_check_still_applies_when_skipped(
+        self, mock_request, monkeypatch
+    ):
+        monkeypatch.setenv("PROWLER_SKIP_OUTBOUND_HOST_CHECK", "true")
+        adapter = OciRegistryAdapter(registry_url="https://registry.example.com")
+
+        with pytest.raises(ImageRegistryAuthError, match="Disallowed URL scheme"):
+            adapter._validate_outbound_url("file:///etc/passwd", enforce_origin=False)
+
+        mock_request.assert_not_called()
+
+
 class TestObtainBearerTokenAppliesValidator:
     """Integration: malicious Www-Authenticate realm must be rejected before the second call."""
 
@@ -1325,3 +1424,151 @@ 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
+
+    @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
+
+    @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"