From 74ad1186cf0cc66cfd3d4e68c9c0cfa296721f6b Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Mon, 9 Feb 2026 14:00:42 +0100 Subject: [PATCH] fix(image): harden input validation and fix traceback edge case MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- prowler/CHANGELOG.md | 2 +- prowler/lib/cli/parser.py | 2 +- .../providers/image/exceptions/__init__.py | 6 ++ .../providers/image/exceptions/exceptions.py | 39 ++++++++ prowler/providers/image/image_provider.py | 83 +++++++++++++---- tests/providers/image/image_provider_test.py | 90 +++++++++++++++++++ 6 files changed, 203 insertions(+), 19 deletions(-) diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 581d956d1f..e158d53458 100644 --- a/prowler/CHANGELOG.md +++ b/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 diff --git a/prowler/lib/cli/parser.py b/prowler/lib/cli/parser.py index b90789046e..c3045dbd76 100644 --- a/prowler/lib/cli/parser.py +++ b/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) diff --git a/prowler/providers/image/exceptions/__init__.py b/prowler/providers/image/exceptions/__init__.py index 95b0b3a110..6daaa639b5 100644 --- a/prowler/providers/image/exceptions/__init__.py +++ b/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", diff --git a/prowler/providers/image/exceptions/exceptions.py b/prowler/providers/image/exceptions/exceptions.py index f819462883..ab83a2ed10 100644 --- a/prowler/providers/image/exceptions/exceptions.py +++ b/prowler/providers/image/exceptions/exceptions.py @@ -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 + ) diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index df06bfe966..f5a27117a3 100644 --- a/prowler/providers/image/image_provider.py +++ b/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,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. diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py index f52095d27f..ab9a11bdad 100644 --- a/tests/providers/image/image_provider_test.py +++ b/tests/providers/image/image_provider_test.py @@ -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