ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(image): make image the resource instead of individual findings

Browse Source 
- Add image_sha field to CheckReportImage from Trivy Metadata.ImageID
- Build resource_uid as "{image_name}:{short_sha}" in both CLI and API paths
- Set ServiceName to "container-image" for all image findings
- Pass clean image name as resource_name, Trivy target as resource_details
This commit is contained in:
Andoni A.
2026-02-18 14:47:37 +01:00
parent acbfdfa4b2
commit 6be6f558c1
7 changed files with 169 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/lib/check/models.py
+2
View File
@@ -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", "")
prowler/lib/outputs/finding.py
+5 -3
View File
@@ -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(
prowler/lib/scan/scan.py
+9 -3
View File
@@ -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,
prowler/providers/image/image_provider.py
+32 -9
View File
@@ -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:
tests/lib/scan/scan_test.py
+6 -1
View File
@@ -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."""
tests/providers/image/image_fixtures.py
+53 -2
View File
@@ -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)
tests/providers/image/image_provider_test.py
+62 -8
View File
@@ -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."""