diff --git a/prowler/lib/check/models.py b/prowler/lib/check/models.py index 648b962039..624a1b8e30 100644 --- a/prowler/lib/check/models.py +++ b/prowler/lib/check/models.py @@ -864,6 +864,7 @@ class CheckReportImage(Check_Report): resource_name: str resource_id: str image_digest: str + image_sha: str package_name: str installed_version: str fixed_version: str @@ -895,6 +896,7 @@ class CheckReportImage(Check_Report): or finding.get("ID", "") ) self.image_digest = finding.get("PkgID", "") + self.image_sha = "" self.package_name = finding.get("PkgName", "") self.installed_version = finding.get("InstalledVersion", "") self.fixed_version = finding.get("FixedVersion", "") 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 f07f28359d..1d6d8604cc 100644 --- a/prowler/lib/scan/scan.py +++ b/prowler/lib/scan/scan.py @@ -358,11 +358,17 @@ class Scan: for i, (image_name, image_reports) in enumerate( self._provider.scan_per_image() ): + # Build resource UID from image name + SHA (all reports share the same SHA) + image_sha = image_reports[0].image_sha if image_reports else "" + resource_uid = ( + f"{image_name}:{image_sha}" if image_sha else image_name + ) + findings = [] for report in image_reports: finding_uid = ( f"{report.check_metadata.CheckID}" - f"-{report.resource_name}" + f"-{image_name}" f"-{report.resource_id}" ) status_enum = ( @@ -382,9 +388,9 @@ class Scan: status=status_enum, status_extended=report.status_extended, muted=report.muted, - resource_uid=report.resource_id, + resource_uid=resource_uid, resource_metadata=report.resource, - resource_name=report.resource_name, + resource_name=image_name, resource_details=report.resource_details, resource_tags={}, region=report.region, diff --git a/prowler/providers/image/image_provider.py b/prowler/providers/image/image_provider.py index 41d499b122..be16580ba1 100644 --- a/prowler/providers/image/image_provider.py +++ b/prowler/providers/image/image_provider.py @@ -326,15 +326,20 @@ class ImageProvider(Provider): """Clean up any resources after scanning.""" def _process_finding( - self, finding: dict, image_name: str, finding_type: str + self, + finding: dict, + image: str, + trivy_target: str, + image_sha: 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.) + image: The clean container image name (e.g., "alpine:3.18") + trivy_target: The Trivy target string (e.g., "alpine:3.18 (alpine 3.18.0)") + image_sha: Short SHA from Trivy Metadata.ImageID for resource uniqueness Returns: CheckReportImage: The processed check report @@ -377,7 +382,7 @@ class ImageProvider(Provider): "CheckID": finding_id, "CheckTitle": finding.get("Title", finding_id), "CheckType": ["Container Image Security"], - "ServiceName": finding_type, + "ServiceName": "container-image", "SubServiceName": "", "ResourceIdTemplate": "", "Severity": trivy_severity, @@ -410,11 +415,13 @@ class ImageProvider(Provider): metadata = json.dumps(metadata_dict) report = CheckReportImage( - metadata=metadata, finding=finding, image_name=image_name + metadata=metadata, finding=finding, image_name=image ) report.status = finding_status report.status_extended = self._build_status_extended(finding) report.region = self.region + report.image_sha = image_sha + report.resource_details = trivy_target return report except Exception as error: @@ -563,6 +570,19 @@ class ImageProvider(Provider): logger.info(f"No findings for image: {image}") return + # Extract image digest for resource uniqueness + trivy_metadata = output.get("Metadata", {}) + image_id = trivy_metadata.get("ImageID", "") + if not image_id: + repo_digests = trivy_metadata.get("RepoDigests", []) + if repo_digests: + image_id = ( + repo_digests[0].split("@")[-1] + if "@" in repo_digests[0] + else "" + ) + short_sha = image_id.replace("sha256:", "")[:12] if image_id else "" + except json.JSONDecodeError as error: logger.error(f"Failed to parse Trivy output for {image}: {error}") logger.debug(f"Trivy stdout: {process.stdout[:500]}") @@ -573,11 +593,12 @@ class ImageProvider(Provider): 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) + report = self._process_finding( + vuln, image, target, image_sha=short_sha + ) batch.append(report) if len(batch) >= self.FINDING_BATCH_SIZE: yield batch @@ -585,7 +606,9 @@ class ImageProvider(Provider): # Process Secrets for secret in result.get("Secrets", []): - report = self._process_finding(secret, target, "secret") + report = self._process_finding( + secret, image, target, image_sha=short_sha + ) batch.append(report) if len(batch) >= self.FINDING_BATCH_SIZE: yield batch @@ -594,7 +617,7 @@ class ImageProvider(Provider): # Process Misconfigurations (from Dockerfile) for misconfig in result.get("Misconfigurations", []): report = self._process_finding( - misconfig, target, "misconfiguration" + misconfig, image, target, image_sha=short_sha ) batch.append(report) if len(batch) >= self.FINDING_BATCH_SIZE: diff --git a/tests/lib/scan/scan_test.py b/tests/lib/scan/scan_test.py index 93cae2f55f..1584663504 100644 --- a/tests/lib/scan/scan_test.py +++ b/tests/lib/scan/scan_test.py @@ -450,7 +450,7 @@ class TestImageScanPath: ) @staticmethod - def _make_report(check_id, image_name, status="FAIL"): + def _make_report(check_id, image_name, status="FAIL", image_sha="abc123def456"): report = MagicMock() report.check_metadata = TestImageScanPath._make_check_metadata(check_id) report.resource_name = image_name @@ -461,6 +461,7 @@ class TestImageScanPath: report.resource = {} report.resource_details = "" report.region = "container" + report.image_sha = image_sha return report @pytest.fixture @@ -492,6 +493,10 @@ class TestImageScanPath: assert f.status.name in ("PASS", "FAIL") assert f.auth_method == "Registry" assert f.account_name == "Container Registry" + # resource_uid should be image_name:sha + assert ":abc123def456" in f.resource_uid + # resource_name should be clean image name + assert f.resource_name in ("img1:latest", "img2:latest") def test_image_scan_progress_tracking(self, mock_image_provider): """Verify progress increments per image.""" diff --git a/tests/providers/image/image_fixtures.py b/tests/providers/image/image_fixtures.py index a51f934bda..920a7f225a 100644 --- a/tests/providers/image/image_fixtures.py +++ b/tests/providers/image/image_fixtures.py @@ -45,8 +45,16 @@ SAMPLE_UNKNOWN_SEVERITY_FINDING = { "Description": "An issue with unknown severity.", } +# Sample image SHA for testing (first 12 chars of a sha256 digest) +SAMPLE_IMAGE_SHA = "c1aabb73d233" +SAMPLE_IMAGE_ID = f"sha256:{SAMPLE_IMAGE_SHA}abcdef1234567890" + # Full Trivy JSON output structure with a single vulnerability SAMPLE_TRIVY_IMAGE_OUTPUT = { + "Metadata": { + "ImageID": SAMPLE_IMAGE_ID, + "RepoDigests": [f"alpine@sha256:{SAMPLE_IMAGE_SHA}abcdef1234567890"], + }, "Results": [ { "Target": "alpine:3.18 (alpine 3.18.0)", @@ -55,11 +63,15 @@ SAMPLE_TRIVY_IMAGE_OUTPUT = { "Secrets": [], "Misconfigurations": [], } - ] + ], } # Full Trivy JSON output with mixed finding types SAMPLE_TRIVY_MULTI_TYPE_OUTPUT = { + "Metadata": { + "ImageID": SAMPLE_IMAGE_ID, + "RepoDigests": [f"myimage@sha256:{SAMPLE_IMAGE_SHA}abcdef1234567890"], + }, "Results": [ { "Target": "myimage:latest (debian 12)", @@ -68,7 +80,36 @@ SAMPLE_TRIVY_MULTI_TYPE_OUTPUT = { "Secrets": [SAMPLE_SECRET_FINDING], "Misconfigurations": [SAMPLE_MISCONFIGURATION_FINDING], } - ] + ], +} + +# Trivy output with only RepoDigests (no ImageID) for fallback testing +SAMPLE_TRIVY_REPO_DIGEST_ONLY_OUTPUT = { + "Metadata": { + "RepoDigests": ["alpine@sha256:e5f6g7h8i9j0abcdef1234567890"], + }, + "Results": [ + { + "Target": "alpine:3.18 (alpine 3.18.0)", + "Type": "alpine", + "Vulnerabilities": [SAMPLE_VULNERABILITY_FINDING], + "Secrets": [], + "Misconfigurations": [], + } + ], +} + +# Trivy output with no Metadata at all +SAMPLE_TRIVY_NO_METADATA_OUTPUT = { + "Results": [ + { + "Target": "alpine:3.18 (alpine 3.18.0)", + "Type": "alpine", + "Vulnerabilities": [SAMPLE_VULNERABILITY_FINDING], + "Secrets": [], + "Misconfigurations": [], + } + ], } @@ -90,3 +131,13 @@ def get_invalid_trivy_output(): def get_multi_type_trivy_output(): """Return Trivy output with multiple finding types as string.""" return json.dumps(SAMPLE_TRIVY_MULTI_TYPE_OUTPUT) + + +def get_repo_digest_only_trivy_output(): + """Return Trivy output with only RepoDigests (no ImageID) as string.""" + return json.dumps(SAMPLE_TRIVY_REPO_DIGEST_ONLY_OUTPUT) + + +def get_no_metadata_trivy_output(): + """Return Trivy output with no Metadata as string.""" + return json.dumps(SAMPLE_TRIVY_NO_METADATA_OUTPUT) diff --git a/tests/providers/image/image_provider_test.py b/tests/providers/image/image_provider_test.py index 59ae20ce34..fc36b9e617 100644 --- a/tests/providers/image/image_provider_test.py +++ b/tests/providers/image/image_provider_test.py @@ -20,6 +20,7 @@ from prowler.providers.image.exceptions.exceptions import ( ) from prowler.providers.image.image_provider import ImageProvider from tests.providers.image.image_fixtures import ( + SAMPLE_IMAGE_SHA, SAMPLE_MISCONFIGURATION_FINDING, SAMPLE_SECRET_FINDING, SAMPLE_UNKNOWN_SEVERITY_FINDING, @@ -27,6 +28,8 @@ from tests.providers.image.image_fixtures import ( get_empty_trivy_output, get_invalid_trivy_output, get_multi_type_trivy_output, + get_no_metadata_trivy_output, + get_repo_digest_only_trivy_output, get_sample_trivy_json_output, ) @@ -120,21 +123,24 @@ class TestImageProvider: provider = _make_provider() report = provider._process_finding( SAMPLE_VULNERABILITY_FINDING, + "alpine:3.18", "alpine:3.18 (alpine 3.18.0)", - "alpine", + image_sha="c1aabb73d233", ) 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.ServiceName == "container-image" 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.resource_name == "alpine:3.18" + assert report.image_sha == "c1aabb73d233" + assert report.resource_details == "alpine:3.18 (alpine 3.18.0)" assert report.region == "container" assert report.check_metadata.Categories == ["vulnerability"] assert report.check_metadata.RelatedUrl == "" @@ -145,14 +151,14 @@ class TestImageProvider: report = provider._process_finding( SAMPLE_SECRET_FINDING, "myimage:latest", - "secret", + "myimage:latest (debian 12)", ) 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" + assert report.check_metadata.ServiceName == "container-image" assert report.check_metadata.Categories == ["secrets"] def test_process_finding_misconfiguration(self): @@ -161,13 +167,13 @@ class TestImageProvider: report = provider._process_finding( SAMPLE_MISCONFIGURATION_FINDING, "myimage:latest", - "misconfiguration", + "myimage:latest (debian 12)", ) assert isinstance(report, CheckReportImage) assert report.check_metadata.CheckID == "DS001" assert report.check_metadata.Severity == "medium" - assert report.check_metadata.ServiceName == "misconfiguration" + assert report.check_metadata.ServiceName == "container-image" assert report.check_metadata.Categories == [] def test_process_finding_unknown_severity(self): @@ -176,7 +182,7 @@ class TestImageProvider: report = provider._process_finding( SAMPLE_UNKNOWN_SEVERITY_FINDING, "myimage:latest", - "alpine", + "myimage:latest (alpine 3.18.0)", ) assert report.check_metadata.Severity == "informational" @@ -195,6 +201,9 @@ class TestImageProvider: assert len(reports) == 1 assert reports[0].check_metadata.CheckID == "CVE-2024-1234" + assert reports[0].image_sha == SAMPLE_IMAGE_SHA + assert reports[0].resource_name == "alpine:3.18" + assert reports[0].check_metadata.ServiceName == "container-image" @patch("subprocess.run") def test_run_scan_empty_output(self, mock_subprocess): @@ -394,6 +403,51 @@ class TestImageProvider: for _ in provider._scan_single_image("private/image:latest"): pass + @patch("subprocess.run") + def test_sha_extraction_from_image_id(self, mock_subprocess): + """Test that image_sha is extracted from Trivy Metadata.ImageID.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_sample_trivy_json_output(), stderr="" + ) + + reports = [] + for batch in provider._scan_single_image("alpine:3.18"): + reports.extend(batch) + + assert len(reports) == 1 + assert reports[0].image_sha == SAMPLE_IMAGE_SHA + + @patch("subprocess.run") + def test_sha_extraction_fallback_to_repo_digests(self, mock_subprocess): + """Test that image_sha falls back to RepoDigests when ImageID is absent.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_repo_digest_only_trivy_output(), stderr="" + ) + + reports = [] + for batch in provider._scan_single_image("alpine:3.18"): + reports.extend(batch) + + assert len(reports) == 1 + assert reports[0].image_sha == "e5f6g7h8i9j0" + + @patch("subprocess.run") + def test_sha_extraction_no_metadata(self, mock_subprocess): + """Test that image_sha is empty when no Metadata is present.""" + provider = _make_provider() + mock_subprocess.return_value = MagicMock( + returncode=0, stdout=get_no_metadata_trivy_output(), stderr="" + ) + + reports = [] + for batch in provider._scan_single_image("alpine:3.18"): + reports.extend(batch) + + assert len(reports) == 1 + assert reports[0].image_sha == "" + @patch("subprocess.run") def test_run_scan_propagates_scan_error(self, mock_subprocess): """Test that run_scan() re-raises ImageScanError instead of swallowing it."""