feat(image): add container image provider for CLI scanning (#9984)

prowler/CHANGELOG.md

@@ -14,6 +14,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - CSA CCM 4.0 for the GCP provider [(#10042)](https://github.com/prowler-cloud/prowler/pull/10042)
 - CSA CCM for the Azure provider [(#10039)](https://github.com/prowler-cloud/prowler/pull/10039)
 - 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)
 
 ### 🔄 Changed
 

prowler/__main__.py

@@ -8,6 +8,7 @@ from colorama import Fore, Style
 from colorama import init as colorama_init
 
 from prowler.config.config import (
+    EXTERNAL_TOOL_PROVIDERS,
     csv_file_suffix,
     get_available_compliance_frameworks,
     html_file_suffix,
@@ -122,6 +123,8 @@ from prowler.providers.common.quick_inventory import run_provider_quick_inventor
 from prowler.providers.gcp.models import GCPOutputOptions
 from prowler.providers.github.models import GithubOutputOptions
 from prowler.providers.iac.models import IACOutputOptions
+from prowler.providers.image.exceptions.exceptions import ImageBaseException
+from prowler.providers.image.models import ImageOutputOptions
 from prowler.providers.kubernetes.models import KubernetesOutputOptions
 from prowler.providers.llm.models import LLMOutputOptions
 from prowler.providers.m365.models import M365OutputOptions
@@ -209,8 +212,8 @@ def prowler():
     # Load compliance frameworks
     logger.debug("Loading compliance frameworks from .json files")
 
-    # Skip compliance frameworks for IAC and LLM providers
-    if provider != "iac" and provider != "llm":
+    # Skip compliance frameworks for external-tool providers
+    if provider not in EXTERNAL_TOOL_PROVIDERS:
         bulk_compliance_frameworks = Compliance.get_bulk(provider)
         # Complete checks metadata with the compliance framework specification
         bulk_checks_metadata = update_checks_metadata_with_compliance(
@@ -267,8 +270,8 @@ def prowler():
     if not args.only_logs:
         global_provider.print_credentials()
 
-    # Skip service and check loading for IAC and LLM providers
-    if provider != "iac" and provider != "llm":
+    # Skip service and check loading for external-tool providers
+    if provider not in EXTERNAL_TOOL_PROVIDERS:
         # Import custom checks from folder
         if checks_folder:
             custom_checks = parse_checks_from_folder(global_provider, checks_folder)
@@ -355,6 +358,8 @@ def prowler():
         )
     elif provider == "iac":
         output_options = IACOutputOptions(args, bulk_checks_metadata)
+    elif provider == "image":
+        output_options = ImageOutputOptions(args, bulk_checks_metadata)
     elif provider == "llm":
         output_options = LLMOutputOptions(args, bulk_checks_metadata)
     elif provider == "oraclecloud":
@@ -378,8 +383,8 @@ def prowler():
     # Execute checks
     findings = []
 
-    if provider == "iac" or provider == "llm":
-        # For IAC and LLM providers, run the scan directly
+    if provider in EXTERNAL_TOOL_PROVIDERS:
+        # For external-tool providers, run the scan directly
         if provider == "llm":
 
             def streaming_callback(findings_batch):
@@ -389,7 +394,11 @@ def prowler():
             findings = global_provider.run_scan(streaming_callback=streaming_callback)
         else:
             # Original behavior for IAC or non-verbose LLM
-            findings = global_provider.run()
+            try:
+                findings = global_provider.run()
+            except ImageBaseException as error:
+                logger.critical(f"{error}")
+                sys.exit(1)
             # Note: IaC doesn't support granular progress tracking since Trivy runs as a black box
             # and returns all findings at once. Progress tracking would just be 0% → 100%.
 

prowler/config/config.py

@@ -63,8 +63,13 @@ class Provider(str, Enum):
     ORACLECLOUD = "oraclecloud"
     ALIBABACLOUD = "alibabacloud"
     OPENSTACK = "openstack"
+    IMAGE = "image"
 
 
+# Providers that delegate scanning to an external tool (e.g. Trivy, promptfoo)
+# and bypass standard check/service loading.
+EXTERNAL_TOOL_PROVIDERS = frozenset({"iac", "llm", "image"})
+
 # Compliance
 actual_directory = pathlib.Path(os.path.dirname(os.path.realpath(__file__)))
 
@@ -75,7 +80,10 @@ def get_available_compliance_frameworks(provider=None):
     if provider:
         providers = [provider]
     for provider in providers:
-        with os.scandir(f"{actual_directory}/../compliance/{provider}") as files:
+        compliance_dir = f"{actual_directory}/../compliance/{provider}"
+        if not os.path.isdir(compliance_dir):
+            continue
+        with os.scandir(compliance_dir) as files:
             for file in files:
                 if file.is_file() and file.name.endswith(".json"):
                     available_compliance_frameworks.append(

prowler/lib/check/models.py

@@ -10,7 +10,7 @@ from typing import Any, Dict, Optional, Set
 from pydantic.v1 import BaseModel, Field, ValidationError, validator
 from pydantic.v1.error_wrappers import ErrorWrapper
 
-from prowler.config.config import Provider
+from prowler.config.config import EXTERNAL_TOOL_PROVIDERS, Provider
 from prowler.lib.check.compliance_models import Compliance
 from prowler.lib.check.utils import recover_checks_from_provider
 from prowler.lib.logger import logger
@@ -159,11 +159,7 @@ class CheckMetadata(BaseModel):
             raise ValueError("ServiceName must be a non-empty string")
 
         check_id = values.get("CheckID")
-        if (
-            check_id
-            and values.get("Provider") != "iac"
-            and values.get("Provider") != "llm"
-        ):
+        if check_id and values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
             service_from_check_id = check_id.split("_")[0]
             if service_name != service_from_check_id:
                 raise ValueError(
@@ -179,11 +175,7 @@ class CheckMetadata(BaseModel):
         if not check_id:
             raise ValueError("CheckID must be a non-empty string")
 
-        if (
-            check_id
-            and values.get("Provider") != "iac"
-            and values.get("Provider") != "llm"
-        ):
+        if check_id and values.get("Provider") not in EXTERNAL_TOOL_PROVIDERS:
             if "-" in check_id:
                 raise ValueError(
                     f"CheckID {check_id} contains a hyphen, which is not allowed"
@@ -865,6 +857,49 @@ class CheckReportIAC(Check_Report):
         )
 
 
+@dataclass
+class CheckReportImage(Check_Report):
+    """Contains the Container Image Check's finding information using Trivy."""
+
+    resource_name: str
+    resource_id: str
+    image_digest: str
+    package_name: str
+    installed_version: str
+    fixed_version: str
+
+    def __init__(
+        self,
+        metadata: Optional[dict] = None,
+        finding: Optional[dict] = None,
+        image_name: str = "",
+    ) -> None:
+        """
+        Initialize the Container Image Check's finding information from a Trivy vulnerability/secret dict.
+
+        Args:
+            metadata (Dict): Check metadata.
+            finding (dict): A single vulnerability/secret result from Trivy's JSON output.
+            image_name (str): The container image name being scanned.
+        """
+        if metadata is None:
+            metadata = {}
+        if finding is None:
+            finding = {}
+        super().__init__(metadata, finding)
+
+        self.resource_name = image_name
+        self.resource_id = (
+            finding.get("VulnerabilityID", "")
+            or finding.get("RuleID", "")
+            or finding.get("ID", "")
+        )
+        self.image_digest = finding.get("PkgID", "")
+        self.package_name = finding.get("PkgName", "")
+        self.installed_version = finding.get("InstalledVersion", "")
+        self.fixed_version = finding.get("FixedVersion", "")
+
+
 @dataclass
 class CheckReportLLM(Check_Report):
     """Contains the LLM Check's finding information."""

prowler/lib/check/utils.py

@@ -14,8 +14,8 @@ def recover_checks_from_provider(
     Returns a list of tuples with the following format (check_name, check_path)
     """
     try:
-        # Bypass check loading for IAC provider since it uses Trivy directly
-        if provider == "iac" or provider == "llm":
+        # Bypass check loading for IAC, LLM, and Image providers since they use external tools directly
+        if provider in ("iac", "llm", "image"):
             return []
 
         checks = []

prowler/lib/cli/parser.py

@@ -27,10 +27,10 @@ class ProwlerArgumentParser:
         self.parser = argparse.ArgumentParser(
             prog="prowler",
             formatter_class=RawTextHelpFormatter,
-            usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,dashboard,iac} ...",
+            usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,dashboard,iac,image} ...",
             epilog="""
 Available Cloud Providers:
-  {aws,azure,gcp,kubernetes,m365,github,iac,llm,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack}
+  {aws,azure,gcp,kubernetes,m365,github,iac,llm,image,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack}
     aws                 AWS Provider
     azure               Azure Provider
     gcp                 GCP Provider
@@ -43,6 +43,7 @@ Available Cloud Providers:
     alibabacloud        Alibaba Cloud Provider
     iac                 IaC Provider (Beta)
     llm                 LLM Provider (Beta)
+    image               Container Image Provider
     nhn                 NHN Provider (Unofficial)
     mongodbatlas        MongoDB Atlas Provider (Beta)
 

prowler/lib/outputs/finding.py

@@ -380,6 +380,23 @@ class Finding(BaseModel):
                 output_data["resource_uid"] = check_output.resource_id
                 output_data["region"] = check_output.region
 
+            elif provider.type == "image":
+                output_data["auth_method"] = provider.auth_method
+                output_data["account_uid"] = "image"
+                output_data["account_name"] = "image"
+                output_data["resource_name"] = getattr(
+                    check_output, "resource_name", ""
+                )
+                output_data["resource_uid"] = getattr(check_output, "resource_id", "")
+                output_data["region"] = getattr(check_output, "region", "container")
+                output_data["package_name"] = getattr(check_output, "package_name", "")
+                output_data["installed_version"] = getattr(
+                    check_output, "installed_version", ""
+                )
+                output_data["fixed_version"] = getattr(
+                    check_output, "fixed_version", ""
+                )
+
             # check_output Unique ID
             # TODO: move this to a function
             # TODO: in Azure, GCP and K8s there are findings without resource_name

prowler/lib/outputs/summary_table.py

@@ -93,6 +93,9 @@ def display_summary_table(
                 if provider.identity.project_name
                 else provider.identity.project_id
             )
+        elif provider.type == "image":
+            entity_type = "Image"
+            audited_entities = ", ".join(provider.images)
 
         # Check if there are findings and that they are not all MANUAL
         if findings and not all(finding.status == "MANUAL" for finding in findings):

prowler/providers/common/provider.py

@@ -274,6 +274,18 @@ class Provider(ABC):
                         config_path=arguments.config_file,
                         fixer_config=fixer_config,
                     )
+                elif "image" in provider_class_name.lower():
+                    provider_class(
+                        images=arguments.images,
+                        image_list_file=arguments.image_list_file,
+                        scanners=arguments.scanners,
+                        image_config_scanners=arguments.image_config_scanners,
+                        trivy_severity=arguments.trivy_severity,
+                        ignore_unfixed=arguments.ignore_unfixed,
+                        timeout=arguments.timeout,
+                        config_path=arguments.config_file,
+                        fixer_config=fixer_config,
+                    )
                 elif "mongodbatlas" in provider_class_name.lower():
                     provider_class(
                         atlas_public_key=arguments.atlas_public_key,

prowler/providers/image/exceptions/exceptions.py

@@ -0,0 +1,164 @@
+from prowler.exceptions.exceptions import ProwlerException
+
+
+# Exceptions codes from 11000 to 11999 are reserved for Image exceptions
+class ImageBaseException(ProwlerException):
+    """Base class for Image provider errors."""
+
+    IMAGE_ERROR_CODES = {
+        (11000, "ImageNoImagesProvidedError"): {
+            "message": "No container images provided for scanning.",
+            "remediation": "Provide at least one image using --image or --image-list-file.",
+        },
+        (11001, "ImageListFileNotFoundError"): {
+            "message": "Image list file not found.",
+            "remediation": "Ensure the image list file exists at the specified path.",
+        },
+        (11002, "ImageListFileReadError"): {
+            "message": "Error reading image list file.",
+            "remediation": "Check file permissions and format. The file should contain one image per line.",
+        },
+        (11003, "ImageFindingProcessingError"): {
+            "message": "Error processing image scan finding.",
+            "remediation": "Check the Trivy output format and ensure the finding structure is valid.",
+        },
+        (11004, "ImageTrivyBinaryNotFoundError"): {
+            "message": "Trivy binary not found.",
+            "remediation": "Install Trivy from https://trivy.dev/latest/getting-started/installation/",
+        },
+        (11005, "ImageScanError"): {
+            "message": "Error scanning container image.",
+            "remediation": "Check the image name and ensure it is accessible.",
+        },
+        (11006, "ImageInvalidTimeoutError"): {
+            "message": "Invalid timeout format.",
+            "remediation": "Use a valid timeout like '5m', '300s', or '1h'.",
+        },
+        (11007, "ImageInvalidScannerError"): {
+            "message": "Invalid scanner type.",
+            "remediation": "Use valid scanners: vuln, secret, misconfig, license.",
+        },
+        (11008, "ImageInvalidSeverityError"): {
+            "message": "Invalid severity level.",
+            "remediation": "Use valid severities: CRITICAL, HIGH, MEDIUM, LOW, UNKNOWN.",
+        },
+        (11009, "ImageInvalidNameError"): {
+            "message": "Invalid container image name.",
+            "remediation": "Use a valid image reference (e.g., 'alpine:3.18', 'registry.example.com/repo/image:tag').",
+        },
+        (11010, "ImageInvalidConfigScannerError"): {
+            "message": "Invalid image config scanner type.",
+            "remediation": "Use valid image config scanners: misconfig, secret.",
+        },
+    }
+
+    def __init__(self, code, file=None, original_exception=None, message=None):
+        error_info = self.IMAGE_ERROR_CODES.get((code, self.__class__.__name__))
+        if message:
+            error_info["message"] = message
+        super().__init__(
+            code,
+            source="Image",
+            file=file,
+            original_exception=original_exception,
+            error_info=error_info,
+        )
+
+
+class ImageNoImagesProvidedError(ImageBaseException):
+    """Exception raised when no container images are provided for scanning."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11000, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageListFileNotFoundError(ImageBaseException):
+    """Exception raised when the image list file is not found."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11001, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageListFileReadError(ImageBaseException):
+    """Exception raised when the image list file cannot be read."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11002, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageFindingProcessingError(ImageBaseException):
+    """Exception raised when a finding cannot be processed."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11003, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageTrivyBinaryNotFoundError(ImageBaseException):
+    """Exception raised when the Trivy binary is not found."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11004, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageScanError(ImageBaseException):
+    """Exception raised when a general scan error occurs."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11005, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageInvalidTimeoutError(ImageBaseException):
+    """Exception raised when an invalid timeout format is provided."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11006, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageInvalidScannerError(ImageBaseException):
+    """Exception raised when an invalid scanner type is provided."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11007, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageInvalidSeverityError(ImageBaseException):
+    """Exception raised when an invalid severity level is provided."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11008, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageInvalidNameError(ImageBaseException):
+    """Exception raised when an invalid container image name is provided."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11009, file=file, original_exception=original_exception, message=message
+        )
+
+
+class ImageInvalidConfigScannerError(ImageBaseException):
+    """Exception raised when an invalid image config scanner type is provided."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            11010, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/image/image_provider.py

@@ -0,0 +1,707 @@
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import sys
+from typing import Generator
+
+from alive_progress import alive_bar
+from colorama import Fore, Style
+
+from prowler.config.config import (
+    default_config_file_path,
+    load_and_validate_config_file,
+)
+from prowler.lib.check.models import CheckReportImage
+from prowler.lib.logger import logger
+from prowler.lib.utils.utils import print_boxes
+from prowler.providers.common.models import Audit_Metadata, Connection
+from prowler.providers.common.provider import Provider
+from prowler.providers.image.exceptions.exceptions import (
+    ImageFindingProcessingError,
+    ImageInvalidConfigScannerError,
+    ImageInvalidNameError,
+    ImageInvalidScannerError,
+    ImageInvalidSeverityError,
+    ImageInvalidTimeoutError,
+    ImageListFileNotFoundError,
+    ImageListFileReadError,
+    ImageNoImagesProvidedError,
+    ImageScanError,
+    ImageTrivyBinaryNotFoundError,
+)
+from prowler.providers.image.lib.arguments.arguments import (
+    IMAGE_CONFIG_SCANNERS_CHOICES,
+    SCANNERS_CHOICES,
+    SEVERITY_CHOICES,
+)
+
+
+class ImageProvider(Provider):
+    """
+    Container Image Provider using Trivy for vulnerability and secret scanning.
+
+    This is a Tool/Wrapper provider that delegates all scanning logic to Trivy's
+    `trivy image` command and converts the output to Prowler's finding format.
+    """
+
+    _type: str = "image"
+    FINDING_BATCH_SIZE: int = 100
+    MAX_IMAGE_LIST_LINES: int = 10_000
+    MAX_IMAGE_NAME_LENGTH: int = 500
+    _IMAGE_NAME_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9.\-_/:@]+$")
+    _SHELL_METACHARACTERS = frozenset(";|&$`\n\r")
+    audit_metadata: Audit_Metadata
+
+    def __init__(
+        self,
+        images: list[str] | None = None,
+        image_list_file: str | None = None,
+        scanners: list[str] | None = None,
+        image_config_scanners: list[str] | None = None,
+        trivy_severity: list[str] | None = None,
+        ignore_unfixed: bool = False,
+        timeout: str = "5m",
+        config_path: str | None = None,
+        config_content: dict | None = None,
+        fixer_config: dict | None = None,
+    ):
+        logger.info("Instantiating Image Provider...")
+
+        self.images = images if images is not None else []
+        self.image_list_file = image_list_file
+        self.scanners = scanners if scanners is not None else ["vuln", "secret"]
+        self.image_config_scanners = (
+            image_config_scanners if image_config_scanners is not None else []
+        )
+        self.trivy_severity = trivy_severity if trivy_severity is not None else []
+        self.ignore_unfixed = ignore_unfixed
+        self.timeout = timeout
+        self.region = "container"
+        self.audited_account = "image-scan"
+        self._session = None
+        self._identity = "prowler"
+        self._auth_method = "No auth"
+
+        self._validate_inputs()
+
+        # Load images from file if provided
+        if image_list_file:
+            self._load_images_from_file(image_list_file)
+
+        for image in self.images:
+            self._validate_image_name(image)
+
+        if not self.images:
+            raise ImageNoImagesProvidedError(
+                file=__file__,
+                message="No images provided for scanning.",
+            )
+
+        # Audit Config
+        if config_content:
+            self._audit_config = config_content
+        else:
+            if not config_path:
+                config_path = default_config_file_path
+            self._audit_config = load_and_validate_config_file(self._type, config_path)
+
+        # Fixer Config
+        self._fixer_config = fixer_config if fixer_config is not None else {}
+
+        # Mutelist (not needed for Image provider since Trivy has its own logic)
+        self._mutelist = None
+
+        self.audit_metadata = Audit_Metadata(
+            provider=self._type,
+            account_id=self.audited_account,
+            account_name="image",
+            region=self.region,
+            services_scanned=0,
+            expected_checks=[],
+            completed_checks=0,
+            audit_progress=0,
+        )
+
+        Provider.set_global_provider(self)
+
+    def _load_images_from_file(self, file_path: str) -> None:
+        """Load image names from a file (one per line)."""
+        try:
+            line_count = 0
+            with open(file_path, "r") as f:
+                for line in f:
+                    line_count += 1
+                    if line_count > self.MAX_IMAGE_LIST_LINES:
+                        raise ImageListFileReadError(
+                            file=file_path,
+                            message=f"Image list file exceeds maximum of {self.MAX_IMAGE_LIST_LINES} lines.",
+                        )
+                    line = line.strip()
+                    if not line or line.startswith("#"):
+                        continue
+                    if len(line) > self.MAX_IMAGE_NAME_LENGTH:
+                        logger.warning(
+                            f"Skipping image name exceeding {self.MAX_IMAGE_NAME_LENGTH} chars at line {line_count} in {file_path}"
+                        )
+                        continue
+                    self.images.append(line)
+            logger.info(f"Loaded {len(self.images)} images from {file_path}")
+        except FileNotFoundError:
+            raise ImageListFileNotFoundError(
+                file=file_path,
+                message=f"Image list file not found: {file_path}",
+            )
+        except (ImageListFileReadError, ImageListFileNotFoundError):
+            raise
+        except Exception as error:
+            raise ImageListFileReadError(
+                file=file_path,
+                original_exception=error,
+                message=f"Error reading image list file: {error}",
+            )
+
+    def _validate_inputs(self) -> None:
+        """Validate timeout, scanners, and severity inputs."""
+        if not re.fullmatch(r"\d+[smh]", self.timeout):
+            raise ImageInvalidTimeoutError(
+                file=__file__,
+                message=f"Invalid timeout format: '{self.timeout}'. Expected pattern like '5m', '300s', or '1h'.",
+            )
+
+        for scanner in self.scanners:
+            if scanner not in SCANNERS_CHOICES:
+                raise ImageInvalidScannerError(
+                    file=__file__,
+                    message=f"Invalid scanner: '{scanner}'. Valid options: {', '.join(SCANNERS_CHOICES)}.",
+                )
+
+        for config_scanner in self.image_config_scanners:
+            if config_scanner not in IMAGE_CONFIG_SCANNERS_CHOICES:
+                raise ImageInvalidConfigScannerError(
+                    file=__file__,
+                    message=f"Invalid image config scanner: '{config_scanner}'. Valid options: {', '.join(IMAGE_CONFIG_SCANNERS_CHOICES)}.",
+                )
+
+        for severity in self.trivy_severity:
+            if severity not in SEVERITY_CHOICES:
+                raise ImageInvalidSeverityError(
+                    file=__file__,
+                    message=f"Invalid severity: '{severity}'. Valid options: {', '.join(SEVERITY_CHOICES)}.",
+                )
+
+    def _validate_image_name(self, name: str) -> None:
+        """Validate a container image name for safety and correctness."""
+        if not name:
+            raise ImageInvalidNameError(
+                file=__file__,
+                message="Image name must not be empty.",
+            )
+
+        if len(name) > self.MAX_IMAGE_NAME_LENGTH:
+            raise ImageInvalidNameError(
+                file=__file__,
+                message=f"Image name exceeds maximum length of {self.MAX_IMAGE_NAME_LENGTH} characters: '{name[:50]}...'",
+            )
+
+        if any(c in self._SHELL_METACHARACTERS for c in name):
+            raise ImageInvalidNameError(
+                file=__file__,
+                message=f"Image name contains invalid characters: '{name}'",
+            )
+
+        if not self._IMAGE_NAME_PATTERN.fullmatch(name):
+            raise ImageInvalidNameError(
+                file=__file__,
+                message=f"Image name does not match valid OCI reference format: '{name}'",
+            )
+
+    @property
+    def auth_method(self) -> str:
+        return self._auth_method
+
+    @property
+    def type(self) -> str:
+        return self._type
+
+    @property
+    def identity(self) -> str:
+        return self._identity
+
+    @property
+    def session(self) -> None:
+        return self._session
+
+    @property
+    def audit_config(self) -> dict:
+        return self._audit_config
+
+    @property
+    def fixer_config(self) -> dict:
+        return self._fixer_config
+
+    def setup_session(self) -> None:
+        """Image provider doesn't need a session since it uses Trivy directly"""
+        return None
+
+    def _process_finding(
+        self, finding: dict, image_name: str, finding_type: str
+    ) -> CheckReportImage:
+        """
+        Process a single finding and create a CheckReportImage object.
+
+        Args:
+            finding: The finding object from Trivy output
+            image_name: The container image name being scanned
+            finding_type: The type of finding (Vulnerability, Secret, etc.)
+
+        Returns:
+            CheckReportImage: The processed check report
+        """
+        try:
+            # Determine finding ID based on type
+            if "VulnerabilityID" in finding:
+                finding_id = finding["VulnerabilityID"]
+                finding_description = finding.get(
+                    "Description", finding.get("Title", "")
+                )
+                finding_status = "FAIL"
+            elif "RuleID" in finding:
+                # Secret finding
+                finding_id = finding["RuleID"]
+                finding_description = finding.get("Title", "Secret detected")
+                finding_status = "FAIL"
+            else:
+                finding_id = finding.get("ID", "UNKNOWN")
+                finding_description = finding.get("Description", "")
+                finding_status = finding.get("Status", "FAIL")
+
+            # Build remediation text for vulnerabilities
+            remediation_text = ""
+            if finding.get("FixedVersion"):
+                remediation_text = f"Upgrade {finding.get('PkgName', 'package')} to version {finding['FixedVersion']}"
+            elif finding.get("Resolution"):
+                remediation_text = finding["Resolution"]
+
+            # Convert Trivy severity to Prowler severity (lowercase, map UNKNOWN to informational)
+            trivy_severity = finding.get("Severity", "UNKNOWN").lower()
+            if trivy_severity == "unknown":
+                trivy_severity = "informational"
+
+            metadata_dict = {
+                "Provider": "image",
+                "CheckID": finding_id,
+                "CheckTitle": finding.get("Title", finding_id),
+                "CheckType": ["Container Image Security"],
+                "ServiceName": finding_type,
+                "SubServiceName": "",
+                "ResourceIdTemplate": "",
+                "Severity": trivy_severity,
+                "ResourceType": "container-image",
+                "ResourceGroup": "container",
+                "Description": finding_description,
+                "Risk": finding.get(
+                    "Description", "Vulnerability detected in container image"
+                ),
+                "RelatedUrl": finding.get("PrimaryURL", ""),
+                "Remediation": {
+                    "Code": {
+                        "NativeIaC": "",
+                        "Terraform": "",
+                        "CLI": "",
+                        "Other": "",
+                    },
+                    "Recommendation": {
+                        "Text": remediation_text,
+                        "Url": finding.get("PrimaryURL", ""),
+                    },
+                },
+                "Categories": [],
+                "DependsOn": [],
+                "RelatedTo": [],
+                "Notes": "",
+            }
+
+            # Convert metadata dict to JSON string
+            metadata = json.dumps(metadata_dict)
+
+            report = CheckReportImage(
+                metadata=metadata, finding=finding, image_name=image_name
+            )
+            report.status = finding_status
+            report.status_extended = self._build_status_extended(finding)
+            report.region = self.region
+            return report
+
+        except Exception as error:
+            raise ImageFindingProcessingError(
+                file=__file__,
+                original_exception=error,
+                message=f"Error processing finding: {error}",
+            )
+
+    def _build_status_extended(self, finding: dict) -> str:
+        """Build a detailed status message for the finding."""
+        parts = []
+
+        if finding.get("VulnerabilityID"):
+            parts.append(f"{finding['VulnerabilityID']}")
+
+        if finding.get("PkgName"):
+            pkg_info = finding["PkgName"]
+            if finding.get("InstalledVersion"):
+                pkg_info += f"@{finding['InstalledVersion']}"
+            parts.append(f"in package {pkg_info}")
+
+        if finding.get("FixedVersion"):
+            parts.append(f"(fix available: {finding['FixedVersion']})")
+        elif finding.get("Status") == "will_not_fix":
+            parts.append("(no fix available)")
+
+        if finding.get("Title"):
+            parts.append(f"- {finding['Title']}")
+
+        return (
+            " ".join(parts) if parts else finding.get("Description", "Finding detected")
+        )
+
+    def run(self) -> list[CheckReportImage]:
+        """Execute the container image scan."""
+        reports = []
+        for batch in self.run_scan():
+            reports.extend(batch)
+        return reports
+
+    def run_scan(self) -> Generator[list[CheckReportImage], None, None]:
+        """
+        Run Trivy scan on all configured images.
+
+        Yields:
+            list[CheckReportImage]: Batches of findings
+        """
+        for image in self.images:
+            try:
+                yield from self._scan_single_image(image)
+            except (ImageScanError, ImageTrivyBinaryNotFoundError):
+                raise
+            except Exception as error:
+                logger.error(f"Error scanning image {image}: {error}")
+                continue
+
+    def _scan_single_image(
+        self, image: str
+    ) -> Generator[list[CheckReportImage], None, None]:
+        """
+        Scan a single container image with Trivy.
+
+        Args:
+            image: The container image name/tag to scan
+
+        Yields:
+            list[CheckReportImage]: Batches of findings
+        """
+        try:
+            logger.info(f"Scanning container image: {image}")
+
+            # Build Trivy command
+            trivy_command = [
+                "trivy",
+                "image",
+                "--format",
+                "json",
+                "--scanners",
+                ",".join(self.scanners),
+                "--timeout",
+                self.timeout,
+            ]
+
+            if self.image_config_scanners:
+                trivy_command.extend(
+                    ["--image-config-scanners", ",".join(self.image_config_scanners)]
+                )
+
+            if self.trivy_severity:
+                trivy_command.extend(["--severity", ",".join(self.trivy_severity)])
+
+            if self.ignore_unfixed:
+                trivy_command.append("--ignore-unfixed")
+
+            trivy_command.append(image)
+
+            # Execute Trivy
+            process = self._execute_trivy(trivy_command, image)
+
+            # Log stderr output
+            if process.stderr:
+                self._log_trivy_stderr(process.stderr)
+
+            # Check for Trivy failure
+            if process.returncode != 0:
+                error_msg = self._extract_trivy_errors(process.stderr)
+                categorized_msg = self._categorize_trivy_error(error_msg)
+                raise ImageScanError(
+                    file=__file__,
+                    message=f"Trivy scan failed for {image}: {categorized_msg}",
+                )
+
+            # Parse JSON output
+            try:
+                output = json.loads(process.stdout)
+                results = output.get("Results", [])
+
+                if not results:
+                    logger.info(f"No findings for image: {image}")
+                    return
+
+            except json.JSONDecodeError as error:
+                logger.error(f"Failed to parse Trivy output for {image}: {error}")
+                logger.debug(f"Trivy stdout: {process.stdout[:500]}")
+                return
+
+            # Process findings in batches
+            batch = []
+
+            for result in results:
+                target = result.get("Target", image)
+                result_type = result.get("Type", "unknown")
+
+                # Process Vulnerabilities
+                for vuln in result.get("Vulnerabilities", []):
+                    report = self._process_finding(vuln, target, result_type)
+                    batch.append(report)
+                    if len(batch) >= self.FINDING_BATCH_SIZE:
+                        yield batch
+                        batch = []
+
+                # Process Secrets
+                for secret in result.get("Secrets", []):
+                    report = self._process_finding(secret, target, "secret")
+                    batch.append(report)
+                    if len(batch) >= self.FINDING_BATCH_SIZE:
+                        yield batch
+                        batch = []
+
+                # Process Misconfigurations (from Dockerfile)
+                for misconfig in result.get("Misconfigurations", []):
+                    report = self._process_finding(
+                        misconfig, target, "misconfiguration"
+                    )
+                    batch.append(report)
+                    if len(batch) >= self.FINDING_BATCH_SIZE:
+                        yield batch
+                        batch = []
+
+            # Yield remaining findings
+            if batch:
+                yield batch
+
+        except (ImageScanError, ImageTrivyBinaryNotFoundError):
+            raise
+        except Exception as error:
+            if "No such file or directory: 'trivy'" in str(error):
+                raise ImageTrivyBinaryNotFoundError(
+                    file=__file__,
+                    original_exception=error,
+                    message="Trivy binary not found. Please install Trivy from https://trivy.dev/latest/getting-started/installation/",
+                )
+            logger.error(f"Error scanning image {image}: {error}")
+
+    def _execute_trivy(self, command: list, image: str) -> subprocess.CompletedProcess:
+        """Execute Trivy command with optional progress bar."""
+        try:
+            if sys.stdout.isatty():
+                with alive_bar(
+                    ctrl_c=False,
+                    bar="blocks",
+                    spinner="classic",
+                    stats=False,
+                    enrich_print=False,
+                ) as bar:
+                    bar.title = f"-> Scanning {image}..."
+                    process = subprocess.run(
+                        command,
+                        capture_output=True,
+                        text=True,
+                    )
+                    bar.title = f"-> Scan completed for {image}"
+                    return process
+            else:
+                logger.info(f"Scanning {image}...")
+                process = subprocess.run(
+                    command,
+                    capture_output=True,
+                    text=True,
+                )
+                logger.info(f"Scan completed for {image}")
+                return process
+        except (AttributeError, OSError):
+            logger.info(f"Scanning {image}...")
+            return subprocess.run(command, capture_output=True, text=True)
+
+    def _log_trivy_stderr(self, stderr: str) -> None:
+        """Parse and log Trivy's stderr output."""
+        for line in stderr.strip().split("\n"):
+            if line.strip():
+                parts = line.split()
+                if len(parts) >= 3:
+                    level = parts[1]
+                    message = " ".join(parts[2:])
+                    if level == "ERROR":
+                        logger.error(message)
+                    elif level == "WARN":
+                        logger.warning(message)
+                    elif level == "INFO":
+                        logger.info(message)
+                    elif level == "DEBUG":
+                        logger.debug(message)
+                    else:
+                        logger.info(message)
+                else:
+                    logger.info(line)
+
+    @staticmethod
+    def _extract_trivy_errors(stderr: str) -> str:
+        """Extract only ERROR-level messages from Trivy stderr output."""
+        if not stderr:
+            return "Unknown error"
+        error_lines = []
+        for line in stderr.strip().split("\n"):
+            parts = line.split()
+            if len(parts) >= 3 and parts[1] == "ERROR":
+                error_lines.append(" ".join(parts[2:]))
+            elif len(parts) >= 3 and parts[1] == "FATAL":
+                error_lines.append(" ".join(parts[2:]))
+        if error_lines:
+            return "; ".join(error_lines)[:500]
+        # Fallback: no ERROR lines found, return last non-empty line
+        for line in reversed(stderr.strip().split("\n")):
+            if line.strip():
+                return line.strip()[:500]
+        return "Unknown error"
+
+    @staticmethod
+    def _categorize_trivy_error(error_msg: str) -> str:
+        """Categorize a Trivy error message to provide actionable guidance."""
+        lower = error_msg.lower()
+
+        if any(kw in lower for kw in ("401", "403", "unauthorized", "denied")):
+            return f"Auth failure — check `docker login`: {error_msg}"
+        if any(kw in lower for kw in ("404", "manifest unknown", "not found")):
+            return f"Image not found — check name/tag/registry: {error_msg}"
+        if any(kw in lower for kw in ("429", "rate limit", "too many requests")):
+            return f"Rate limited — wait or authenticate: {error_msg}"
+        if any(kw in lower for kw in ("timeout", "connection refused", "no such host")):
+            return f"Network issue — check connectivity: {error_msg}"
+
+        return error_msg
+
+    def print_credentials(self) -> None:
+        """Print scan configuration."""
+        report_title = f"{Style.BRIGHT}Scanning container images:{Style.RESET_ALL}"
+
+        report_lines = []
+        if len(self.images) <= 3:
+            for img in self.images:
+                report_lines.append(f"Image: {Fore.YELLOW}{img}{Style.RESET_ALL}")
+        else:
+            report_lines.append(
+                f"Images: {Fore.YELLOW}{len(self.images)} images{Style.RESET_ALL}"
+            )
+
+        report_lines.append(
+            f"Scanners: {Fore.YELLOW}{', '.join(self.scanners)}{Style.RESET_ALL}"
+        )
+
+        if self.image_config_scanners:
+            report_lines.append(
+                f"Image config scanners: {Fore.YELLOW}{', '.join(self.image_config_scanners)}{Style.RESET_ALL}"
+            )
+
+        if self.trivy_severity:
+            report_lines.append(
+                f"Severity filter: {Fore.YELLOW}{', '.join(self.trivy_severity)}{Style.RESET_ALL}"
+            )
+
+        if self.ignore_unfixed:
+            report_lines.append(f"Ignore unfixed: {Fore.YELLOW}Yes{Style.RESET_ALL}")
+
+        report_lines.append(f"Timeout: {Fore.YELLOW}{self.timeout}{Style.RESET_ALL}")
+
+        print_boxes(report_lines, report_title)
+
+    @staticmethod
+    def test_connection(
+        image: str | None = None,
+        raise_on_exception: bool = True,
+        provider_id: str | None = None,
+    ) -> "Connection":
+        """
+        Test connection to container registry by attempting to inspect an image.
+
+        Args:
+            image: Container image to test
+            raise_on_exception: Whether to raise exceptions
+            provider_id: Fallback for image name
+
+        Returns:
+            Connection: Connection object with success status
+        """
+        try:
+            if provider_id and not image:
+                image = provider_id
+
+            if not image:
+                return Connection(is_connected=False, error="Image name is required")
+
+            # Test by running trivy with --skip-update to just test image access
+            process = subprocess.run(
+                [
+                    "trivy",
+                    "image",
+                    "--skip-db-update",
+                    "--download-db-only=false",
+                    image,
+                ],
+                capture_output=True,
+                text=True,
+                timeout=60,
+            )
+
+            if process.returncode == 0:
+                return Connection(is_connected=True)
+            else:
+                error_msg = process.stderr or "Unknown error"
+                if "401" in error_msg or "unauthorized" in error_msg.lower():
+                    return Connection(
+                        is_connected=False,
+                        error="Authentication failed. Check registry credentials.",
+                    )
+                elif "not found" in error_msg.lower() or "404" in error_msg:
+                    return Connection(
+                        is_connected=False,
+                        error="Image not found in registry.",
+                    )
+                else:
+                    return Connection(
+                        is_connected=False,
+                        error=f"Failed to access image: {error_msg[:200]}",
+                    )
+
+        except subprocess.TimeoutExpired:
+            return Connection(
+                is_connected=False,
+                error="Connection timed out",
+            )
+        except FileNotFoundError:
+            return Connection(
+                is_connected=False,
+                error="Trivy binary not found. Please install Trivy.",
+            )
+        except Exception as error:
+            if raise_on_exception:
+                raise
+            return Connection(
+                is_connected=False,
+                error=f"Unexpected error: {str(error)}",
+            )

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

@@ -0,0 +1,103 @@
+SCANNERS_CHOICES = [
+    "vuln",
+    "secret",
+    "misconfig",
+    "license",
+]
+
+IMAGE_CONFIG_SCANNERS_CHOICES = [
+    "misconfig",
+    "secret",
+]
+
+SEVERITY_CHOICES = [
+    "CRITICAL",
+    "HIGH",
+    "MEDIUM",
+    "LOW",
+    "UNKNOWN",
+]
+
+
+def init_parser(self):
+    """Init the Image Provider CLI parser"""
+    image_parser = self.subparsers.add_parser(
+        "image", parents=[self.common_providers_parser], help="Container Image Provider"
+    )
+
+    # Image Selection
+    image_selection_group = image_parser.add_argument_group("Image Selection")
+    image_selection_group.add_argument(
+        "--image",
+        "-I",
+        dest="images",
+        action="append",
+        default=[],
+        help="Container image to scan. Can be specified multiple times. Examples: nginx:latest, alpine:3.18, myregistry.io/myapp:v1.0",
+    )
+
+    image_selection_group.add_argument(
+        "--image-list",
+        dest="image_list_file",
+        default=None,
+        help="Path to a file containing list of images to scan (one per line). Lines starting with # are treated as comments.",
+    )
+
+    # Scan Configuration
+    scan_config_group = image_parser.add_argument_group("Scan Configuration")
+    scan_config_group.add_argument(
+        "--scanners",
+        "--scanner",
+        dest="scanners",
+        nargs="+",
+        default=["vuln", "secret"],
+        choices=SCANNERS_CHOICES,
+        help="Trivy scanners to use. Default: vuln, secret. Available: vuln, secret, misconfig, license",
+    )
+
+    scan_config_group.add_argument(
+        "--image-config-scanners",
+        dest="image_config_scanners",
+        nargs="+",
+        default=[],
+        choices=IMAGE_CONFIG_SCANNERS_CHOICES,
+        help="Trivy image config scanners (scans Dockerfile-level metadata). Available: misconfig, secret",
+    )
+
+    scan_config_group.add_argument(
+        "--trivy-severity",
+        dest="trivy_severity",
+        nargs="+",
+        default=[],
+        choices=SEVERITY_CHOICES,
+        help="Filter Trivy findings by severity. Default: all severities. Available: CRITICAL, HIGH, MEDIUM, LOW, UNKNOWN",
+    )
+
+    scan_config_group.add_argument(
+        "--ignore-unfixed",
+        dest="ignore_unfixed",
+        action="store_true",
+        default=False,
+        help="Ignore vulnerabilities without available fixes.",
+    )
+
+    scan_config_group.add_argument(
+        "--timeout",
+        dest="timeout",
+        default="5m",
+        help="Trivy scan timeout. Default: 5m. Examples: 10m, 1h",
+    )
+
+
+def validate_arguments(arguments):
+    """Validate Image provider arguments."""
+    images = getattr(arguments, "images", [])
+    image_list_file = getattr(arguments, "image_list_file", None)
+
+    if not images and not image_list_file:
+        return (
+            False,
+            "At least one image must be specified using --image (-I) or --image-list.",
+        )
+
+    return (True, "")

prowler/providers/image/models.py

@@ -0,0 +1,21 @@
+from prowler.config.config import output_file_timestamp
+from prowler.providers.common.models import ProviderOutputOptions
+
+
+class ImageOutputOptions(ProviderOutputOptions):
+    """
+    ImageOutputOptions customizes output filename logic for container image scanning.
+
+    Attributes inherited from ProviderOutputOptions:
+        - output_filename (str): The base filename used for generated reports.
+        - output_directory (str): The directory to store the output files.
+    """
+
+    def __init__(self, arguments, bulk_checks_metadata):
+        super().__init__(arguments, bulk_checks_metadata)
+
+        # If --output-filename is not specified, build a default name
+        if not getattr(arguments, "output_filename", None):
+            self.output_filename = f"prowler-output-image-{output_file_timestamp}"
+        else:
+            self.output_filename = arguments.output_filename

tests/lib/cli/parser_test.py

@@ -17,7 +17,7 @@ prowler_command = "prowler"
 
 # capsys
 # https://docs.pytest.org/en/7.1.x/how-to/capture-stdout-stderr.html
-prowler_default_usage_error = "usage: prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,dashboard,iac} ..."
+prowler_default_usage_error = "usage: prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,nhn,mongodbatlas,oraclecloud,alibabacloud,cloudflare,openstack,dashboard,iac,image} ..."
 
 
 def mock_get_available_providers():
@@ -29,6 +29,7 @@ def mock_get_available_providers():
         "m365",
         "github",
         "iac",
+        "image",
         "nhn",
         "mongodbatlas",
         "oraclecloud",

tests/providers/image/image_fixtures.py

@@ -0,0 +1,92 @@
+import json
+
+# Sample vulnerability finding from Trivy
+SAMPLE_VULNERABILITY_FINDING = {
+    "VulnerabilityID": "CVE-2024-1234",
+    "PkgID": "openssl@1.1.1k-r0",
+    "PkgName": "openssl",
+    "InstalledVersion": "1.1.1k-r0",
+    "FixedVersion": "1.1.1l-r0",
+    "Severity": "HIGH",
+    "Title": "OpenSSL Buffer Overflow",
+    "Description": "A buffer overflow vulnerability in OpenSSL allows remote attackers to execute arbitrary code.",
+    "PrimaryURL": "https://avd.aquasec.com/nvd/cve-2024-1234",
+}
+
+# Sample secret finding from Trivy
+SAMPLE_SECRET_FINDING = {
+    "RuleID": "aws-access-key-id",
+    "Category": "AWS",
+    "Severity": "CRITICAL",
+    "Title": "AWS Access Key ID",
+    "StartLine": 10,
+    "EndLine": 10,
+    "Match": "AKIA...",
+}
+
+# Sample misconfiguration finding from Trivy
+SAMPLE_MISCONFIGURATION_FINDING = {
+    "ID": "DS001",
+    "Title": "Dockerfile should not use latest tag",
+    "Description": "Using latest tag can cause unpredictable builds.",
+    "Severity": "MEDIUM",
+    "Resolution": "Use a specific version tag instead of latest",
+    "PrimaryURL": "https://avd.aquasec.com/misconfig/ds001",
+}
+
+# Sample finding with UNKNOWN severity
+SAMPLE_UNKNOWN_SEVERITY_FINDING = {
+    "VulnerabilityID": "CVE-2024-9999",
+    "PkgID": "test-pkg@0.0.1",
+    "PkgName": "test-pkg",
+    "InstalledVersion": "0.0.1",
+    "Severity": "UNKNOWN",
+    "Title": "Unknown severity issue",
+    "Description": "An issue with unknown severity.",
+}
+
+# Full Trivy JSON output structure with a single vulnerability
+SAMPLE_TRIVY_IMAGE_OUTPUT = {
+    "Results": [
+        {
+            "Target": "alpine:3.18 (alpine 3.18.0)",
+            "Type": "alpine",
+            "Vulnerabilities": [SAMPLE_VULNERABILITY_FINDING],
+            "Secrets": [],
+            "Misconfigurations": [],
+        }
+    ]
+}
+
+# Full Trivy JSON output with mixed finding types
+SAMPLE_TRIVY_MULTI_TYPE_OUTPUT = {
+    "Results": [
+        {
+            "Target": "myimage:latest (debian 12)",
+            "Type": "debian",
+            "Vulnerabilities": [SAMPLE_VULNERABILITY_FINDING],
+            "Secrets": [SAMPLE_SECRET_FINDING],
+            "Misconfigurations": [SAMPLE_MISCONFIGURATION_FINDING],
+        }
+    ]
+}
+
+
+def get_sample_trivy_json_output():
+    """Return sample Trivy JSON output as string."""
+    return json.dumps(SAMPLE_TRIVY_IMAGE_OUTPUT)
+
+
+def get_empty_trivy_output():
+    """Return empty Trivy output as string."""
+    return json.dumps({"Results": []})
+
+
+def get_invalid_trivy_output():
+    """Return invalid JSON output as string."""
+    return "invalid json output"
+
+
+def get_multi_type_trivy_output():
+    """Return Trivy output with multiple finding types as string."""
+    return json.dumps(SAMPLE_TRIVY_MULTI_TYPE_OUTPUT)

tests/providers/image/image_provider_test.py

@@ -0,0 +1,595 @@
+import tempfile
+from unittest import mock
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from prowler.lib.check.models import CheckReportImage
+from prowler.providers.image.exceptions.exceptions import (
+    ImageInvalidConfigScannerError,
+    ImageInvalidNameError,
+    ImageInvalidScannerError,
+    ImageInvalidSeverityError,
+    ImageInvalidTimeoutError,
+    ImageListFileNotFoundError,
+    ImageListFileReadError,
+    ImageNoImagesProvidedError,
+    ImageScanError,
+    ImageTrivyBinaryNotFoundError,
+)
+from prowler.providers.image.image_provider import ImageProvider
+from tests.providers.image.image_fixtures import (
+    SAMPLE_MISCONFIGURATION_FINDING,
+    SAMPLE_SECRET_FINDING,
+    SAMPLE_UNKNOWN_SEVERITY_FINDING,
+    SAMPLE_VULNERABILITY_FINDING,
+    get_empty_trivy_output,
+    get_invalid_trivy_output,
+    get_multi_type_trivy_output,
+    get_sample_trivy_json_output,
+)
+
+
+def _make_provider(**kwargs):
+    """Helper to create an ImageProvider with test defaults."""
+    defaults = {
+        "images": ["alpine:3.18"],
+        "config_content": {},
+    }
+    defaults.update(kwargs)
+    return ImageProvider(**defaults)
+
+
+class TestImageProvider:
+    def test_image_provider(self):
+        """Test default initialization."""
+        provider = _make_provider()
+
+        assert provider._type == "image"
+        assert provider.type == "image"
+        assert provider.images == ["alpine:3.18"]
+        assert provider.scanners == ["vuln", "secret"]
+        assert provider.image_config_scanners == []
+        assert provider.trivy_severity == []
+        assert provider.ignore_unfixed is False
+        assert provider.timeout == "5m"
+        assert provider.region == "container"
+        assert provider.audited_account == "image-scan"
+        assert provider.identity == "prowler"
+        assert provider.auth_method == "No auth"
+        assert provider.session is None
+        assert provider.audit_config == {}
+        assert provider.fixer_config == {}
+        assert provider._mutelist is None
+
+    def test_image_provider_custom_params(self):
+        """Test initialization with custom parameters."""
+        provider = _make_provider(
+            images=["nginx:1.25", "redis:7"],
+            scanners=["vuln", "secret", "misconfig"],
+            trivy_severity=["HIGH", "CRITICAL"],
+            ignore_unfixed=True,
+            timeout="10m",
+            fixer_config={"key": "value"},
+        )
+
+        assert provider.images == ["nginx:1.25", "redis:7"]
+        assert provider.scanners == ["vuln", "secret", "misconfig"]
+        assert provider.trivy_severity == ["HIGH", "CRITICAL"]
+        assert provider.ignore_unfixed is True
+        assert provider.timeout == "10m"
+        assert provider.fixer_config == {"key": "value"}
+
+    def test_image_provider_with_image_list_file(self):
+        """Test loading images from a file, skipping comments and blank lines."""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
+            f.write("# Comment line\n")
+            f.write("alpine:3.18\n")
+            f.write("\n")
+            f.write("  nginx:latest  \n")
+            f.write("# Another comment\n")
+            f.write("redis:7\n")
+            f.name
+
+        provider = _make_provider(
+            images=None,
+            image_list_file=f.name,
+        )
+
+        assert "alpine:3.18" in provider.images
+        assert "nginx:latest" in provider.images
+        assert "redis:7" in provider.images
+        assert len(provider.images) == 3
+
+    def test_image_provider_no_images(self):
+        """Test that ImageNoImagesProvidedError is raised when no images are given."""
+        with pytest.raises(ImageNoImagesProvidedError):
+            _make_provider(images=[])
+
+    def test_image_provider_image_list_file_not_found(self):
+        """Test that ImageListFileNotFoundError is raised for missing file."""
+        with pytest.raises(ImageListFileNotFoundError):
+            _make_provider(
+                images=None,
+                image_list_file="/nonexistent/path/images.txt",
+            )
+
+    def test_process_finding_vulnerability(self):
+        """Test processing a vulnerability finding."""
+        provider = _make_provider()
+        report = provider._process_finding(
+            SAMPLE_VULNERABILITY_FINDING,
+            "alpine:3.18 (alpine 3.18.0)",
+            "alpine",
+        )
+
+        assert isinstance(report, CheckReportImage)
+        assert report.status == "FAIL"
+        assert report.check_metadata.CheckID == "CVE-2024-1234"
+        assert report.check_metadata.Severity == "high"
+        assert report.check_metadata.ServiceName == "alpine"
+        assert report.check_metadata.ResourceType == "container-image"
+        assert report.check_metadata.ResourceGroup == "container"
+        assert report.package_name == "openssl"
+        assert report.installed_version == "1.1.1k-r0"
+        assert report.fixed_version == "1.1.1l-r0"
+        assert report.resource_name == "alpine:3.18 (alpine 3.18.0)"
+        assert report.region == "container"
+
+    def test_process_finding_secret(self):
+        """Test processing a secret finding (identified by RuleID)."""
+        provider = _make_provider()
+        report = provider._process_finding(
+            SAMPLE_SECRET_FINDING,
+            "myimage:latest",
+            "secret",
+        )
+
+        assert isinstance(report, CheckReportImage)
+        assert report.status == "FAIL"
+        assert report.check_metadata.CheckID == "aws-access-key-id"
+        assert report.check_metadata.Severity == "critical"
+        assert report.check_metadata.ServiceName == "secret"
+
+    def test_process_finding_misconfiguration(self):
+        """Test processing a misconfiguration finding (identified by ID)."""
+        provider = _make_provider()
+        report = provider._process_finding(
+            SAMPLE_MISCONFIGURATION_FINDING,
+            "myimage:latest",
+            "misconfiguration",
+        )
+
+        assert isinstance(report, CheckReportImage)
+        assert report.check_metadata.CheckID == "DS001"
+        assert report.check_metadata.Severity == "medium"
+        assert report.check_metadata.ServiceName == "misconfiguration"
+
+    def test_process_finding_unknown_severity(self):
+        """Test that UNKNOWN severity is mapped to informational."""
+        provider = _make_provider()
+        report = provider._process_finding(
+            SAMPLE_UNKNOWN_SEVERITY_FINDING,
+            "myimage:latest",
+            "alpine",
+        )
+
+        assert report.check_metadata.Severity == "informational"
+
+    @patch("subprocess.run")
+    def test_run_scan_success(self, mock_subprocess):
+        """Test successful scan with mocked subprocess."""
+        provider = _make_provider()
+        mock_subprocess.return_value = MagicMock(
+            returncode=0, stdout=get_sample_trivy_json_output(), stderr=""
+        )
+
+        reports = []
+        for batch in provider.run_scan():
+            reports.extend(batch)
+
+        assert len(reports) == 1
+        assert reports[0].check_metadata.CheckID == "CVE-2024-1234"
+
+    @patch("subprocess.run")
+    def test_run_scan_empty_output(self, mock_subprocess):
+        """Test scan with empty Trivy output produces no findings."""
+        provider = _make_provider()
+        mock_subprocess.return_value = MagicMock(
+            returncode=0, stdout=get_empty_trivy_output(), stderr=""
+        )
+
+        reports = []
+        for batch in provider.run_scan():
+            reports.extend(batch)
+
+        assert len(reports) == 0
+
+    @patch("subprocess.run")
+    def test_run_scan_invalid_json(self, mock_subprocess):
+        """Test scan with malformed output doesn't crash."""
+        provider = _make_provider()
+        mock_subprocess.return_value = MagicMock(
+            returncode=0, stdout=get_invalid_trivy_output(), stderr=""
+        )
+
+        reports = []
+        for batch in provider.run_scan():
+            reports.extend(batch)
+
+        assert len(reports) == 0
+
+    @patch("subprocess.run")
+    def test_run_scan_trivy_not_found(self, mock_subprocess):
+        """Test that ImageTrivyBinaryNotFoundError is raised when trivy is missing."""
+        provider = _make_provider()
+        mock_subprocess.side_effect = FileNotFoundError(
+            "[Errno 2] No such file or directory: 'trivy'"
+        )
+
+        with pytest.raises(ImageTrivyBinaryNotFoundError):
+            for _ in provider._scan_single_image("alpine:3.18"):
+                pass
+
+    @patch("subprocess.run")
+    def test_run_scan_multiple_images(self, mock_subprocess):
+        """Test scanning multiple images makes separate subprocess calls."""
+        provider = _make_provider(images=["alpine:3.18", "nginx:latest"])
+        mock_subprocess.return_value = MagicMock(
+            returncode=0, stdout=get_sample_trivy_json_output(), stderr=""
+        )
+
+        reports = []
+        for batch in provider.run_scan():
+            reports.extend(batch)
+
+        assert mock_subprocess.call_count == 2
+
+    @patch("subprocess.run")
+    def test_run_scan_multi_type_output(self, mock_subprocess):
+        """Test scan with vulnerabilities, secrets, and misconfigurations."""
+        provider = _make_provider()
+        mock_subprocess.return_value = MagicMock(
+            returncode=0, stdout=get_multi_type_trivy_output(), stderr=""
+        )
+
+        reports = []
+        for batch in provider.run_scan():
+            reports.extend(batch)
+
+        assert len(reports) == 3
+
+        check_ids = [r.check_metadata.CheckID for r in reports]
+        assert "CVE-2024-1234" in check_ids
+        assert "aws-access-key-id" in check_ids
+        assert "DS001" in check_ids
+
+    def test_print_credentials(self):
+        """Test that print_credentials outputs image names."""
+        provider = _make_provider()
+        with mock.patch("builtins.print") as mock_print:
+            provider.print_credentials()
+            output = " ".join(
+                str(call.args[0]) for call in mock_print.call_args_list if call.args
+            )
+            assert "alpine:3.18" in output
+
+    @patch("subprocess.run")
+    def test_test_connection_success(self, mock_subprocess):
+        """Test successful connection returns is_connected=True."""
+        mock_subprocess.return_value = MagicMock(returncode=0, stderr="")
+
+        result = ImageProvider.test_connection(image="alpine:3.18")
+
+        assert result.is_connected is True
+
+    @patch("subprocess.run")
+    def test_test_connection_auth_failure(self, mock_subprocess):
+        """Test 401 error returns auth failure."""
+        mock_subprocess.return_value = MagicMock(
+            returncode=1, stderr="401 unauthorized"
+        )
+
+        result = ImageProvider.test_connection(image="private/image:latest")
+
+        assert result.is_connected is False
+        assert "Authentication failed" in result.error
+
+    @patch("subprocess.run")
+    def test_test_connection_not_found(self, mock_subprocess):
+        """Test 404 error returns not found."""
+        mock_subprocess.return_value = MagicMock(returncode=1, stderr="404 not found")
+
+        result = ImageProvider.test_connection(image="nonexistent/image:latest")
+
+        assert result.is_connected is False
+        assert "not found" in result.error
+
+    def test_build_status_extended(self):
+        """Test status message content for different finding types."""
+        provider = _make_provider()
+
+        # Vulnerability with fix
+        status = provider._build_status_extended(SAMPLE_VULNERABILITY_FINDING)
+        assert "CVE-2024-1234" in status
+        assert "openssl" in status
+        assert "fix available" in status
+
+        # Finding with no special fields
+        status = provider._build_status_extended({"Description": "Simple finding"})
+        assert status == "Simple finding"
+
+        # Finding with will_not_fix status
+        finding_no_fix = {
+            "VulnerabilityID": "CVE-2024-0000",
+            "PkgName": "libc",
+            "Status": "will_not_fix",
+            "Title": "Some vuln",
+        }
+        status = provider._build_status_extended(finding_no_fix)
+        assert "no fix available" in status
+
+    def test_validate_arguments(self):
+        """Test valid and invalid argument combinations."""
+        # Valid: images provided
+        provider = _make_provider(images=["alpine:3.18"])
+        assert provider.images == ["alpine:3.18"]
+
+        # Invalid: empty images and no file
+        with pytest.raises(ImageNoImagesProvidedError):
+            _make_provider(images=[])
+
+        # Valid: custom scanners
+        provider = _make_provider(scanners=["vuln"])
+        assert provider.scanners == ["vuln"]
+
+    def test_setup_session(self):
+        """Test that setup_session returns None."""
+        provider = _make_provider()
+        assert provider.setup_session() is None
+
+    @patch("subprocess.run")
+    def test_run_method(self, mock_subprocess):
+        """Test that run() collects all batches into a list."""
+        provider = _make_provider()
+        mock_subprocess.return_value = MagicMock(
+            returncode=0, stdout=get_sample_trivy_json_output(), stderr=""
+        )
+
+        reports = provider.run()
+
+        assert isinstance(reports, list)
+        assert len(reports) == 1
+
+    @patch("subprocess.run")
+    def test_scan_single_image_trivy_nonzero_exit(self, mock_subprocess):
+        """Test that a non-zero Trivy exit code raises ImageScanError."""
+        provider = _make_provider()
+        mock_subprocess.return_value = MagicMock(
+            returncode=1,
+            stdout="",
+            stderr="fatal error: unable to pull image",
+        )
+
+        with pytest.raises(ImageScanError):
+            for _ in provider._scan_single_image("alpine:3.18"):
+                pass
+
+    @patch("subprocess.run")
+    def test_scan_single_image_auth_failure(self, mock_subprocess):
+        """Test that a 401 unauthorized stderr raises ImageScanError with message."""
+        provider = _make_provider()
+        mock_subprocess.return_value = MagicMock(
+            returncode=1,
+            stdout="",
+            stderr="ERROR 401 unauthorized: authentication required",
+        )
+
+        with pytest.raises(ImageScanError, match="401 unauthorized"):
+            for _ in provider._scan_single_image("private/image:latest"):
+                pass
+
+    @patch("subprocess.run")
+    def test_run_scan_propagates_scan_error(self, mock_subprocess):
+        """Test that run_scan() re-raises ImageScanError instead of swallowing it."""
+        provider = _make_provider()
+        mock_subprocess.return_value = MagicMock(
+            returncode=1,
+            stdout="",
+            stderr="image not found",
+        )
+
+        with pytest.raises(ImageScanError):
+            for _ in provider.run_scan():
+                pass
+
+
+class TestImageProviderInputValidation:
+    def test_invalid_timeout_format_raises_error(self):
+        """Test that a non-matching timeout string raises ImageInvalidTimeoutError."""
+        with pytest.raises(ImageInvalidTimeoutError):
+            _make_provider(timeout="invalid")
+
+    def test_invalid_timeout_no_unit_raises_error(self):
+        """Test that a numeric timeout without a unit raises ImageInvalidTimeoutError."""
+        with pytest.raises(ImageInvalidTimeoutError):
+            _make_provider(timeout="300")
+
+    def test_invalid_timeout_wrong_unit_raises_error(self):
+        """Test that a timeout with an unsupported unit raises ImageInvalidTimeoutError."""
+        with pytest.raises(ImageInvalidTimeoutError):
+            _make_provider(timeout="5d")
+
+    def test_valid_timeout_seconds(self):
+        """Test that a seconds-based timeout is accepted."""
+        provider = _make_provider(timeout="300s")
+        assert provider.timeout == "300s"
+
+    def test_valid_timeout_hours(self):
+        """Test that an hours-based timeout is accepted."""
+        provider = _make_provider(timeout="1h")
+        assert provider.timeout == "1h"
+
+    def test_invalid_scanner_raises_error(self):
+        """Test that an invalid scanner name raises ImageInvalidScannerError."""
+        with pytest.raises(ImageInvalidScannerError):
+            _make_provider(scanners=["vuln", "bad"])
+
+    def test_invalid_severity_raises_error(self):
+        """Test that an invalid severity level raises ImageInvalidSeverityError."""
+        with pytest.raises(ImageInvalidSeverityError):
+            _make_provider(trivy_severity=["HIGH", "SUPER_HIGH"])
+
+    def test_valid_all_scanners(self):
+        """Test that all valid scanner choices are accepted."""
+        provider = _make_provider(scanners=["vuln", "secret", "misconfig", "license"])
+        assert provider.scanners == ["vuln", "secret", "misconfig", "license"]
+
+    def test_valid_all_severities(self):
+        """Test that all valid severity choices are accepted."""
+        provider = _make_provider(
+            trivy_severity=["CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN"]
+        )
+        assert provider.trivy_severity == [
+            "CRITICAL",
+            "HIGH",
+            "MEDIUM",
+            "LOW",
+            "UNKNOWN",
+        ]
+
+    def test_image_config_scanners_defaults_to_empty(self):
+        """Test that image_config_scanners defaults to an empty list."""
+        provider = _make_provider()
+        assert provider.image_config_scanners == []
+
+    def test_valid_image_config_scanners(self):
+        """Test that valid image config scanners are accepted."""
+        provider = _make_provider(image_config_scanners=["misconfig", "secret"])
+        assert provider.image_config_scanners == ["misconfig", "secret"]
+
+    def test_invalid_image_config_scanner_raises_error(self):
+        """Test that an invalid image config scanner raises ImageInvalidConfigScannerError."""
+        with pytest.raises(ImageInvalidConfigScannerError):
+            _make_provider(image_config_scanners=["misconfig", "vuln"])
+
+    @patch("subprocess.run")
+    def test_trivy_command_includes_image_config_scanners(self, mock_subprocess):
+        """Test that Trivy command includes --image-config-scanners when set."""
+        provider = _make_provider(image_config_scanners=["misconfig", "secret"])
+        mock_subprocess.return_value = MagicMock(
+            returncode=0, stdout=get_empty_trivy_output(), stderr=""
+        )
+
+        for _ in provider._scan_single_image("alpine:3.18"):
+            pass
+
+        call_args = mock_subprocess.call_args[0][0]
+        assert "--image-config-scanners" in call_args
+        idx = call_args.index("--image-config-scanners")
+        assert call_args[idx + 1] == "misconfig,secret"
+
+    @patch("subprocess.run")
+    def test_trivy_command_omits_image_config_scanners_when_empty(
+        self, mock_subprocess
+    ):
+        """Test that Trivy command omits --image-config-scanners when empty."""
+        provider = _make_provider(image_config_scanners=[])
+        mock_subprocess.return_value = MagicMock(
+            returncode=0, stdout=get_empty_trivy_output(), stderr=""
+        )
+
+        for _ in provider._scan_single_image("alpine:3.18"):
+            pass
+
+        call_args = mock_subprocess.call_args[0][0]
+        assert "--image-config-scanners" not in call_args
+
+
+class TestImageProviderErrorCategorization:
+    def test_categorize_auth_failure(self):
+        """Test that auth-related errors are categorized correctly."""
+        result = ImageProvider._categorize_trivy_error(
+            "401 unauthorized: access denied"
+        )
+        assert "Auth failure" in result
+
+    def test_categorize_not_found(self):
+        """Test that not-found errors are categorized correctly."""
+        result = ImageProvider._categorize_trivy_error(
+            "manifest unknown: image not found"
+        )
+        assert "Image not found" in result
+
+    def test_categorize_rate_limit(self):
+        """Test that rate-limit errors are categorized correctly."""
+        result = ImageProvider._categorize_trivy_error("429 too many requests")
+        assert "Rate limited" in result
+
+    def test_categorize_network_issue(self):
+        """Test that network errors are categorized correctly."""
+        result = ImageProvider._categorize_trivy_error("connection refused to registry")
+        assert "Network issue" in result
+
+    def test_categorize_unknown_error(self):
+        """Test that unrecognized errors are returned as-is."""
+        msg = "some unknown trivy error"
+        result = ImageProvider._categorize_trivy_error(msg)
+        assert result == msg
+
+
+class TestImageProviderNameValidation:
+    @pytest.mark.parametrize(
+        "bad_name",
+        [
+            "alpine;rm -rf /",
+            "image|cat /etc/passwd",
+            "image&background",
+            "image$VAR",
+            "image`whoami`",
+            "image\ninjected",
+            "image\rinjected",
+        ],
+    )
+    def test_image_provider_invalid_image_name_shell_chars(self, bad_name):
+        """Test that image names with shell metacharacters raise ImageInvalidNameError."""
+        with pytest.raises(ImageInvalidNameError):
+            _make_provider(images=[bad_name])
+
+    def test_image_provider_invalid_image_name_empty(self):
+        """Test that an empty string image name raises ImageInvalidNameError."""
+        with pytest.raises(ImageInvalidNameError):
+            _make_provider(images=[""])
+
+    @pytest.mark.parametrize(
+        "valid_name",
+        [
+            "alpine:3.18",
+            "nginx:latest",
+            "registry.example.com/repo/image:tag",
+            "ghcr.io/owner/image:v1.2.3",
+            "myimage@sha256:abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890",
+            "localhost:5000/myimage:latest",
+        ],
+    )
+    def test_image_provider_valid_image_names(self, valid_name):
+        """Test that various valid image name formats pass validation."""
+        provider = _make_provider(images=[valid_name])
+        assert valid_name in provider.images
+
+    def test_image_provider_image_name_too_long(self):
+        """Test that a name exceeding 500 chars raises ImageInvalidNameError."""
+        long_name = "a" * 501
+        with pytest.raises(ImageInvalidNameError):
+            _make_provider(images=[long_name])
+
+    def test_image_provider_file_too_many_lines(self):
+        """Test that a file with more than MAX_IMAGE_LIST_LINES raises ImageListFileReadError."""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
+            for i in range(10_001):
+                f.write(f"image{i}:latest\n")
+            f.flush()
+            file_path = f.name
+
+        with pytest.raises(ImageListFileReadError):
+            _make_provider(images=None, image_list_file=file_path)