diff --git a/api/src/backend/tasks/jobs/export.py b/api/src/backend/tasks/jobs/export.py index c2a947d249..4b8498f7e7 100644 --- a/api/src/backend/tasks/jobs/export.py +++ b/api/src/backend/tasks/jobs/export.py @@ -137,6 +137,7 @@ COMPLIANCE_CLASS_MAP = { # IaC provider doesn't have specific compliance frameworks yet # Trivy handles its own compliance checks ], + "image": [], "oraclecloud": [ (lambda name: name.startswith("cis_"), OracleCloudCIS), (lambda name: name.startswith("csa_"), OracleCloudCSA), diff --git a/prowler/lib/check/checks_loader.py b/prowler/lib/check/checks_loader.py index dab5535d77..bc1ea937be 100644 --- a/prowler/lib/check/checks_loader.py +++ b/prowler/lib/check/checks_loader.py @@ -22,8 +22,8 @@ def load_checks_to_execute( ) -> set: """Generate the list of checks to execute based on the cloud provider and the input arguments given""" try: - # Bypass check loading for IAC provider since it uses Trivy directly - if provider == "iac": + # Bypass check loading for providers that use Trivy directly + if provider in ("iac", "image"): return set() # Local subsets diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py index 82f64850f8..2c2d5a61f2 100644 --- a/prowler/lib/outputs/finding.py +++ b/prowler/lib/outputs/finding.py @@ -384,10 +384,12 @@ class Finding(BaseModel): 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", "" + image_name = getattr(check_output, "resource_name", "") + image_sha = getattr(check_output, "image_sha", "") + output_data["resource_name"] = image_name + output_data["resource_uid"] = ( + f"{image_name}:{image_sha}" if image_sha else image_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( diff --git a/prowler/lib/scan/scan.py b/prowler/lib/scan/scan.py index 6af2d26fbc..0b38f16b1b 100644 --- a/prowler/lib/scan/scan.py +++ b/prowler/lib/scan/scan.py @@ -27,6 +27,7 @@ from prowler.lib.scan.exceptions.exceptions import ( from prowler.providers.common.models import Audit_Metadata, ProviderOutputOptions from prowler.providers.common.provider import Provider from prowler.providers.iac.iac_provider import IacProvider +from prowler.providers.image.image_provider import ImageProvider class Scan: @@ -92,10 +93,10 @@ class Scan: except ValueError: raise ScanInvalidStatusError(f"Invalid status provided: {s}.") - # Special setup for IaC provider - override inputs to work with traditional flow - if provider.type == "iac": - # IaC doesn't use traditional Prowler checks, so clear all input parameters - # to avoid validation errors and let it flow through the normal logic + # Special setup for IaC/Image providers - override inputs to work with traditional flow + if provider.type in ("iac", "image"): + # These providers don't use traditional Prowler checks, so clear all input parameters + # to avoid validation errors and let them flow through the normal logic checks = None services = None excluded_checks = None @@ -160,8 +161,8 @@ class Scan: ) # Load checks to execute - if provider.type == "iac": - self._checks_to_execute = ["iac_scan"] # Dummy check name for IaC + if provider.type in ("iac", "image"): + self._checks_to_execute = [f"{provider.type}_scan"] else: self._checks_to_execute = sorted( load_checks_to_execute( @@ -200,8 +201,8 @@ class Scan: self._number_of_checks_to_execute = len(self._checks_to_execute) # Set up service-based checks tracking - if provider.type == "iac": - service_checks_to_execute = {"iac": set(["iac_scan"])} + if provider.type in ("iac", "image"): + service_checks_to_execute = {provider.type: set([f"{provider.type}_scan"])} else: service_checks_to_execute = get_service_checks_to_execute( self._checks_to_execute @@ -346,6 +347,75 @@ class Scan: self._duration = int((end_time - start_time).total_seconds()) return + # Special handling for Image provider + elif self._provider.type == "image": + if isinstance(self._provider, ImageProvider): + logger.info("Running Image scan with Trivy...") + + total_images = len(self._provider.images) + images_completed = 0 + + for image_name, image_findings in self._provider.scan_per_image(): + findings = [] + + for report in image_findings: + finding_uid = f"{report.check_metadata.CheckID}-{report.resource_name}-{report.resource_id}" + + status_enum = ( + Status.FAIL if report.status == "FAIL" else Status.PASS + ) + if report.muted: + status_enum = Status.MUTED + + image_sha = getattr(report, "image_sha", "") + resource_uid = ( + f"{image_name}:{image_sha}" if image_sha else image_name + ) + + finding = Finding( + auth_method="Registry", + timestamp=datetime.datetime.now(timezone.utc), + account_uid=getattr(self._provider, "registry", None) + or "image", + account_name="Container Registry", + metadata=report.check_metadata, + uid=finding_uid, + status=status_enum, + status_extended=report.status_extended, + muted=report.muted, + resource_uid=resource_uid, + resource_metadata=report.resource, + resource_name=image_name, + resource_details=report.resource_details, + resource_tags={}, + region=report.region, + compliance={}, + raw=report.resource, + ) + findings.append(finding) + + # Filter the findings by the status + if self._status: + findings = [f for f in findings if f.status in self._status] + + images_completed += 1 + progress = ( + images_completed / total_images * 100 + if total_images > 0 + else 100.0 + ) + + yield (progress, findings) + + # Update progress + self._number_of_checks_completed = 1 + self._number_of_checks_to_execute = 1 + + # Calculate duration + end_time = datetime.datetime.now() + self._duration = int((end_time - start_time).total_seconds()) + return + for check_name in checks_to_execute: try: # Recover service from check name