fix(image): harden registry adapter layer against SSRF, credential leaks, and edge cases

- Add SSRF protection for bearer token realm URLs (reject private/loopback IPs, non-http(s) schemes)
- Protect credentials from serialization via __getstate__ redaction and private attrs with properties
- Change missing DockerHub namespace exception from AuthError to CatalogError
- Add empty-token guards in OCI bearer flow and DockerHub JWT/registry token endpoints
- Narrow except Exception to except (ValueError, UnicodeDecodeError) in base64 decode

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

@@ -28,17 +28,31 @@ class RegistryAdapter(ABC):
     ) -> None:
         self.registry_url = registry_url
         self.username = username
-        self.password = password
-        self.token = token
+        self._password = password
+        self._token = token
         self.verify_ssl = verify_ssl
 
+    @property
+    def password(self) -> str | None:
+        return self._password
+
+    @property
+    def token(self) -> str | None:
+        return self._token
+
+    def __getstate__(self) -> dict:
+        state = self.__dict__.copy()
+        state["_password"] = "***" if state.get("_password") else None
+        state["_token"] = "***" if state.get("_token") else None
+        return state
+
     def __repr__(self) -> str:
         return (
             f"{self.__class__.__name__}("
             f"registry_url={self.registry_url!r}, "
             f"username={self.username!r}, "
-            f"password={'<redacted>' if self.password else None}, "
-            f"token={'<redacted>' if self.token else None})"
+            f"password={'<redacted>' if self._password else None}, "
+            f"token={'<redacted>' if self._token else None})"
         )
 
     @abstractmethod

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

@@ -64,7 +64,7 @@ class DockerHubAdapter(RegistryAdapter):
 
     def list_repositories(self) -> list[str]:
         if not self.namespace:
-            raise ImageRegistryAuthError(
+            raise ImageRegistryCatalogError(
                 file=__file__,
                 message="Docker Hub requires a namespace. Use --registry docker.io/{org_or_user}.",
             )
@@ -127,6 +127,11 @@ class DockerHubAdapter(RegistryAdapter):
                 message=f"Docker Hub login failed (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
             )
         self._hub_jwt = resp.json().get("token")
+        if not self._hub_jwt:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message="Docker Hub login returned an empty JWT token. Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
+            )
 
     def _get_registry_token(self, repository: str) -> str:
         if repository in self._registry_tokens:
@@ -151,6 +156,11 @@ class DockerHubAdapter(RegistryAdapter):
                 message=f"Failed to obtain Docker Hub registry token for {repository} (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
             )
         token = resp.json().get("token", "")
+        if not token:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Docker Hub registry token endpoint returned an empty token for {repository}. Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
+            )
         self._registry_tokens[repository] = token
         return token
 

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

@@ -3,8 +3,10 @@
 from __future__ import annotations
 
 import base64
+import ipaddress
 import re
 from typing import TYPE_CHECKING
+from urllib.parse import urlparse
 
 from prowler.lib.logger import logger
 from prowler.providers.image.exceptions.exceptions import (
@@ -126,6 +128,7 @@ class OciRegistryAdapter(RegistryAdapter):
                 message=f"Cannot parse token endpoint from registry {self.registry_url}. Www-Authenticate: {www_authenticate[:200]}",
             )
         realm = match.group(1)
+        self._validate_realm_url(realm)
         params: dict = {}
         service_match = re.search(r'service="([^"]+)"', www_authenticate)
         if service_match:
@@ -145,7 +148,36 @@ class OciRegistryAdapter(RegistryAdapter):
                 message=f"Failed to obtain bearer token from {realm} (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
             )
         data = resp.json()
-        return data.get("token") or data.get("access_token", "")
+        token = data.get("token") or data.get("access_token", "")
+        if not token:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Token endpoint {realm} returned an empty token. Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
+            )
+        return token
+
+    @staticmethod
+    def _validate_realm_url(realm: str) -> None:
+        parsed = urlparse(realm)
+        if parsed.scheme not in ("http", "https"):
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Bearer token realm has disallowed scheme: {parsed.scheme}. Only http/https are allowed.",
+            )
+        if parsed.scheme == "http":
+            logger.warning(
+                f"Bearer token realm uses HTTP (not HTTPS): {realm}"
+            )
+        hostname = parsed.hostname or ""
+        try:
+            addr = ipaddress.ip_address(hostname)
+            if addr.is_private or addr.is_loopback or addr.is_link_local:
+                raise ImageRegistryAuthError(
+                    file=__file__,
+                    message=f"Bearer token realm points to a private/loopback address: {hostname}. This may indicate an SSRF attempt.",
+                )
+        except ValueError:
+            pass
 
     def _resolve_basic_credentials(self) -> tuple[str | None, str | None]:
         """Decode pre-encoded base64 auth tokens (e.g., from aws ecr get-authorization-token).
@@ -159,7 +191,7 @@ class OciRegistryAdapter(RegistryAdapter):
             decoded = base64.b64decode(self.password).decode("utf-8")
             if decoded.startswith(f"{self.username}:"):
                 return self.username, decoded[len(self.username) + 1 :]
-        except Exception:
+        except (ValueError, UnicodeDecodeError):
             logger.debug("Password is not a base64-encoded auth token, using as-is")
         return self.username, self.password