feat(image): add registry scan mode with OCI, Docker Hub, and ECR support

Add pluggable registry adapter pattern for enumerating and scanning all
container images in a private registry. Supports:
- Generic OCI registries (Harbor, GitLab, JFrog) via /v2/_catalog
- Docker Hub via Hub REST API with namespace-based enumeration
- AWS ECR via standard OCI endpoints with ECR auth token

New CLI flags: --registry, --image-filter, --tag-filter, --max-images,
--registry-insecure

prowler/CHANGELOG.md

@@ -17,6 +17,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - OCI regions updater script and CI workflow [(#10020)](https://github.com/prowler-cloud/prowler/pull/10020)
 - `image` provider for container image scanning with Trivy integration [(#9984)](https://github.com/prowler-cloud/prowler/pull/9984)
 - Private registry authentication for the Image provider via environment variables (`REGISTRY_USERNAME`/`REGISTRY_PASSWORD`, `REGISTRY_TOKEN`) [(#9985)](https://github.com/prowler-cloud/prowler/pull/9985)
+- Registry scan mode for Image provider: enumerate and scan all images from OCI, Docker Hub, and ECR registries with `--registry`, `--image-filter`, `--tag-filter`, `--max-images`, and `--registry-insecure` flags
 - CSA CCM 4.0 for the Alibaba Cloud provider [(#10061)](https://github.com/prowler-cloud/prowler/pull/10061)
 
 ### 🔄 Changed

prowler/providers/common/provider.py

@@ -285,6 +285,16 @@ class Provider(ABC):
                         timeout=arguments.timeout,
                         config_path=arguments.config_file,
                         fixer_config=fixer_config,
+                        registry_username=getattr(arguments, "registry_username", None),
+                        registry_password=getattr(arguments, "registry_password", None),
+                        registry_token=getattr(arguments, "registry_token", None),
+                        registry=getattr(arguments, "registry", None),
+                        image_filter=getattr(arguments, "image_filter", None),
+                        tag_filter=getattr(arguments, "tag_filter", None),
+                        max_images=getattr(arguments, "max_images", 0),
+                        registry_insecure=getattr(
+                            arguments, "registry_insecure", False
+                        ),
                     )
                 elif "mongodbatlas" in provider_class_name.lower():
                     provider_class(

prowler/providers/image/exceptions/exceptions.py

@@ -58,6 +58,26 @@ class ImageBaseException(ProwlerException):
             "message": "Docker binary not found.",
             "remediation": "Install Docker to enable private registry authentication via docker login.",
         },
+        (11013, "ImageRegistryAuthError"): {
+            "message": "Registry authentication failed.",
+            "remediation": "Check REGISTRY_USERNAME/REGISTRY_PASSWORD or REGISTRY_TOKEN environment variables.",
+        },
+        (11014, "ImageRegistryCatalogError"): {
+            "message": "Registry does not support catalog listing.",
+            "remediation": "Use --image or --image-list instead of --registry.",
+        },
+        (11015, "ImageRegistryNetworkError"): {
+            "message": "Network error communicating with registry.",
+            "remediation": "Check registry URL and network connectivity.",
+        },
+        (11016, "ImageMaxImagesExceededError"): {
+            "message": "Discovered images exceed --max-images limit.",
+            "remediation": "Use --image-filter or --tag-filter to narrow results, or increase --max-images.",
+        },
+        (11017, "ImageInvalidFilterError"): {
+            "message": "Invalid regex filter pattern.",
+            "remediation": "Check the regex syntax for --image-filter or --tag-filter.",
+        },
     }
 
     def __init__(self, code, file=None, original_exception=None, message=None):
@@ -188,3 +208,48 @@ class ImageDockerNotFoundError(ImageBaseException):
         super().__init__(
             11012, file=file, original_exception=original_exception, message=message
         )
+
+
+class ImageRegistryAuthError(ImageBaseException):
+    """Exception raised when registry authentication fails."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11013, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageRegistryCatalogError(ImageBaseException):
+    """Exception raised when registry does not support catalog listing."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11014, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageRegistryNetworkError(ImageBaseException):
+    """Exception raised when a network error occurs communicating with a registry."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11015, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageMaxImagesExceededError(ImageBaseException):
+    """Exception raised when discovered images exceed --max-images limit."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11016, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageInvalidFilterError(ImageBaseException):
+    """Exception raised when an invalid regex filter pattern is provided."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11017, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/image/image_provider.py

@@ -24,12 +24,14 @@ from prowler.providers.image.exceptions.exceptions import (
     ImageDockerNotFoundError,
     ImageFindingProcessingError,
     ImageInvalidConfigScannerError,
+    ImageInvalidFilterError,
     ImageInvalidNameError,
     ImageInvalidScannerError,
     ImageInvalidSeverityError,
     ImageInvalidTimeoutError,
     ImageListFileNotFoundError,
     ImageListFileReadError,
+    ImageMaxImagesExceededError,
     ImageNoImagesProvidedError,
     ImageScanError,
     ImageTrivyBinaryNotFoundError,
@@ -39,6 +41,7 @@ from prowler.providers.image.lib.arguments.arguments import (
     SCANNERS_CHOICES,
     SEVERITY_CHOICES,
 )
+from prowler.providers.image.lib.registry.factory import create_registry_adapter
 
 
 class ImageProvider(Provider):
@@ -72,6 +75,11 @@ class ImageProvider(Provider):
         registry_username: str | None = None,
         registry_password: str | None = None,
         registry_token: str | None = None,
+        registry: str | None = None,
+        image_filter: str | None = None,
+        tag_filter: str | None = None,
+        max_images: int = 0,
+        registry_insecure: bool = False,
     ):
         logger.info("Instantiating Image Provider...")
 
@@ -90,8 +98,12 @@ class ImageProvider(Provider):
         self._identity = "prowler"
 
         # Registry authentication (follows IaC pattern: explicit params, env vars internal)
-        self.registry_username = registry_username or os.environ.get("REGISTRY_USERNAME")
-        self.registry_password = registry_password or os.environ.get("REGISTRY_PASSWORD")
+        self.registry_username = registry_username or os.environ.get(
+            "REGISTRY_USERNAME"
+        )
+        self.registry_password = registry_password or os.environ.get(
+            "REGISTRY_PASSWORD"
+        )
         self.registry_token = registry_token or os.environ.get("REGISTRY_TOKEN")
         self._logged_in_registries: set = set()
 
@@ -104,12 +116,43 @@ class ImageProvider(Provider):
         else:
             self._auth_method = "No auth"
 
+        # Registry scan mode
+        self.registry = registry
+        self.image_filter = image_filter
+        self.tag_filter = tag_filter
+        self.max_images = max_images
+        self.registry_insecure = registry_insecure
+
+        # Compile regex filters
+        self._image_filter_re = None
+        self._tag_filter_re = None
+        if self.image_filter:
+            try:
+                self._image_filter_re = re.compile(self.image_filter)
+            except re.error as exc:
+                raise ImageInvalidFilterError(
+                    file=__file__,
+                    message=f"Invalid --image-filter regex '{self.image_filter}': {exc}",
+                )
+        if self.tag_filter:
+            try:
+                self._tag_filter_re = re.compile(self.tag_filter)
+            except re.error as exc:
+                raise ImageInvalidFilterError(
+                    file=__file__,
+                    message=f"Invalid --tag-filter regex '{self.tag_filter}': {exc}",
+                )
+
         self._validate_inputs()
 
         # Load images from file if provided
         if image_list_file:
             self._load_images_from_file(image_list_file)
 
+        # Registry scan mode: enumerate images from registry
+        if self.registry:
+            self._enumerate_registry()
+
         for image in self.images:
             self._validate_image_name(image)
 
@@ -739,6 +782,83 @@ class ImageProvider(Provider):
 
         return error_msg
 
+    def _enumerate_registry(self) -> None:
+        """Enumerate images from a registry using the appropriate adapter."""
+        verify_ssl = not self.registry_insecure
+        adapter = create_registry_adapter(
+            registry_url=self.registry,
+            username=self.registry_username,
+            password=self.registry_password,
+            token=self.registry_token,
+            verify_ssl=verify_ssl,
+        )
+
+        repositories = adapter.list_repositories()
+        logger.info(
+            f"Discovered {len(repositories)} repositories from registry {self.registry}"
+        )
+
+        # Apply image filter
+        if self._image_filter_re:
+            repositories = [r for r in repositories if self._image_filter_re.search(r)]
+            logger.info(
+                f"{len(repositories)} repositories match --image-filter '{self.image_filter}'"
+            )
+
+        if not repositories:
+            logger.warning(
+                f"No repositories found in registry {self.registry} (after filtering)"
+            )
+            return
+
+        # Determine if this is a Docker Hub adapter (for image reference format)
+        from prowler.providers.image.lib.registry.dockerhub_adapter import (
+            DockerHubAdapter,
+        )
+
+        is_dockerhub = isinstance(adapter, DockerHubAdapter)
+
+        discovered_images = []
+        for repo in repositories:
+            tags = adapter.list_tags(repo)
+
+            # Apply tag filter
+            if self._tag_filter_re:
+                tags = [t for t in tags if self._tag_filter_re.search(t)]
+
+            for tag in tags:
+                if is_dockerhub:
+                    # Docker Hub images don't need a host prefix
+                    image_ref = f"{repo}:{tag}"
+                else:
+                    # OCI registries need the full host/repo:tag reference
+                    registry_host = self.registry.rstrip("/")
+                    for prefix in ("https://", "http://"):
+                        if registry_host.startswith(prefix):
+                            registry_host = registry_host[len(prefix) :]
+                            break
+                    image_ref = f"{registry_host}/{repo}:{tag}"
+                discovered_images.append(image_ref)
+
+        # Check max-images limit
+        if self.max_images and len(discovered_images) > self.max_images:
+            raise ImageMaxImagesExceededError(
+                file=__file__,
+                message=f"Discovered {len(discovered_images)} images, exceeding --max-images {self.max_images}. Use --image-filter or --tag-filter to narrow results.",
+            )
+
+        # Deduplicate with explicit images
+        existing = set(self.images)
+        for img in discovered_images:
+            if img not in existing:
+                self.images.append(img)
+                existing.add(img)
+
+        logger.info(
+            f"Discovered {len(discovered_images)} images from registry {self.registry} "
+            f"({len(repositories)} repositories). Total images to scan: {len(self.images)}"
+        )
+
     def print_credentials(self) -> None:
         """Print scan configuration."""
         report_title = f"{Style.BRIGHT}Scanning container images:{Style.RESET_ALL}"
@@ -775,6 +895,19 @@ class ImageProvider(Provider):
             f"Authentication method: {Fore.YELLOW}{self.auth_method}{Style.RESET_ALL}"
         )
 
+        if self.registry:
+            report_lines.append(
+                f"Registry: {Fore.YELLOW}{self.registry}{Style.RESET_ALL}"
+            )
+            if self.image_filter:
+                report_lines.append(
+                    f"Image filter: {Fore.YELLOW}{self.image_filter}{Style.RESET_ALL}"
+                )
+            if self.tag_filter:
+                report_lines.append(
+                    f"Tag filter: {Fore.YELLOW}{self.tag_filter}{Style.RESET_ALL}"
+                )
+
         print_boxes(report_lines, report_title)
 
     @staticmethod

prowler/providers/image/lib/arguments/arguments.py

@@ -88,16 +88,85 @@ def init_parser(self):
         help="Trivy scan timeout. Default: 5m. Examples: 10m, 1h",
     )
 
+    # Registry Scan Mode
+    registry_group = image_parser.add_argument_group("Registry Scan Mode")
+    registry_group.add_argument(
+        "--registry",
+        dest="registry",
+        default=None,
+        help="Registry URL to enumerate and scan all images. Examples: myregistry.io, docker.io/myorg, 123456789.dkr.ecr.us-east-1.amazonaws.com",
+    )
+    registry_group.add_argument(
+        "--image-filter",
+        dest="image_filter",
+        default=None,
+        help="Regex to filter repository names during registry enumeration (re.search). Example: '^prod/.*'",
+    )
+    registry_group.add_argument(
+        "--tag-filter",
+        dest="tag_filter",
+        default=None,
+        help=r"Regex to filter tags during registry enumeration (re.search). Example: '^(latest|v\d+\.\d+\.\d+)$'",
+    )
+    registry_group.add_argument(
+        "--max-images",
+        dest="max_images",
+        type=int,
+        default=0,
+        help="Maximum number of images to scan from registry. 0 = unlimited. Aborts if exceeded.",
+    )
+    registry_group.add_argument(
+        "--registry-insecure",
+        dest="registry_insecure",
+        action="store_true",
+        default=False,
+        help="Skip TLS verification for registry connections (for self-signed certificates).",
+    )
+
 
 def validate_arguments(arguments):
     """Validate Image provider arguments."""
     images = getattr(arguments, "images", [])
     image_list_file = getattr(arguments, "image_list_file", None)
+    registry = getattr(arguments, "registry", None)
+    image_filter = getattr(arguments, "image_filter", None)
+    tag_filter = getattr(arguments, "tag_filter", None)
+    max_images = getattr(arguments, "max_images", 0)
+    registry_insecure = getattr(arguments, "registry_insecure", False)
 
-    if not images and not image_list_file:
+    if not images and not image_list_file and not registry:
         return (
             False,
-            "At least one image must be specified using --image (-I) or --image-list.",
+            "At least one image source must be specified using --image (-I), --image-list, or --registry.",
         )
 
+    # Registry-only flags require --registry
+    if not registry:
+        if image_filter:
+            return (False, "--image-filter requires --registry.")
+        if tag_filter:
+            return (False, "--tag-filter requires --registry.")
+        if max_images:
+            return (False, "--max-images requires --registry.")
+        if registry_insecure:
+            return (False, "--registry-insecure requires --registry.")
+
+    # Docker Hub namespace validation
+    if registry:
+        url = registry.rstrip("/")
+        for prefix in ("https://", "http://"):
+            if url.startswith(prefix):
+                url = url[len(prefix) :]
+                break
+        stripped = url
+        for prefix in ("registry-1.docker.io", "docker.io"):
+            if stripped.startswith(prefix):
+                stripped = stripped[len(prefix) :].lstrip("/")
+                if not stripped:
+                    return (
+                        False,
+                        "Docker Hub requires a namespace. Use --registry docker.io/{org_or_user}.",
+                    )
+                break
+
     return (True, "")

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

@@ -0,0 +1,28 @@
+"""Registry adapter abstract base class."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+
+class RegistryAdapter(ABC):
+    """Abstract base class for registry adapters."""
+
+    def __init__(
+        self, registry_url, username=None, password=None, token=None, verify_ssl=True
+    ):
+        self.registry_url = registry_url
+        self.username = username
+        self.password = password
+        self.token = token
+        self.verify_ssl = verify_ssl
+
+    @abstractmethod
+    def list_repositories(self) -> list[str]:
+        """Enumerate all repository names in the registry."""
+        ...
+
+    @abstractmethod
+    def list_tags(self, repository: str) -> list[str]:
+        """Enumerate all tags for a repository."""
+        ...

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

@@ -0,0 +1,221 @@
+"""Docker Hub registry adapter."""
+
+from __future__ import annotations
+
+import re
+import time
+
+import requests
+
+from prowler.lib.logger import logger
+from prowler.providers.image.exceptions.exceptions import (
+    ImageRegistryAuthError,
+    ImageRegistryNetworkError,
+)
+from prowler.providers.image.lib.registry.base import RegistryAdapter
+
+_MAX_RETRIES = 3
+_BACKOFF_BASE = 1
+_HUB_API = "https://hub.docker.com"
+_REGISTRY_HOST = "https://registry-1.docker.io"
+_AUTH_URL = "https://auth.docker.io/token"
+
+
+class DockerHubAdapter(RegistryAdapter):
+    """Adapter for Docker Hub using the Hub REST API + OCI tag listing."""
+
+    def __init__(
+        self, registry_url, username=None, password=None, token=None, verify_ssl=True
+    ):
+        super().__init__(registry_url, username, password, token, verify_ssl)
+        self.namespace = self._extract_namespace(registry_url)
+        self._hub_jwt = None
+        self._registry_tokens = {}
+
+    @staticmethod
+    def _extract_namespace(registry_url):
+        url = registry_url.rstrip("/")
+        for prefix in (
+            "https://registry-1.docker.io",
+            "http://registry-1.docker.io",
+            "https://docker.io",
+            "http://docker.io",
+            "registry-1.docker.io",
+            "docker.io",
+            "https://",
+            "http://",
+        ):
+            if url.startswith(prefix):
+                url = url[len(prefix) :]
+                break
+        url = url.lstrip("/")
+        parts = url.split("/")
+        namespace = parts[0] if parts and parts[0] else ""
+        return namespace
+
+    def list_repositories(self):
+        if not self.namespace:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message="Docker Hub requires a namespace. Use --registry docker.io/{org_or_user}.",
+            )
+        self._hub_login()
+        repositories = []
+        url = f"{_HUB_API}/v2/namespaces/{self.namespace}/repositories"
+        params = {"page_size": 100}
+        while url:
+            resp = self._hub_request("GET", url, params=params)
+            self._check_hub_response(resp, "repository listing")
+            data = resp.json()
+            for repo in data.get("results", []):
+                name = repo.get("name", "")
+                if name:
+                    repositories.append(f"{self.namespace}/{name}")
+            url = data.get("next")
+            params = {}
+        return repositories
+
+    def list_tags(self, repository):
+        token = self._get_registry_token(repository)
+        tags = []
+        url = f"{_REGISTRY_HOST}/v2/{repository}/tags/list"
+        params = {"n": 100}
+        while url:
+            resp = self._registry_request("GET", url, token, params=params)
+            if resp.status_code in (401, 403):
+                raise ImageRegistryAuthError(
+                    file=__file__,
+                    message=f"Authentication failed for tag listing of {repository} on Docker Hub. Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
+                )
+            if resp.status_code != 200:
+                logger.warning(
+                    f"Failed to list tags for {repository} (HTTP {resp.status_code}): {resp.text[:200]}"
+                )
+                break
+            data = resp.json()
+            tags.extend(data.get("tags", []) or [])
+            url = self._next_page_url(resp)
+            params = {}
+        return tags
+
+    def _hub_login(self):
+        if self._hub_jwt:
+            return
+        if not self.username or not self.password:
+            return
+        resp = requests.post(
+            f"{_HUB_API}/v2/users/login",
+            json={"username": self.username, "password": self.password},
+            verify=self.verify_ssl,
+            timeout=30,
+        )
+        if resp.status_code != 200:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Docker Hub login failed (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
+            )
+        self._hub_jwt = resp.json().get("token")
+
+    def _get_registry_token(self, repository):
+        if repository in self._registry_tokens:
+            return self._registry_tokens[repository]
+        params = {
+            "service": "registry.docker.io",
+            "scope": f"repository:{repository}:pull",
+        }
+        auth = None
+        if self.username and self.password:
+            auth = (self.username, self.password)
+        resp = requests.get(
+            _AUTH_URL, params=params, auth=auth, verify=self.verify_ssl, timeout=30
+        )
+        if resp.status_code != 200:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                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", "")
+        self._registry_tokens[repository] = token
+        return token
+
+    def _hub_request(self, method, url, **kwargs):
+        kwargs.setdefault("timeout", 30)
+        kwargs.setdefault("verify", self.verify_ssl)
+        headers = kwargs.pop("headers", {})
+        if self._hub_jwt:
+            headers["Authorization"] = f"Bearer {self._hub_jwt}"
+        kwargs["headers"] = headers
+        return self._request_with_retry(method, url, **kwargs)
+
+    def _registry_request(self, method, url, token, **kwargs):
+        kwargs.setdefault("timeout", 30)
+        kwargs.setdefault("verify", self.verify_ssl)
+        headers = kwargs.pop("headers", {})
+        headers["Authorization"] = f"Bearer {token}"
+        kwargs["headers"] = headers
+        return self._request_with_retry(method, url, **kwargs)
+
+    def _request_with_retry(self, method, url, **kwargs):
+        last_exception = None
+        for attempt in range(1, _MAX_RETRIES + 1):
+            try:
+                resp = requests.request(method, url, **kwargs)
+                if resp.status_code == 429:
+                    wait = _BACKOFF_BASE * (2 ** (attempt - 1))
+                    logger.warning(
+                        f"Rate limited by Docker Hub, retrying in {wait}s (attempt {attempt}/{_MAX_RETRIES})"
+                    )
+                    time.sleep(wait)
+                    continue
+                return resp
+            except requests.exceptions.ConnectionError as exc:
+                last_exception = exc
+                if attempt < _MAX_RETRIES:
+                    wait = _BACKOFF_BASE * (2 ** (attempt - 1))
+                    logger.warning(
+                        f"Connection error to Docker Hub, retrying in {wait}s (attempt {attempt}/{_MAX_RETRIES})"
+                    )
+                    time.sleep(wait)
+                    continue
+            except requests.exceptions.Timeout as exc:
+                raise ImageRegistryNetworkError(
+                    file=__file__,
+                    message="Connection timed out to Docker Hub.",
+                    original_exception=exc,
+                )
+        raise ImageRegistryNetworkError(
+            file=__file__,
+            message=f"Failed to connect to Docker Hub after {_MAX_RETRIES} attempts.",
+            original_exception=last_exception,
+        )
+
+    def _check_hub_response(self, resp, context):
+        if resp.status_code == 200:
+            return
+        if resp.status_code in (401, 403):
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Authentication failed for {context} on Docker Hub (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD environment variables.",
+            )
+        if resp.status_code == 404:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Namespace '{self.namespace}' not found on Docker Hub. Check the namespace in --registry docker.io/{{namespace}}.",
+            )
+        raise ImageRegistryNetworkError(
+            file=__file__,
+            message=f"Unexpected error during {context} on Docker Hub (HTTP {resp.status_code}): {resp.text[:200]}",
+        )
+
+    @staticmethod
+    def _next_page_url(resp):
+        link_header = resp.headers.get("Link", "")
+        if not link_header:
+            return None
+        match = re.search(r'<([^>]+)>;\s*rel="next"', link_header)
+        if match:
+            next_url = match.group(1)
+            if next_url.startswith("/"):
+                return f"{_REGISTRY_HOST}{next_url}"
+            return next_url
+        return None

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

@@ -0,0 +1,45 @@
+"""Factory for auto-detecting registry type and returning the appropriate adapter."""
+
+from __future__ import annotations
+
+import re
+
+from prowler.providers.image.lib.registry.dockerhub_adapter import DockerHubAdapter
+from prowler.providers.image.lib.registry.oci_adapter import OciRegistryAdapter
+
+_DOCKER_HUB_PATTERN = re.compile(
+    r"^(https?://)?(docker\.io|registry-1\.docker\.io)(/|$)", re.IGNORECASE
+)
+_ECR_PATTERN = re.compile(
+    r"^(https?://)?\d+\.dkr\.ecr\.[a-z0-9-]+\.amazonaws\.com(/|$)", re.IGNORECASE
+)
+
+
+def create_registry_adapter(
+    registry_url, username=None, password=None, token=None, verify_ssl=True
+):
+    """Auto-detect registry type from URL and return the appropriate adapter."""
+    if _DOCKER_HUB_PATTERN.search(registry_url):
+        return DockerHubAdapter(
+            registry_url=registry_url,
+            username=username,
+            password=password,
+            token=token,
+            verify_ssl=verify_ssl,
+        )
+    return OciRegistryAdapter(
+        registry_url=registry_url,
+        username=username,
+        password=password,
+        token=token,
+        verify_ssl=verify_ssl,
+    )
+
+
+def detect_registry_type(registry_url):
+    """Return a string identifying the detected registry type."""
+    if _DOCKER_HUB_PATTERN.search(registry_url):
+        return "dockerhub"
+    if _ECR_PATTERN.search(registry_url):
+        return "ecr"
+    return "oci"

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

@@ -0,0 +1,187 @@
+"""Generic OCI Distribution Spec registry adapter."""
+
+from __future__ import annotations
+
+import re
+import time
+
+import requests
+
+from prowler.lib.logger import logger
+from prowler.providers.image.exceptions.exceptions import (
+    ImageRegistryAuthError,
+    ImageRegistryCatalogError,
+    ImageRegistryNetworkError,
+)
+from prowler.providers.image.lib.registry.base import RegistryAdapter
+
+_MAX_RETRIES = 3
+_BACKOFF_BASE = 1
+
+
+class OciRegistryAdapter(RegistryAdapter):
+    """Adapter for registries implementing OCI Distribution Spec."""
+
+    def __init__(
+        self, registry_url, username=None, password=None, token=None, verify_ssl=True
+    ):
+        super().__init__(registry_url, username, password, token, verify_ssl)
+        self._base_url = self._normalise_url(registry_url)
+        self._bearer_token = None
+
+    @staticmethod
+    def _normalise_url(url):
+        url = url.rstrip("/")
+        if not url.startswith(("http://", "https://")):
+            url = f"https://{url}"
+        return url
+
+    def list_repositories(self):
+        self._ensure_auth()
+        repositories = []
+        url = f"{self._base_url}/v2/_catalog"
+        params = {"n": 200}
+        while url:
+            resp = self._authed_request("GET", url, params=params)
+            if resp.status_code == 404:
+                raise ImageRegistryCatalogError(
+                    file=__file__,
+                    message=f"Registry at {self.registry_url} does not support catalog listing (/_catalog returned 404). Use --image or --image-list instead.",
+                )
+            self._check_response(resp, "catalog listing")
+            data = resp.json()
+            repositories.extend(data.get("repositories", []))
+            url = self._next_page_url(resp)
+            params = {}
+        return repositories
+
+    def list_tags(self, repository):
+        self._ensure_auth(repository=repository)
+        tags = []
+        url = f"{self._base_url}/v2/{repository}/tags/list"
+        params = {"n": 200}
+        while url:
+            resp = self._authed_request("GET", url, params=params)
+            self._check_response(resp, f"tag listing for {repository}")
+            data = resp.json()
+            tags.extend(data.get("tags", []) or [])
+            url = self._next_page_url(resp)
+            params = {}
+        return tags
+
+    def _ensure_auth(self, repository=None):
+        if self._bearer_token:
+            return
+        if self.token:
+            self._bearer_token = self.token
+            return
+        ping_url = f"{self._base_url}/v2/"
+        resp = self._request_with_retry("GET", ping_url)
+        if resp.status_code == 200:
+            return
+        if resp.status_code == 401:
+            www_auth = resp.headers.get("Www-Authenticate", "")
+            self._bearer_token = self._obtain_bearer_token(www_auth, repository)
+            return
+        if resp.status_code == 403:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Access denied to registry {self.registry_url} (HTTP 403). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
+            )
+
+    def _obtain_bearer_token(self, www_authenticate, repository=None):
+        match = re.search(r'realm="([^"]+)"', www_authenticate)
+        if not match:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Cannot parse token endpoint from registry {self.registry_url}. Www-Authenticate: {www_authenticate[:200]}",
+            )
+        realm = match.group(1)
+        params = {}
+        service_match = re.search(r'service="([^"]+)"', www_authenticate)
+        if service_match:
+            params["service"] = service_match.group(1)
+        scope_match = re.search(r'scope="([^"]+)"', www_authenticate)
+        if scope_match:
+            params["scope"] = scope_match.group(1)
+        elif repository:
+            params["scope"] = f"repository:{repository}:pull"
+        auth = None
+        if self.username and self.password:
+            auth = (self.username, self.password)
+        resp = self._request_with_retry("GET", realm, params=params, auth=auth)
+        if resp.status_code != 200:
+            raise ImageRegistryAuthError(
+                file=__file__,
+                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", "")
+
+    def _authed_request(self, method, url, **kwargs):
+        headers = kwargs.pop("headers", {})
+        if self._bearer_token:
+            headers["Authorization"] = f"Bearer {self._bearer_token}"
+        elif self.username and self.password:
+            kwargs.setdefault("auth", (self.username, self.password))
+        kwargs["headers"] = headers
+        return self._request_with_retry(method, url, **kwargs)
+
+    def _request_with_retry(self, method, url, **kwargs):
+        kwargs.setdefault("timeout", 30)
+        kwargs.setdefault("verify", self.verify_ssl)
+        last_exception = None
+        for attempt in range(1, _MAX_RETRIES + 1):
+            try:
+                resp = requests.request(method, url, **kwargs)
+                if resp.status_code == 429:
+                    wait = _BACKOFF_BASE * (2 ** (attempt - 1))
+                    logger.warning(
+                        f"Rate limited by registry, retrying in {wait}s (attempt {attempt}/{_MAX_RETRIES})"
+                    )
+                    time.sleep(wait)
+                    continue
+                return resp
+            except requests.exceptions.ConnectionError as exc:
+                last_exception = exc
+                if attempt < _MAX_RETRIES:
+                    wait = _BACKOFF_BASE * (2 ** (attempt - 1))
+                    logger.warning(
+                        f"Connection error to registry, retrying in {wait}s (attempt {attempt}/{_MAX_RETRIES})"
+                    )
+                    time.sleep(wait)
+                    continue
+            except requests.exceptions.Timeout as exc:
+                raise ImageRegistryNetworkError(
+                    file=__file__,
+                    message=f"Connection timed out to {self.registry_url}.",
+                    original_exception=exc,
+                )
+        raise ImageRegistryNetworkError(
+            file=__file__,
+            message=f"Failed to connect to {self.registry_url} after {_MAX_RETRIES} attempts.",
+            original_exception=last_exception,
+        )
+
+    def _check_response(self, resp, context):
+        if resp.status_code == 200:
+            return
+        if resp.status_code in (401, 403):
+            raise ImageRegistryAuthError(
+                file=__file__,
+                message=f"Authentication failed for {context} on {self.registry_url} (HTTP {resp.status_code}). Check REGISTRY_USERNAME and REGISTRY_PASSWORD.",
+            )
+        raise ImageRegistryNetworkError(
+            file=__file__,
+            message=f"Unexpected error during {context} on {self.registry_url} (HTTP {resp.status_code}): {resp.text[:200]}",
+        )
+
+    @staticmethod
+    def _next_page_url(resp):
+        link_header = resp.headers.get("Link", "")
+        if not link_header:
+            return None
+        match = re.search(r'<([^>]+)>;\s*rel="next"', link_header)
+        if match:
+            return match.group(1)
+        return None