diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 773776ce20..c4c53ef93b 100644 --- a/prowler/CHANGELOG.md +++ b/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 diff --git a/prowler/__main__.py b/prowler/__main__.py index 121642c675..2d7f7df1cb 100644 --- a/prowler/__main__.py +++ b/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%. diff --git a/prowler/config/config.py b/prowler/config/config.py index daa7378280..77674c7f93 100644 --- a/prowler/config/config.py +++ b/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( diff --git a/prowler/lib/check/models.py b/prowler/lib/check/models.py index 3ec9a10f47..648b962039 100644 --- a/prowler/lib/check/models.py +++ b/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.""" diff --git a/prowler/lib/check/utils.py b/prowler/lib/check/utils.py index 6d5ff69644..45f16cde32 100644 --- a/prowler/lib/check/utils.py +++ b/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 = [] diff --git a/prowler/lib/cli/parser.py b/prowler/lib/cli/parser.py index e21fb8e9d2..c3045dbd76 100644 --- a/prowler/lib/cli/parser.py +++ b/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) diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py index 128adc8dae..82f64850f8 100644 --- a/prowler/lib/outputs/finding.py +++ b/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 diff --git a/prowler/lib/outputs/summary_table.py b/prowler/lib/outputs/summary_table.py index 1a403d6a75..6a1d978d7c 100644 --- a/prowler/lib/outputs/summary_table.py +++ b/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): diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 339220df30..6f29c0b800 100644 --- a/prowler/providers/common/provider.py +++ b/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, diff --git a/prowler/providers/image/__init__.py b/prowler/providers/image/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/exceptions/__init__.py b/prowler/providers/image/exceptions/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/exceptions/exceptions.py b/prowler/providers/image/exceptions/exceptions.py new file mode 100644 index 0000000000..0e115a83ec --- /dev/null +++ b/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 + ) diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py new file mode 100644 index 0000000000..1e11651253 --- /dev/null +++ b/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)}", + ) diff --git a/prowler/providers/image/lib/__init__.py b/prowler/providers/image/lib/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/lib/arguments/__init__.py b/prowler/providers/image/lib/arguments/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/image/lib/arguments/arguments.py b/prowler/providers/image/lib/arguments/arguments.py new file mode 100644 index 0000000000..3809aec667 --- /dev/null +++ b/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, "") diff --git a/prowler/providers/image/models.py b/prowler/providers/image/models.py new file mode 100644 index 0000000000..232f8a6e39 --- /dev/null +++ b/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 diff --git a/tests/lib/cli/parser_test.py b/tests/lib/cli/parser_test.py index 324b9b6c20..f2192d8d0b 100644 --- a/tests/lib/cli/parser_test.py +++ b/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", diff --git a/tests/providers/image/image_fixtures.py b/tests/providers/image/image_fixtures.py new file mode 100644 index 0000000000..a51f934bda --- /dev/null +++ b/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) diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py new file mode 100644 index 0000000000..75bebaefc5 --- /dev/null +++ b/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)