diff --git a/docs/user-guide/providers/image/getting-started-image.mdx b/docs/user-guide/providers/image/getting-started-image.mdx
index 6382f10e5c..bb1c294af5 100644
--- a/docs/user-guide/providers/image/getting-started-image.mdx
+++ b/docs/user-guide/providers/image/getting-started-image.mdx
@@ -306,9 +306,21 @@ prowler image --registry internal-registry.local --registry-insecure
```
-Skipping TLS verification disables certificate validation for registry connections. Use this flag only for trusted internal registries with self-signed certificates.
+Skipping TLS verification disables certificate validation for registry connections, including the Trivy image pull (`TRIVY_INSECURE`). Use this flag only for trusted internal registries with self-signed certificates.
+#### On-Premises Registries and Private Networks
+
+
+
+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:
+
+```bash
+export 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.
+
#### Supported Registries
Registry Scan Mode supports the following registry types:
diff --git a/prowler/changelog.d/image-private-network-allowlist.added.md b/prowler/changelog.d/image-private-network-allowlist.added.md
new file mode 100644
index 0000000000..fc9d1cc39d
--- /dev/null
+++ b/prowler/changelog.d/image-private-network-allowlist.added.md
@@ -0,0 +1 @@
+`PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS` environment variable so the image provider can reach container registries on allowlisted private networks, keeping every other non-public address blocked
diff --git a/prowler/changelog.d/image-registry-basic-auth-fallback.fixed.md b/prowler/changelog.d/image-registry-basic-auth-fallback.fixed.md
new file mode 100644
index 0000000000..d5ff396694
--- /dev/null
+++ b/prowler/changelog.d/image-registry-basic-auth-fallback.fixed.md
@@ -0,0 +1 @@
+Basic authentication fallback in the image provider when a registry rejects the negotiated bearer token, so registries like Harbor that guard catalog listing behind Basic can be enumerated
diff --git a/prowler/changelog.d/image-registry-bearer-auth-switch.fixed.md b/prowler/changelog.d/image-registry-bearer-auth-switch.fixed.md
new file mode 100644
index 0000000000..9b76bf292c
--- /dev/null
+++ b/prowler/changelog.d/image-registry-bearer-auth-switch.fixed.md
@@ -0,0 +1 @@
+Registry catalog listing when the server answers with a Bearer challenge after negotiating Basic (or anonymous) authentication, switching to a bearer token obtained from the challenge instead of failing
diff --git a/prowler/changelog.d/image-registry-insecure-trivy.fixed.md b/prowler/changelog.d/image-registry-insecure-trivy.fixed.md
new file mode 100644
index 0000000000..637852e147
--- /dev/null
+++ b/prowler/changelog.d/image-registry-insecure-trivy.fixed.md
@@ -0,0 +1 @@
+`--registry-insecure` now propagates to Trivy via `TRIVY_INSECURE`, so images in registries with self-signed certificates can be pulled and scanned, not just enumerated
diff --git a/prowler/providers/image/exceptions/exceptions.py b/prowler/providers/image/exceptions/exceptions.py
index 387b443ce3..72e362a72a 100644
--- a/prowler/providers/image/exceptions/exceptions.py
+++ b/prowler/providers/image/exceptions/exceptions.py
@@ -70,6 +70,10 @@ class ImageBaseException(ProwlerException):
"message": "Invalid regex filter pattern.",
"remediation": "Check the regex syntax for --image-filter or --tag-filter.",
},
+ (11019, "ImageInvalidAllowedNetworksError"): {
+ "message": "Malformed private-network allowlist.",
+ "remediation": "PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS must be a comma-separated list of IPs or CIDRs (e.g. 192.168.65.254/32,10.20.0.0/16).",
+ },
}
def __init__(self, code, file=None, original_exception=None, message=None):
@@ -227,3 +231,12 @@ class ImageInvalidFilterError(ImageBaseException):
super().__init__(
11017, file=file, original_exception=original_exception, message=message
)
+
+
+class ImageInvalidAllowedNetworksError(ImageBaseException):
+ """Exception raised when the private-network allowlist is malformed."""
+
+ def __init__(self, file=None, original_exception=None, message=None):
+ super().__init__(
+ 11019, file=file, original_exception=original_exception, message=message
+ )
diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py
index 7a245e4125..8eb4f2b198 100644
--- a/prowler/providers/image/image_provider.py
+++ b/prowler/providers/image/image_provider.py
@@ -718,6 +718,8 @@ class ImageProvider(Provider):
env["TRIVY_PASSWORD"] = self.registry_password
elif self.registry_token:
env["TRIVY_REGISTRY_TOKEN"] = self.registry_token
+ if self.registry_insecure:
+ env["TRIVY_INSECURE"] = "true"
return env
def _execute_trivy(self, command: list, image: str) -> subprocess.CompletedProcess:
diff --git a/prowler/providers/image/lib/registry/base.py b/prowler/providers/image/lib/registry/base.py
index 298583620f..b8c08955b6 100644
--- a/prowler/providers/image/lib/registry/base.py
+++ b/prowler/providers/image/lib/registry/base.py
@@ -3,6 +3,7 @@
from __future__ import annotations
import ipaddress
+import os
import re
import socket
import time
@@ -15,6 +16,7 @@ import tldextract
from prowler.config.config import prowler_version
from prowler.lib.logger import logger
from prowler.providers.image.exceptions.exceptions import (
+ ImageInvalidAllowedNetworksError,
ImageRegistryAuthError,
ImageRegistryNetworkError,
)
@@ -48,6 +50,30 @@ def _registrable_domain(host: str) -> str | None:
return f"{ext.domain}.{ext.suffix}"
+ALLOWED_PRIVATE_NETWORKS_ENV = "PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS"
+
+
+def _parse_allowed_private_networks(
+ raw: str | None,
+) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
+ """Parse the comma-separated IP/CIDR allowlist; malformed entries fail loudly."""
+ if raw is None or not raw.strip():
+ return ()
+ networks = []
+ for entry in raw.split(","):
+ entry = entry.strip()
+ if not entry:
+ continue
+ try:
+ networks.append(ipaddress.ip_network(entry, strict=False))
+ except ValueError as exc:
+ raise ImageInvalidAllowedNetworksError(
+ file=__file__,
+ message=f"Malformed entry {entry!r} in {ALLOWED_PRIVATE_NETWORKS_ENV}: {exc}",
+ )
+ return tuple(networks)
+
+
class RegistryAdapter(ABC):
"""Abstract base class for registry adapters."""
@@ -64,6 +90,14 @@ class RegistryAdapter(ABC):
self._password = password
self._token = token
self.verify_ssl = verify_ssl
+ self._allowed_private_networks = _parse_allowed_private_networks(
+ os.environ.get(ALLOWED_PRIVATE_NETWORKS_ENV)
+ )
+ if self._allowed_private_networks:
+ logger.warning(
+ f"{ALLOWED_PRIVATE_NETWORKS_ENV} is set — SSRF protection relaxed for private networks: "
+ + ", ".join(str(net) for net in self._allowed_private_networks)
+ )
@property
def password(self) -> str | None:
@@ -106,6 +140,30 @@ class RegistryAdapter(ABC):
"""
return self.registry_url
+ def _ip_is_allowed(self, ip_str: str) -> bool:
+ """Whether ip_str falls inside an operator-allowlisted private network."""
+ try:
+ addr = ipaddress.ip_address(ip_str)
+ except ValueError:
+ return False
+ return any(
+ addr.version == network.version and addr in network
+ for network in self._allowed_private_networks
+ )
+
+ def _host_in_allowed_networks(self, host: str) -> bool:
+ """Whether host is a literal allowlisted IP or resolves only to allowlisted IPs."""
+ try:
+ ipaddress.ip_address(host)
+ except ValueError:
+ try:
+ infos = socket.getaddrinfo(host, None)
+ except socket.gaierror:
+ return False
+ ips = {sockaddr[0] for *_, sockaddr in infos}
+ return bool(ips) and all(self._ip_is_allowed(ip) for ip in ips)
+ return self._ip_is_allowed(host)
+
def _validate_outbound_url(
self,
url: str,
@@ -119,10 +177,12 @@ class RegistryAdapter(ABC):
- canonicalise via ``requests.PreparedRequest`` so validator and connector
parse the same string the same way;
- reject schemes other than http/https;
- - reject literal non-public IPs (private, loopback, link-local, ...);
- - reject hostnames whose A/AAAA records resolve to non-public IPs;
+ - reject literal non-public IPs (private, loopback, link-local, ...)
+ unless inside PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS;
+ - reject hostnames whose A/AAAA records resolve to non-public IPs,
+ with the same allowlist exception;
- when ``enforce_origin=True``, reject hosts that don't share the
- registry's registrable domain.
+ registry's registrable domain or resolve into the allowlist.
Returns the canonical URL the caller should pass to ``requests``.
"""
@@ -165,7 +225,9 @@ class RegistryAdapter(ABC):
infos = []
for *_, sockaddr in infos:
resolved_ip = sockaddr[0]
- if _ip_is_non_public(resolved_ip):
+ if _ip_is_non_public(resolved_ip) and not self._ip_is_allowed(
+ resolved_ip
+ ):
raise ImageRegistryAuthError(
file=__file__,
message=(
@@ -174,7 +236,9 @@ class RegistryAdapter(ABC):
),
)
else:
- if any(getattr(addr, prop) for prop in _NON_PUBLIC_IP_PROPERTIES):
+ if any(
+ getattr(addr, prop) for prop in _NON_PUBLIC_IP_PROPERTIES
+ ) and not self._ip_is_allowed(host):
raise ImageRegistryAuthError(
file=__file__,
message=(
@@ -188,7 +252,10 @@ class RegistryAdapter(ABC):
if registry_host and host != registry_host:
target_d = _registrable_domain(host)
registry_d = _registrable_domain(registry_host)
- if not (target_d and registry_d and target_d == registry_d):
+ same_domain = bool(target_d and registry_d and target_d == registry_d)
+ # Non-public TLDs (.local, .internal, bare hostnames) have no
+ # registrable domain; fall back to the operator allowlist.
+ if not same_domain and not self._host_in_allowed_networks(host):
raise ImageRegistryAuthError(
file=__file__,
message=(
diff --git a/prowler/providers/image/lib/registry/oci_adapter.py b/prowler/providers/image/lib/registry/oci_adapter.py
index 878525bd04..c5c35aef06 100644
--- a/prowler/providers/image/lib/registry/oci_adapter.py
+++ b/prowler/providers/image/lib/registry/oci_adapter.py
@@ -45,6 +45,22 @@ class OciRegistryAdapter(RegistryAdapter):
def _origin_url(self) -> str:
return self._base_url
+ @staticmethod
+ def _find_challenge(www_authenticate: str, scheme: str) -> str | None:
+ """Extract one scheme's challenge from a (possibly multi-challenge) header.
+
+ RFC 7235 allows several comma-separated challenges in any order, e.g.
+ ``Basic realm="registry", Bearer realm="...",service="..."``.
+ """
+ match = re.search(
+ rf'(?:^|,)\s*{scheme}\b((?:\s*[\w-]+="[^"]*"\s*,?)*)',
+ www_authenticate,
+ re.IGNORECASE,
+ )
+ if not match:
+ return None
+ return f"{scheme} {match.group(1).strip().rstrip(',')}"
+
def list_repositories(self) -> list[str]:
self._ensure_auth()
repositories: list[str] = []
@@ -93,7 +109,8 @@ class OciRegistryAdapter(RegistryAdapter):
if resp.status_code == 401:
www_auth = resp.headers.get("Www-Authenticate", "")
- if not www_auth.lower().startswith("bearer"):
+ bearer_challenge = self._find_challenge(www_auth, "Bearer")
+ if not bearer_challenge:
# Basic auth challenge (e.g., AWS ECR)
if self.username and self.password:
self._basic_auth_verified = True
@@ -108,7 +125,7 @@ class OciRegistryAdapter(RegistryAdapter):
)
# Bearer token exchange (standard OCI flow)
- self._bearer_token = self._obtain_bearer_token(www_auth, repository)
+ self._bearer_token = self._obtain_bearer_token(bearer_challenge, repository)
return
if resp.status_code == 403:
raise ImageRegistryAuthError(
@@ -130,7 +147,18 @@ class OciRegistryAdapter(RegistryAdapter):
message=f"Cannot parse token endpoint from registry {self.registry_url}. Www-Authenticate: {www_authenticate[:200]}",
)
realm = self._validate_outbound_url(match.group(1))
- if urlparse(realm).scheme == "http":
+ realm_is_http = urlparse(realm).scheme == "http"
+ if realm_is_http and urlparse(self._base_url).scheme == "https":
+ # Transport downgrade: an on-path attacker could read or replace
+ # the token. An all-HTTP registry is an explicit operator choice.
+ raise ImageRegistryAuthError(
+ file=__file__,
+ message=(
+ f"Registry {self.registry_url} uses HTTPS but its token realm "
+ f"{realm} uses HTTP; refusing to exchange a token over cleartext."
+ ),
+ )
+ if realm_is_http:
logger.warning(f"Bearer token realm uses HTTP (not HTTPS): {realm}")
params: dict = {}
service_match = re.search(r'service="([^"]+)"', www_authenticate)
@@ -184,6 +212,37 @@ class OciRegistryAdapter(RegistryAdapter):
self._bearer_token = None
self._ensure_auth()
resp = self._do_authed_request(method, url, **kwargs)
+ if (
+ resp.status_code == 401
+ and self._bearer_token
+ and self.username
+ and self.password
+ and self._is_same_origin_as_registry(url)
+ and self._find_challenge(resp.headers.get("Www-Authenticate", ""), "Basic")
+ ):
+ # Registries like Harbor guard some endpoints (e.g. /_catalog) with
+ # Basic even when /v2/ negotiates Bearer.
+ logger.debug(
+ f"Bearer token not accepted for {url}, retrying with Basic auth"
+ )
+ user, pwd = self._resolve_basic_credentials()
+ resp = self._request_with_retry(method, url, auth=(user, pwd), **kwargs)
+ if resp.ok:
+ # Stay in Basic mode so later requests (e.g. catalog pages)
+ # skip the doomed Bearer round-trips.
+ self._basic_auth_verified = True
+ self._bearer_token = None
+ if resp.status_code == 401 and not self._bearer_token:
+ bearer_challenge = self._find_challenge(
+ resp.headers.get("Www-Authenticate", ""), "Bearer"
+ )
+ if bearer_challenge and self._is_same_origin_as_registry(url):
+ # The inverse switch: /v2/ negotiated Basic (or anonymous) but
+ # this endpoint demands Bearer. The challenge carries the right
+ # scope.
+ logger.debug(f"Basic auth not accepted for {url}, switching to Bearer")
+ self._bearer_token = self._obtain_bearer_token(bearer_challenge)
+ resp = self._do_authed_request(method, url, **kwargs)
return resp
def _do_authed_request(self, method: str, url: str, **kwargs) -> requests.Response:
diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py
index 92c4236dd8..df80e1d35b 100644
--- a/tests/providers/image/image_provider_test.py
+++ b/tests/providers/image/image_provider_test.py
@@ -678,6 +678,18 @@ class TestImageProviderRegistryAuth:
assert env["TRIVY_REGISTRY_TOKEN"] == "my-token"
+ def test_build_trivy_env_registry_insecure_sets_trivy_insecure(self):
+ provider = _make_provider(registry_insecure=True)
+ env = provider._build_trivy_env()
+
+ assert env["TRIVY_INSECURE"] == "true"
+
+ def test_build_trivy_env_secure_registry_leaves_trivy_insecure_unset(self):
+ provider = _make_provider()
+ env = provider._build_trivy_env()
+
+ assert "TRIVY_INSECURE" not in env
+
@patch("subprocess.run")
def test_execute_trivy_sets_trivy_env_with_basic_auth(self, mock_subprocess):
"""Test that _execute_trivy sets TRIVY_USERNAME/PASSWORD for native Trivy auth."""
@@ -1381,3 +1393,62 @@ class TestRegistryListMode:
# This is the line that crashes: global_provider is None so
# .print_credentials() raises AttributeError.
global_provider.print_credentials()
+
+
+class TestConnectionPrivateNetworkAllowlist:
+ """PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS applies to test_connection."""
+
+ ENV = "PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS"
+
+ @staticmethod
+ def _private_dns(host_to_ip):
+ def _stub(host, *_args, **_kwargs):
+ return [(2, 1, 6, "", (host_to_ip[host], 0))]
+
+ return _stub
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_private_registry_rejected_without_allowlist(
+ self, mock_request, monkeypatch
+ ):
+ monkeypatch.delenv(self.ENV, raising=False)
+ ping = MagicMock(
+ status_code=401,
+ headers={
+ "Www-Authenticate": 'Bearer realm="https://harbor.internal/service/token",service="harbor-registry"'
+ },
+ )
+ mock_request.return_value = ping
+
+ with patch(
+ "prowler.providers.image.lib.registry.base.socket.getaddrinfo",
+ side_effect=self._private_dns({"harbor.internal": "10.20.0.5"}),
+ ):
+ result = ImageProvider.test_connection(
+ image="harbor.internal", raise_on_exception=False
+ )
+
+ assert result.is_connected is False
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_private_registry_permitted_with_allowlist(self, mock_request, monkeypatch):
+ monkeypatch.setenv(self.ENV, "10.20.0.0/16")
+ ping = MagicMock(
+ status_code=401,
+ headers={
+ "Www-Authenticate": 'Bearer realm="https://harbor.internal/service/token",service="harbor-registry"'
+ },
+ )
+ token = MagicMock(status_code=200)
+ token.json.return_value = {"token": "tok"}
+ catalog = MagicMock(status_code=200, headers={})
+ catalog.json.return_value = {"repositories": ["app"]}
+ mock_request.side_effect = [ping, token, catalog]
+
+ with patch(
+ "prowler.providers.image.lib.registry.base.socket.getaddrinfo",
+ side_effect=self._private_dns({"harbor.internal": "10.20.0.5"}),
+ ):
+ result = ImageProvider.test_connection(image="harbor.internal")
+
+ assert result.is_connected is True
diff --git a/tests/providers/image/lib/registry/test_oci_adapter.py b/tests/providers/image/lib/registry/test_oci_adapter.py
index b814019504..f904c00c21 100644
--- a/tests/providers/image/lib/registry/test_oci_adapter.py
+++ b/tests/providers/image/lib/registry/test_oci_adapter.py
@@ -6,6 +6,7 @@ import pytest
import requests
from prowler.providers.image.exceptions.exceptions import (
+ ImageInvalidAllowedNetworksError,
ImageRegistryAuthError,
ImageRegistryCatalogError,
ImageRegistryNetworkError,
@@ -96,6 +97,37 @@ class TestOciAdapterAuth:
adapter._ensure_auth()
assert adapter._bearer_token == "bearer-tok"
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_http_realm_from_https_registry_raises(self, mock_request):
+ ping_resp = MagicMock(
+ status_code=401,
+ headers={
+ "Www-Authenticate": 'Bearer realm="http://auth.reg.io/token",service="registry"'
+ },
+ )
+ mock_request.return_value = ping_resp
+ adapter = OciRegistryAdapter("https://reg.io", username="u", password="p")
+ with pytest.raises(ImageRegistryAuthError, match="cleartext"):
+ adapter._ensure_auth()
+ # The token exchange must never happen: only the /v2/ ping went out
+ assert mock_request.call_count == 1
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_http_realm_from_http_registry_keeps_credentials(self, mock_request):
+ ping_resp = MagicMock(
+ status_code=401,
+ headers={
+ "Www-Authenticate": 'Bearer realm="http://reg.io/token",service="registry"'
+ },
+ )
+ token_resp = MagicMock(status_code=200)
+ token_resp.json.return_value = {"token": "bearer-tok"}
+ mock_request.side_effect = [ping_resp, token_resp]
+ adapter = OciRegistryAdapter("http://reg.io", username="u", password="p")
+ adapter._ensure_auth()
+ token_call = mock_request.call_args_list[1]
+ assert token_call.kwargs.get("auth") == ("u", "p")
+
@patch("prowler.providers.image.lib.registry.base.requests.request")
def test_ensure_auth_403_raises(self, mock_request):
resp = MagicMock(status_code=403)
@@ -218,7 +250,7 @@ class TestOciAdapterAuth:
adapter = OciRegistryAdapter("reg.io", username="u", password="p")
adapter._basic_auth_verified = True
# No bearer token — using basic auth
- resp_401 = MagicMock(status_code=401)
+ resp_401 = MagicMock(status_code=401, headers={})
mock_request.return_value = resp_401
result = adapter._authed_request("GET", "https://reg.io/v2/_catalog")
assert result.status_code == 401
@@ -709,3 +741,387 @@ class TestCredentialRedaction:
adapter = OciRegistryAdapter("reg.io", password="secret", token="tok")
assert adapter.password == "secret"
assert adapter.token == "tok"
+
+
+class TestAllowedPrivateNetworks:
+ """PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS: explicit CIDR allowlist
+ consulted by the SSRF guard; unset preserves the default rejection."""
+
+ ENV = "PROWLER_IMAGE_PROVIDER_ALLOWED_PRIVATE_NETWORKS"
+
+ def test_unset_env_keeps_private_origin_rejected(self, monkeypatch):
+ monkeypatch.delenv(self.ENV, raising=False)
+ adapter = OciRegistryAdapter("https://10.0.0.5:5000")
+ with pytest.raises(ImageRegistryAuthError, match="non-public"):
+ adapter._validate_outbound_url("https://10.0.0.5:5000/v2/_catalog?last=x")
+
+ def test_empty_env_keeps_private_origin_rejected(self, monkeypatch):
+ monkeypatch.setenv(self.ENV, " ")
+ adapter = OciRegistryAdapter("https://10.0.0.5:5000")
+ with pytest.raises(ImageRegistryAuthError, match="non-public"):
+ adapter._validate_outbound_url("https://10.0.0.5:5000/v2/_catalog?last=x")
+
+ def test_allowlisted_literal_ip_permitted(self, monkeypatch):
+ monkeypatch.setenv(self.ENV, "192.168.65.254/32,10.20.0.0/16")
+ adapter = OciRegistryAdapter("https://10.20.0.5:5000")
+ url = adapter._validate_outbound_url(
+ "https://10.20.0.5:5000/v2/_catalog?last=x"
+ )
+ assert url == "https://10.20.0.5:5000/v2/_catalog?last=x"
+
+ def test_allowlisted_resolved_hostname_permitted(self, monkeypatch):
+ monkeypatch.setenv(self.ENV, "10.20.0.0/16")
+ adapter = OciRegistryAdapter("https://harbor.internal")
+ with patch(
+ "prowler.providers.image.lib.registry.base.socket.getaddrinfo",
+ side_effect=_fake_getaddrinfo({"harbor.internal": "10.20.0.5"}),
+ ):
+ url = adapter._validate_outbound_url(
+ "https://harbor.internal/service/token"
+ )
+ assert url == "https://harbor.internal/service/token"
+
+ def test_private_ip_outside_allowlist_rejected(self, monkeypatch):
+ monkeypatch.setenv(self.ENV, "10.20.0.0/16")
+ adapter = OciRegistryAdapter("https://harbor.internal")
+ with patch(
+ "prowler.providers.image.lib.registry.base.socket.getaddrinfo",
+ side_effect=_fake_getaddrinfo(
+ {"harbor.internal": "10.20.0.5", "evil.internal": "192.168.1.99"}
+ ),
+ ):
+ with pytest.raises(ImageRegistryAuthError, match="non-public"):
+ adapter._validate_outbound_url("https://evil.internal/token")
+
+ def test_metadata_ip_rejected_unless_allowlisted(self, monkeypatch):
+ monkeypatch.setenv(self.ENV, "10.20.0.0/16")
+ adapter = OciRegistryAdapter("https://harbor.internal")
+ with patch(
+ "prowler.providers.image.lib.registry.base.socket.getaddrinfo",
+ side_effect=_fake_getaddrinfo({"harbor.internal": "10.20.0.5"}),
+ ):
+ with pytest.raises(ImageRegistryAuthError, match="non-public"):
+ adapter._validate_outbound_url("https://169.254.169.254/latest")
+
+ def test_private_link_from_public_origin_still_rejected(self, monkeypatch):
+ """Regression: the allowlist does not open ranges it does not name."""
+ monkeypatch.setenv(self.ENV, "10.20.0.0/16")
+ adapter = OciRegistryAdapter("https://reg.example.com")
+ with patch(
+ "prowler.providers.image.lib.registry.base.socket.getaddrinfo",
+ side_effect=_fake_getaddrinfo({"reg.example.com": "8.8.8.8"}),
+ ):
+ with pytest.raises(ImageRegistryAuthError, match="non-public"):
+ adapter._validate_outbound_url("http://192.168.0.99/v2/_catalog")
+
+ def test_local_tld_origin_enforcement_falls_back_to_allowlist(self, monkeypatch):
+ """Hosts without a registrable domain pass only if they resolve into the allowlist."""
+ monkeypatch.setenv(self.ENV, "10.20.0.0/16")
+ adapter = OciRegistryAdapter("https://registry.corp.local")
+ with patch(
+ "prowler.providers.image.lib.registry.base.socket.getaddrinfo",
+ side_effect=_fake_getaddrinfo(
+ {"registry.corp.local": "10.20.2.3", "auth.corp.local": "10.20.2.4"}
+ ),
+ ):
+ url = adapter._validate_outbound_url("https://auth.corp.local/token")
+ assert url == "https://auth.corp.local/token"
+
+ def test_malformed_allowlist_fails_loudly(self, monkeypatch):
+ monkeypatch.setenv(self.ENV, "10.20.0.0/16,banana")
+ with pytest.raises(ImageInvalidAllowedNetworksError, match="banana"):
+ OciRegistryAdapter("https://reg.example.com")
+
+ def test_allowlist_logged_as_relaxed_control(self, monkeypatch, caplog):
+ monkeypatch.setenv(self.ENV, "10.20.0.0/16")
+ with caplog.at_level("WARNING"):
+ OciRegistryAdapter("https://reg.example.com")
+ assert any(
+ "10.20.0.0/16" in message and "SSRF" in message
+ for message in caplog.messages
+ )
+
+
+class TestBasicAuthFallback:
+ """Registries like Harbor guard /_catalog behind Basic even when /v2/ negotiates Bearer."""
+
+ _BEARER_CHALLENGE = (
+ 'Bearer realm="https://reg.io/service/token",service="harbor-registry"'
+ )
+
+ def _harbor_responses(self):
+ ping = MagicMock(
+ status_code=401, headers={"Www-Authenticate": self._BEARER_CHALLENGE}
+ )
+ token = MagicMock(status_code=200)
+ token.json.return_value = {"token": "tok"}
+ catalog_401 = MagicMock(
+ status_code=401, headers={"Www-Authenticate": 'Basic realm="harbor"'}
+ )
+ return ping, token, catalog_401
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_catalog_falls_back_to_basic_when_bearer_rejected(self, mock_request):
+ ping, token, catalog_401 = self._harbor_responses()
+ catalog_ok = MagicMock(status_code=200, headers={})
+ catalog_ok.json.return_value = {"repositories": ["library/debian"]}
+ mock_request.side_effect = [
+ ping,
+ token,
+ catalog_401, # bearer without catalog scope
+ ping,
+ token,
+ catalog_401, # bearer retry, same result
+ catalog_ok, # basic fallback
+ ]
+
+ adapter = OciRegistryAdapter("reg.io", username="admin", password="secret")
+ repos = adapter.list_repositories()
+
+ assert repos == ["library/debian"]
+ assert mock_request.call_args.kwargs.get("auth") == ("admin", "secret")
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_basic_challenge_without_credentials_still_fails(self, mock_request):
+ ping, token, catalog_401 = self._harbor_responses()
+ mock_request.side_effect = [ping, token, catalog_401, ping, token, catalog_401]
+
+ adapter = OciRegistryAdapter("reg.io")
+ with pytest.raises(ImageRegistryAuthError, match="catalog listing"):
+ adapter.list_repositories()
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_fallback_with_combined_multi_challenge_header(self, mock_request):
+ # Basic not first in the header must still trigger the fallback
+ ping, token, _ = self._harbor_responses()
+ catalog_401 = MagicMock(
+ status_code=401,
+ headers={
+ "Www-Authenticate": f'{self._BEARER_CHALLENGE}, Basic realm="harbor"'
+ },
+ )
+ catalog_ok = MagicMock(status_code=200, headers={})
+ catalog_ok.json.return_value = {"repositories": ["library/debian"]}
+ mock_request.side_effect = [
+ ping,
+ token,
+ catalog_401,
+ ping,
+ token,
+ catalog_401,
+ catalog_ok,
+ ]
+
+ adapter = OciRegistryAdapter("reg.io", username="admin", password="secret")
+ repos = adapter.list_repositories()
+
+ assert repos == ["library/debian"]
+ assert mock_request.call_args.kwargs.get("auth") == ("admin", "secret")
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_multi_page_catalog_falls_back_only_once(self, mock_request):
+ ping, token, catalog_401 = self._harbor_responses()
+ page1 = MagicMock(
+ status_code=200,
+ ok=True,
+ headers={"Link": '; rel="next"'},
+ )
+ page1.json.return_value = {"repositories": ["a"]}
+ page2 = MagicMock(status_code=200, ok=True, headers={})
+ page2.json.return_value = {"repositories": ["b"]}
+ mock_request.side_effect = [
+ ping,
+ token,
+ catalog_401, # bearer without catalog scope
+ ping,
+ token,
+ catalog_401, # bearer retry, same result
+ page1, # basic fallback succeeds -> basic mode persists
+ page2, # second page goes straight to basic
+ ]
+
+ adapter = OciRegistryAdapter("reg.io", username="admin", password="secret")
+ repos = adapter.list_repositories()
+
+ assert repos == ["a", "b"]
+ assert mock_request.call_count == 8
+ assert adapter._basic_auth_verified is True
+ assert adapter._bearer_token is None
+ page2_call = mock_request.call_args_list[-1]
+ assert page2_call.kwargs.get("auth") == ("admin", "secret")
+ assert "Authorization" not in page2_call.kwargs.get("headers", {})
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_basic_fallback_not_sent_cross_origin(self, mock_request):
+ catalog_401 = MagicMock(
+ status_code=401, headers={"Www-Authenticate": 'Basic realm="other"'}
+ )
+ mock_request.return_value = catalog_401
+
+ adapter = OciRegistryAdapter("reg.io", username="admin", password="secret")
+ adapter._bearer_token = "tok"
+ resp = adapter._authed_request("GET", "https://other.example.com/v2/_catalog")
+
+ assert resp.status_code == 401
+ assert all(
+ call.kwargs.get("auth") is None for call in mock_request.call_args_list
+ )
+
+
+class TestBearerAuthSwitch:
+ """Registries that negotiate Basic on /v2/ but demand Bearer on other endpoints."""
+
+ _BEARER_CHALLENGE = 'Bearer realm="https://reg.io/token",service="registry",scope="registry:catalog:*"'
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_catalog_switches_to_bearer_when_basic_rejected(self, mock_request):
+ ping = MagicMock(
+ status_code=401, headers={"Www-Authenticate": 'Basic realm="registry"'}
+ )
+ catalog_401 = MagicMock(
+ status_code=401, headers={"Www-Authenticate": self._BEARER_CHALLENGE}
+ )
+ token = MagicMock(status_code=200)
+ token.json.return_value = {"token": "switched-tok"}
+ catalog_ok = MagicMock(status_code=200, headers={})
+ catalog_ok.json.return_value = {"repositories": ["library/debian"]}
+ mock_request.side_effect = [ping, catalog_401, token, catalog_ok]
+
+ adapter = OciRegistryAdapter("reg.io", username="admin", password="secret")
+ repos = adapter.list_repositories()
+
+ assert repos == ["library/debian"]
+ assert adapter._bearer_token == "switched-tok"
+ # Token exchange carries the credentials
+ token_call = mock_request.call_args_list[2]
+ assert token_call.kwargs.get("auth") == ("admin", "secret")
+ assert token_call.kwargs.get("params", {}).get("scope") == "registry:catalog:*"
+ # The retry uses the Bearer header, not Basic
+ retry_call = mock_request.call_args_list[3]
+ assert retry_call.kwargs.get("auth") is None
+ assert (
+ retry_call.kwargs.get("headers", {}).get("Authorization")
+ == "Bearer switched-tok"
+ )
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_multi_page_catalog_switches_only_once(self, mock_request):
+ ping = MagicMock(
+ status_code=401, headers={"Www-Authenticate": 'Basic realm="registry"'}
+ )
+ catalog_401 = MagicMock(
+ status_code=401, headers={"Www-Authenticate": self._BEARER_CHALLENGE}
+ )
+ token = MagicMock(status_code=200)
+ token.json.return_value = {"token": "switched-tok"}
+ page1 = MagicMock(
+ status_code=200,
+ headers={"Link": '; rel="next"'},
+ )
+ page1.json.return_value = {"repositories": ["a"]}
+ page2 = MagicMock(status_code=200, headers={})
+ page2.json.return_value = {"repositories": ["b"]}
+ mock_request.side_effect = [ping, catalog_401, token, page1, page2]
+
+ adapter = OciRegistryAdapter("reg.io", username="admin", password="secret")
+ repos = adapter.list_repositories()
+
+ assert repos == ["a", "b"]
+ assert mock_request.call_count == 5
+ page2_call = mock_request.call_args_list[-1]
+ assert (
+ page2_call.kwargs.get("headers", {}).get("Authorization")
+ == "Bearer switched-tok"
+ )
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_anonymous_switch_to_bearer(self, mock_request):
+ ping = MagicMock(status_code=200)
+ catalog_401 = MagicMock(
+ status_code=401, headers={"Www-Authenticate": self._BEARER_CHALLENGE}
+ )
+ token = MagicMock(status_code=200)
+ token.json.return_value = {"token": "anon-tok"}
+ catalog_ok = MagicMock(status_code=200, headers={})
+ catalog_ok.json.return_value = {"repositories": ["public/app"]}
+ mock_request.side_effect = [ping, catalog_401, token, catalog_ok]
+
+ adapter = OciRegistryAdapter("reg.io")
+ repos = adapter.list_repositories()
+
+ assert repos == ["public/app"]
+ token_call = mock_request.call_args_list[2]
+ assert token_call.kwargs.get("auth") is None
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_switch_not_attempted_cross_origin(self, mock_request):
+ catalog_401 = MagicMock(
+ status_code=401, headers={"Www-Authenticate": self._BEARER_CHALLENGE}
+ )
+ mock_request.return_value = catalog_401
+
+ adapter = OciRegistryAdapter("reg.io", username="admin", password="secret")
+ adapter._basic_auth_verified = True
+ resp = adapter._authed_request("GET", "https://other.example.com/v2/_catalog")
+
+ assert resp.status_code == 401
+ assert adapter._bearer_token is None
+ assert mock_request.call_count == 1
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_switch_with_combined_multi_challenge_header(self, mock_request):
+ # RFC 7235: multiple challenges in one header, Basic first
+ combined = (
+ 'Basic realm="registry", '
+ 'Bearer realm="http://reg.io/token",service="registry",scope="registry:catalog:*"'
+ )
+ ping = MagicMock(
+ status_code=401, headers={"Www-Authenticate": 'Basic realm="registry"'}
+ )
+ catalog_401 = MagicMock(status_code=401, headers={"Www-Authenticate": combined})
+ token = MagicMock(status_code=200)
+ token.json.return_value = {"token": "combined-tok"}
+ catalog_ok = MagicMock(status_code=200, headers={})
+ catalog_ok.json.return_value = {"repositories": ["library/debian"]}
+ mock_request.side_effect = [ping, catalog_401, token, catalog_ok]
+
+ adapter = OciRegistryAdapter(
+ "http://reg.io", username="admin", password="secret"
+ )
+ repos = adapter.list_repositories()
+
+ assert repos == ["library/debian"]
+ # The token exchange must hit the Bearer realm, not Basic's realm="registry"
+ token_call = mock_request.call_args_list[2]
+ assert token_call.args[1] == "http://reg.io/token"
+ assert token_call.kwargs.get("params", {}).get("scope") == "registry:catalog:*"
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_ping_with_combined_multi_challenge_prefers_bearer(self, mock_request):
+ combined = 'Basic realm="registry", Bearer realm="https://auth.reg.io/token",service="registry"'
+ ping = MagicMock(status_code=401, headers={"Www-Authenticate": combined})
+ token = MagicMock(status_code=200)
+ token.json.return_value = {"token": "bearer-tok"}
+ mock_request.side_effect = [ping, token]
+
+ adapter = OciRegistryAdapter("reg.io", username="u", password="p")
+ adapter._ensure_auth()
+
+ assert adapter._bearer_token == "bearer-tok"
+ token_call = mock_request.call_args_list[1]
+ assert token_call.args[1] == "https://auth.reg.io/token"
+
+ @patch("prowler.providers.image.lib.registry.base.requests.request")
+ def test_failed_token_exchange_raises_auth_error(self, mock_request):
+ ping = MagicMock(
+ status_code=401, headers={"Www-Authenticate": 'Basic realm="registry"'}
+ )
+ catalog_401 = MagicMock(
+ status_code=401, headers={"Www-Authenticate": self._BEARER_CHALLENGE}
+ )
+ token_denied = MagicMock(status_code=401)
+ mock_request.side_effect = [ping, catalog_401, token_denied]
+
+ adapter = OciRegistryAdapter("reg.io", username="admin", password="wrong")
+ with pytest.raises(ImageRegistryAuthError, match="bearer token"):
+ adapter.list_repositories()