fix(image): harden input validation and fix traceback edge case

- Add timeout, scanner, and severity input validation with custom exceptions - Fix potential None dereference on error.__traceback__ - Modernize typing imports (List → list) - Remove PoC label from CLI help text - Update CHANGELOG entry with PR reference
2026-07-24 13:01:56 +00:00 · 2026-02-09 14:00:42 +01:00
parent cc1197d632
commit 74ad1186cf
6 changed files with 203 additions and 19 deletions
@@ -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

@@ -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)

@@ -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",
@@ -30,6 +30,18 @@ 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.",
+        },
    }

    def __init__(self, code, file=None, original_exception=None, message=None):
@@ -97,3 +109,30 @@ 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
+        )
@@ -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,19 @@ from prowler.providers.common.models import Audit_Metadata, Connection
 from prowler.providers.common.provider import Provider
 from prowler.providers.image.exceptions.exceptions import (
    ImageFindingProcessingError,
+    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 +44,19 @@ class ImageProvider(Provider):
    """

    _type: str = "image"
+    FINDING_BATCH_SIZE: int = 100
    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,6 +73,8 @@ 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)
@@ -122,6 +133,28 @@ 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)}.",
+                )
+
    @property
    def auth_method(self) -> str:
        return self._auth_method
@@ -243,7 +276,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 +304,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 +329,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 +337,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 +372,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 +394,6 @@ class ImageProvider(Provider):

            # Process findings in batches
            batch = []
-            batch_size = 100

            for result in results:
                target = result.get("Target", image)
@@ -370,7 +403,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 +411,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 +421,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 +513,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 +560,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.
@@ -6,6 +6,9 @@ import pytest

 from prowler.lib.check.models import CheckReportImage
 from prowler.providers.image.exceptions.exceptions import (
+    ImageInvalidScannerError,
+    ImageInvalidSeverityError,
+    ImageInvalidTimeoutError,
    ImageListFileNotFoundError,
    ImageNoImagesProvidedError,
    ImageScanError,
@@ -395,3 +398,90 @@ 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