Merge branch 'feat/PROWLER-939-stage-1-image-provider-mvp-for-cli' of https://github.com/prowler-cloud/prowler into feat/PROWLER-941-stage-2-b-image-provider-cli-docs

prowler/CHANGELOG.md

@@ -70,7 +70,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - CIS 5.0 compliance framework for the Azure provider [(#9777)](https://github.com/prowler-cloud/prowler/pull/9777)
 - `Cloudflare` Bot protection, WAF, Privacy, Anti-Scraping and Zone configuration checks [(#9425)](https://github.com/prowler-cloud/prowler/pull/9425)
 - `Cloudflare` `waf` and `dns record` checks [(#9426)](https://github.com/prowler-cloud/prowler/pull/9426)
-- Container Image provider using Trivy for vulnerability and secret scanning
+- `image` provider for container image scanning with Trivy integration [(#9984)](https://github.com/prowler-cloud/prowler/pull/9984)
 
 ### Changed
 

prowler/config/config.py

@@ -63,6 +63,7 @@ class Provider(str, Enum):
     ORACLECLOUD = "oraclecloud"
     ALIBABACLOUD = "alibabacloud"
     OPENSTACK = "openstack"
+    IMAGE = "image"
 
 
 # Compliance

prowler/lib/check/models.py

@@ -897,7 +897,6 @@ class CheckReportImage(Check_Report):
             finding = {}
         super().__init__(metadata, finding)
 
-        self.resource = finding
         self.resource_name = image_name
         self.image_digest = finding.get("PkgID", "")
         self.package_name = finding.get("PkgName", "")

prowler/lib/cli/parser.py

@@ -43,7 +43,7 @@ Available Cloud Providers:
     alibabacloud        Alibaba Cloud Provider
     iac                 IaC Provider (Beta)
     llm                 LLM Provider (Beta)
-    image               Container Image Provider (PoC)
+    image               Container Image Provider
     nhn                 NHN Provider (Unofficial)
     mongodbatlas        MongoDB Atlas Provider (Beta)
 

prowler/providers/image/exceptions/__init__.py

@@ -1,6 +1,9 @@
 from prowler.providers.image.exceptions.exceptions import (
     ImageBaseException,
     ImageFindingProcessingError,
+    ImageInvalidScannerError,
+    ImageInvalidSeverityError,
+    ImageInvalidTimeoutError,
     ImageListFileNotFoundError,
     ImageListFileReadError,
     ImageNoImagesProvidedError,
@@ -11,6 +14,9 @@ from prowler.providers.image.exceptions.exceptions import (
 __all__ = [
     "ImageBaseException",
     "ImageFindingProcessingError",
+    "ImageInvalidScannerError",
+    "ImageInvalidSeverityError",
+    "ImageInvalidTimeoutError",
     "ImageListFileNotFoundError",
     "ImageListFileReadError",
     "ImageNoImagesProvidedError",

prowler/providers/image/exceptions/exceptions.py

@@ -30,6 +30,22 @@ class ImageBaseException(ProwlerException):
             "message": "Error scanning container image.",
             "remediation": "Check the image name and ensure it is accessible.",
         },
+        (9006, "ImageInvalidTimeoutError"): {
+            "message": "Invalid timeout format.",
+            "remediation": "Use a valid timeout like '5m', '300s', or '1h'.",
+        },
+        (9007, "ImageInvalidScannerError"): {
+            "message": "Invalid scanner type.",
+            "remediation": "Use valid scanners: vuln, secret, misconfig, license.",
+        },
+        (9008, "ImageInvalidSeverityError"): {
+            "message": "Invalid severity level.",
+            "remediation": "Use valid severities: CRITICAL, HIGH, MEDIUM, LOW, UNKNOWN.",
+        },
+        (9009, "ImageInvalidNameError"): {
+            "message": "Invalid container image name.",
+            "remediation": "Use a valid image reference (e.g., 'alpine:3.18', 'registry.example.com/repo/image:tag').",
+        },
     }
 
     def __init__(self, code, file=None, original_exception=None, message=None):
@@ -97,3 +113,39 @@ class ImageScanError(ImageBaseException):
         super().__init__(
             9005, 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__(
+            9006, 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__(
+            9007, 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__(
+            9008, 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__(
+            9009, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/image/image_provider.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 import json
+import re
 import subprocess
 import sys
-from typing import Generator, List
+from typing import Generator
 
 from alive_progress import alive_bar
 from colorama import Fore, Style
@@ -19,12 +20,20 @@ from prowler.providers.common.models import Audit_Metadata, Connection
 from prowler.providers.common.provider import Provider
 from prowler.providers.image.exceptions.exceptions import (
     ImageFindingProcessingError,
+    ImageInvalidNameError,
+    ImageInvalidScannerError,
+    ImageInvalidSeverityError,
+    ImageInvalidTimeoutError,
     ImageListFileNotFoundError,
     ImageListFileReadError,
     ImageNoImagesProvidedError,
     ImageScanError,
     ImageTrivyBinaryNotFoundError,
 )
+from prowler.providers.image.lib.arguments.arguments import (
+    SCANNERS_CHOICES,
+    SEVERITY_CHOICES,
+)
 
 
 class ImageProvider(Provider):
@@ -36,18 +45,23 @@ class ImageProvider(Provider):
     """
 
     _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,
+        image_list_file: str | None = None,
         scanners: list[str] | None = None,
         trivy_severity: list[str] | None = None,
         ignore_unfixed: bool = False,
         timeout: str = "5m",
-        config_path: str = None,
-        config_content: dict = None,
+        config_path: str | None = None,
+        config_content: dict | None = None,
         fixer_config: dict | None = None,
     ):
         logger.info("Instantiating Image Provider...")
@@ -64,10 +78,15 @@ class ImageProvider(Provider):
         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__,
@@ -104,17 +123,32 @@ class ImageProvider(Provider):
     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 line and not line.startswith("#"):
-                        self.images.append(line)
+                    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,
@@ -122,6 +156,54 @@ class ImageProvider(Provider):
                 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 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
@@ -243,7 +325,7 @@ class ImageProvider(Provider):
             raise ImageFindingProcessingError(
                 file=__file__,
                 original_exception=error,
-                message=f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}",
+                message=f"Error processing finding: {error}",
             )
 
     def _build_status_extended(self, finding: dict) -> str:
@@ -271,19 +353,19 @@ class ImageProvider(Provider):
             " ".join(parts) if parts else finding.get("Description", "Finding detected")
         )
 
-    def run(self) -> List[CheckReportImage]:
+    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]:
+    def run_scan(self) -> Generator[list[CheckReportImage], None, None]:
         """
         Run Trivy scan on all configured images.
 
         Yields:
-            List[CheckReportImage]: Batches of findings
+            list[CheckReportImage]: Batches of findings
         """
         for image in self.images:
             try:
@@ -296,7 +378,7 @@ class ImageProvider(Provider):
 
     def _scan_single_image(
         self, image: str
-    ) -> Generator[List[CheckReportImage], None, None]:
+    ) -> Generator[list[CheckReportImage], None, None]:
         """
         Scan a single container image with Trivy.
 
@@ -304,7 +386,7 @@ class ImageProvider(Provider):
             image: The container image name/tag to scan
 
         Yields:
-            List[CheckReportImage]: Batches of findings
+            list[CheckReportImage]: Batches of findings
         """
         try:
             logger.info(f"Scanning container image: {image}")
@@ -339,9 +421,10 @@ class ImageProvider(Provider):
             # 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}: {error_msg}",
+                    message=f"Trivy scan failed for {image}: {categorized_msg}",
                 )
 
             # Parse JSON output
@@ -360,7 +443,6 @@ class ImageProvider(Provider):
 
             # Process findings in batches
             batch = []
-            batch_size = 100
 
             for result in results:
                 target = result.get("Target", image)
@@ -370,7 +452,7 @@ class ImageProvider(Provider):
                 for vuln in result.get("Vulnerabilities", []):
                     report = self._process_finding(vuln, target, result_type)
                     batch.append(report)
-                    if len(batch) >= batch_size:
+                    if len(batch) >= self.FINDING_BATCH_SIZE:
                         yield batch
                         batch = []
 
@@ -378,7 +460,7 @@ class ImageProvider(Provider):
                 for secret in result.get("Secrets", []):
                     report = self._process_finding(secret, target, "secret")
                     batch.append(report)
-                    if len(batch) >= batch_size:
+                    if len(batch) >= self.FINDING_BATCH_SIZE:
                         yield batch
                         batch = []
 
@@ -388,7 +470,7 @@ class ImageProvider(Provider):
                         misconfig, target, "misconfiguration"
                     )
                     batch.append(report)
-                    if len(batch) >= batch_size:
+                    if len(batch) >= self.FINDING_BATCH_SIZE:
                         yield batch
                         batch = []
 
@@ -480,6 +562,22 @@ class ImageProvider(Provider):
                 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}"
@@ -511,9 +609,9 @@ class ImageProvider(Provider):
 
     @staticmethod
     def test_connection(
-        image: str = None,
+        image: str | None = None,
         raise_on_exception: bool = True,
-        provider_id: str = None,
+        provider_id: str | None = None,
     ) -> "Connection":
         """
         Test connection to container registry by attempting to inspect an image.

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_provider_test.py

@@ -6,7 +6,12 @@ import pytest
 
 from prowler.lib.check.models import CheckReportImage
 from prowler.providers.image.exceptions.exceptions import (
+    ImageInvalidNameError,
+    ImageInvalidScannerError,
+    ImageInvalidSeverityError,
+    ImageInvalidTimeoutError,
     ImageListFileNotFoundError,
+    ImageListFileReadError,
     ImageNoImagesProvidedError,
     ImageScanError,
     ImageTrivyBinaryNotFoundError,
@@ -395,3 +400,147 @@ class TestImageProvider:
         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",
+        ]
+
+
+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)