From f42de0d21bef21c9de84e7336154d4e4f980919b Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Thu, 29 Jan 2026 08:41:11 +0100 Subject: [PATCH] feat(image): add container image provider POC Add initial proof of concept for a container image security scanning provider that uses Trivy for vulnerability detection in container images. --- prowler/__main__.py | 15 +- prowler/config/config.py | 5 +- prowler/lib/check/models.py | 33 ++ prowler/lib/check/utils.py | 4 +- prowler/lib/cli/parser.py | 5 +- prowler/lib/outputs/finding.py | 17 + prowler/lib/outputs/summary_table.py | 6 + prowler/providers/common/provider.py | 11 + prowler/providers/image/README.md | 236 ++++++++ prowler/providers/image/__init__.py | 0 prowler/providers/image/image_provider.py | 535 ++++++++++++++++++ prowler/providers/image/lib/__init__.py | 0 .../providers/image/lib/arguments/__init__.py | 0 .../image/lib/arguments/arguments.py | 89 +++ prowler/providers/image/models.py | 21 + 15 files changed, 966 insertions(+), 11 deletions(-) create mode 100644 prowler/providers/image/README.md create mode 100644 prowler/providers/image/__init__.py create mode 100644 prowler/providers/image/image_provider.py create mode 100644 prowler/providers/image/lib/__init__.py create mode 100644 prowler/providers/image/lib/arguments/__init__.py create mode 100644 prowler/providers/image/lib/arguments/arguments.py create mode 100644 prowler/providers/image/models.py diff --git a/prowler/__main__.py b/prowler/__main__.py index a6be311335..174fa326f6 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -118,6 +118,7 @@ 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.models import ImageOutputOptions from prowler.providers.kubernetes.models import KubernetesOutputOptions from prowler.providers.llm.models import LLMOutputOptions from prowler.providers.m365.models import M365OutputOptions @@ -204,8 +205,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 IAC, LLM, and Image providers + if provider not in ("iac", "llm", "image"): bulk_compliance_frameworks = Compliance.get_bulk(provider) # Complete checks metadata with the compliance framework specification bulk_checks_metadata = update_checks_metadata_with_compliance( @@ -262,8 +263,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 IAC, LLM, and Image providers + if provider not in ("iac", "llm", "image"): # Import custom checks from folder if checks_folder: custom_checks = parse_checks_from_folder(global_provider, checks_folder) @@ -346,6 +347,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": @@ -365,8 +368,8 @@ def prowler(): # Execute checks findings = [] - if provider == "iac" or provider == "llm": - # For IAC and LLM providers, run the scan directly + if provider in ("iac", "llm", "image"): + # For IAC, LLM, and Image providers, run the scan directly if provider == "llm": def streaming_callback(findings_batch): diff --git a/prowler/config/config.py b/prowler/config/config.py index e7c54c4a9d..970c7e0078 100644 --- a/prowler/config/config.py +++ b/prowler/config/config.py @@ -73,7 +73,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 8185ca1031..a4d345f6c8 100644 --- a/prowler/lib/check/models.py +++ b/prowler/lib/check/models.py @@ -163,6 +163,7 @@ class CheckMetadata(BaseModel): check_id and values.get("Provider") != "iac" and values.get("Provider") != "llm" + and values.get("Provider") != "image" ): service_from_check_id = check_id.split("_")[0] if service_name != service_from_check_id: @@ -183,6 +184,7 @@ class CheckMetadata(BaseModel): check_id and values.get("Provider") != "iac" and values.get("Provider") != "llm" + and values.get("Provider") != "image" ): if "-" in check_id: raise ValueError( @@ -791,6 +793,37 @@ class CheckReportIAC(Check_Report): ) +@dataclass +class CheckReportImage(Check_Report): + """Contains the Container Image Check's finding information using Trivy.""" + + resource_name: str + image_digest: str + package_name: str + installed_version: str + fixed_version: str + + def __init__( + self, metadata: dict = {}, finding: dict = {}, 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. + """ + super().__init__(metadata, finding) + + self.resource = finding + self.resource_name = image_name + 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 10b8c12abb..bb9fb0e285 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,dashboard,iac} ...", + usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,github,nhn,mongodbatlas,oraclecloud,alibabacloud,dashboard,iac,image} ...", epilog=""" Available Cloud Providers: - {aws,azure,gcp,kubernetes,m365,github,iac,llm,nhn,mongodbatlas,oraclecloud,alibabacloud} + {aws,azure,gcp,kubernetes,m365,github,iac,llm,image,nhn,mongodbatlas,oraclecloud,alibabacloud} aws AWS Provider azure Azure Provider gcp GCP Provider @@ -41,6 +41,7 @@ Available Cloud Providers: alibabacloud Alibaba Cloud Provider iac IaC Provider (Beta) llm LLM Provider (Beta) + image Container Image Provider (PoC) nhn NHN Provider (Unofficial) mongodbatlas MongoDB Atlas Provider (Beta) diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py index 4242116160..e9bd5ef26d 100644 --- a/prowler/lib/outputs/finding.py +++ b/prowler/lib/outputs/finding.py @@ -358,6 +358,23 @@ class Finding(BaseModel): ) 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_name", "") + 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 daf65192bb..6675013978 100644 --- a/prowler/lib/outputs/summary_table.py +++ b/prowler/lib/outputs/summary_table.py @@ -77,6 +77,12 @@ def display_summary_table( elif provider.type == "alibabacloud": entity_type = "Account" audited_entities = provider.identity.account_id + elif provider.type == "image": + entity_type = "Image" + if len(provider.images) == 1: + audited_entities = provider.images[0] + else: + audited_entities = f"{len(provider.images)} 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 16758ed940..573e90b312 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -266,6 +266,17 @@ 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, + 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/README.md b/prowler/providers/image/README.md new file mode 100644 index 0000000000..5f8ae8d94c --- /dev/null +++ b/prowler/providers/image/README.md @@ -0,0 +1,236 @@ +# Container Image Provider (PoC) + +This is a proof of concept implementation of a container image scanning provider for Prowler using Trivy. + +## Overview + +The Image Provider follows the Tool/Wrapper pattern established by the IaC provider. It delegates all scanning logic to Trivy's `trivy image` command and converts the output to Prowler's finding format. + +## Prerequisites + +### Trivy Installation + +Trivy must be installed and available in your PATH. Install using one of these methods: + +**macOS (Homebrew):** +```bash +brew install trivy +``` + +**Linux (apt):** +```bash +sudo apt-get install trivy +``` + +**Linux (rpm):** +```bash +sudo yum install trivy +``` + +**Docker:** +```bash +docker pull aquasecurity/trivy +``` + +For more installation options, see the [Trivy documentation](https://trivy.dev/latest/getting-started/installation/). + +## Usage + +### Basic Scan + +Scan a single container image: +```bash +poetry run python prowler-cli.py image --image nginx:latest +``` + +### Multiple Images + +Scan multiple images in a single run: +```bash +poetry run python prowler-cli.py image --image nginx:latest --image alpine:3.18 --image python:3.11 +``` + +### From File + +Scan images listed in a file (one per line): +```bash +# images.txt +nginx:latest +alpine:3.18 +python:3.11 +# This line is a comment and will be ignored + +poetry run python prowler-cli.py image --image-list images.txt +``` + +### Scanner Selection + +By default, the provider uses vulnerability and secret scanners. Customize with: +```bash +# Vulnerability scanning only +poetry run python prowler-cli.py image --image nginx:latest --scanners vuln + +# All scanners +poetry run python prowler-cli.py image --image nginx:latest --scanners vuln secret misconfig license +``` + +### Severity Filtering + +Filter findings by severity: +```bash +# Critical and high only +poetry run python prowler-cli.py image --image nginx:latest --trivy-severity CRITICAL HIGH +``` + +### Ignore Unfixed Vulnerabilities + +Skip vulnerabilities without available fixes: +```bash +poetry run python prowler-cli.py image --image nginx:latest --ignore-unfixed +``` + +### Custom Timeout + +Adjust Trivy scan timeout (default: 5m): +```bash +poetry run python prowler-cli.py image --image large-image:latest --timeout 10m +``` + +### Output Formats + +Export results in different formats: +```bash +# JSON and CSV (default includes html) +poetry run python prowler-cli.py image --image nginx:latest --output-formats json-ocsf csv + +# Specify output directory +poetry run python prowler-cli.py image --image nginx:latest --output-directory ./scan-results +``` + +## CLI Reference + +``` +prowler image [OPTIONS] + +Options: + --image, -I Container image to scan (can be specified multiple times) + --image-list File containing list of images to scan (one per line) + --scanners Trivy scanners: vuln, secret, misconfig, license + (default: vuln, secret) + --trivy-severity Filter: CRITICAL, HIGH, MEDIUM, LOW, UNKNOWN + --ignore-unfixed Ignore vulnerabilities without fixes + --timeout Trivy scan timeout (default: 5m) + +Standard Prowler Options: + --output-formats, -M Output formats (csv, json-ocsf, html) + --output-directory, -o Output directory + --output-filename, -F Custom output filename + --verbose Show all findings during execution + --no-banner, -b Hide Prowler banner +``` + +## Architecture + +``` +prowler/providers/image/ +├── __init__.py +├── image_provider.py # Main provider class +├── models.py # ImageOutputOptions +├── README.md # This file +└── lib/ + └── arguments/ + ├── __init__.py + └── arguments.py # CLI argument definitions +``` + +### Key Components + +1. **ImageProvider** (`image_provider.py`): + - Builds and executes `trivy image` commands + - Parses JSON output from Trivy + - Converts findings to `CheckReportImage` format + - Supports scanning multiple images in sequence + +2. **CheckReportImage** (`prowler/lib/check/models.py`): + - Extends `Check_Report` base class + - Stores vulnerability-specific fields (package name, versions) + +3. **ImageOutputOptions** (`models.py`): + - Customizes output filename generation + +4. **CLI Arguments** (`lib/arguments/arguments.py`): + - Defines image provider CLI arguments + - Validates required arguments + +## Known Limitations (PoC Scope) + +1. **Public Registries Only**: No authentication for private registries +2. **No Local Tar Support**: Cannot scan local image tar files +3. **No SBOM Export**: Does not generate SBOM output +4. **No Compliance Mapping**: No compliance framework integration +5. **Sequential Scanning**: Images scanned one at a time (no parallelization) + +## Future Work + +For full implementation, consider: + +1. **Registry Authentication**: + - Docker config.json support + - Environment variable credentials + - Cloud provider registry integration (ECR, GCR, ACR) + +2. **Local Image Support**: + - Scan from tar files (`--input` flag) + - Scan from Docker daemon + +3. **SBOM Generation**: + - CycloneDX output + - SPDX output + +4. **Performance**: + - Parallel image scanning + - Caching of vulnerability databases + +5. **Compliance Integration**: + - Map CVEs to compliance frameworks + - Custom compliance definitions + +6. **Enhanced Reporting**: + - Image-specific HTML reports + - Vulnerability trending + +## Trivy Output Format + +Trivy's JSON output structure for image scanning: + +```json +{ + "Results": [ + { + "Target": "nginx:latest (debian 11.7)", + "Type": "debian", + "Vulnerabilities": [ + { + "VulnerabilityID": "CVE-2023-1234", + "PkgName": "openssl", + "InstalledVersion": "1.1.1n-0+deb11u4", + "FixedVersion": "1.1.1n-0+deb11u5", + "Severity": "HIGH", + "Title": "Buffer overflow in...", + "Description": "...", + "PrimaryURL": "https://avd.aquasec.com/nvd/cve-2023-1234" + } + ], + "Secrets": [...], + "Misconfigurations": [...] + } + ] +} +``` + +## References + +- [Trivy Documentation](https://trivy.dev/docs/latest/) +- [Trivy Image Scanning](https://trivy.dev/docs/latest/guide/target/container_image/) +- [Trivy JSON Output](https://trivy.dev/docs/latest/guide/configuration/reporting/) +- [Prowler IaC Provider](../iac/) - Reference implementation 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/image_provider.py b/prowler/providers/image/image_provider.py new file mode 100644 index 0000000000..92cdaa1a03 --- /dev/null +++ b/prowler/providers/image/image_provider.py @@ -0,0 +1,535 @@ +import json +import subprocess +import sys +from typing import Generator, List + +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 + + +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" + audit_metadata: Audit_Metadata + + def __init__( + self, + images: list[str] = [], + image_list_file: str = None, + scanners: list[str] = ["vuln", "secret"], + trivy_severity: list[str] = [], + ignore_unfixed: bool = False, + timeout: str = "5m", + config_path: str = None, + config_content: dict = None, + fixer_config: dict = {}, + ): + logger.info("Instantiating Image Provider...") + + self.images = images or [] + self.image_list_file = image_list_file + self.scanners = scanners + self.trivy_severity = trivy_severity + 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" + + # Load images from file if provided + if image_list_file: + self._load_images_from_file(image_list_file) + + if not self.images: + logger.error("No images provided for scanning") + sys.exit(1) + + # 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 + + # 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: + with open(file_path, "r") as f: + for line in f: + line = line.strip() + if line and not line.startswith("#"): + self.images.append(line) + logger.info(f"Loaded {len(self.images)} images from {file_path}") + except FileNotFoundError: + logger.error(f"Image list file not found: {file_path}") + sys.exit(1) + except Exception as error: + logger.error(f"Error reading image list file: {error}") + sys.exit(1) + + @property + def auth_method(self): + return self._auth_method + + @property + def type(self): + return self._type + + @property + def identity(self): + return self._identity + + @property + def session(self): + return self._session + + @property + def audit_config(self): + return self._audit_config + + @property + def fixer_config(self): + return self._fixer_config + + def setup_session(self): + """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", + "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: + logger.critical( + f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}" + ) + sys.exit(1) + + 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 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.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) + + # 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 = [] + batch_size = 100 + + 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) >= 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) >= 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) >= batch_size: + yield batch + batch = [] + + # Yield remaining findings + if batch: + yield batch + + except Exception as error: + if "No such file or directory: 'trivy'" in str(error): + logger.critical( + "Trivy binary not found. Please install Trivy from " + "https://trivy.dev/latest/getting-started/installation/" + ) + sys.exit(1) + 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) + + def print_credentials(self): + """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.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, + raise_on_exception: bool = True, + provider_id: str = 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..e27efac97e --- /dev/null +++ b/prowler/providers/image/lib/arguments/arguments.py @@ -0,0 +1,89 @@ +SCANNERS_CHOICES = [ + "vuln", + "secret", + "misconfig", + "license", +] + +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( + "--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