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.
This commit is contained in:
Andoni A.
2026-01-29 08:41:11 +01:00
parent b7f83da012
commit f42de0d21b
15 changed files with 966 additions and 11 deletions
+9 -6
View File
@@ -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):
+4 -1
View File
@@ -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(
+33
View File
@@ -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."""
+2 -2
View File
@@ -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 = []
+3 -2
View File
@@ -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)
+17
View File
@@ -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
+6
View File
@@ -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):
+11
View File
@@ -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,
+236
View File
@@ -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
View File
+535
View File
@@ -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)}",
)
@@ -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, "")
+21
View File
@@ -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