feat(image): add IMAGE scan path with per-image progress

- Add scan_per_image() generator to ImageProvider for per-image yields
- Replace single-shot IMAGE execution in Scan.scan() with per-image progress
- Extend checks_loader bypass to include IMAGE provider
This commit is contained in:
Andoni A. committed 2026-02-18 10:19:12 +01:00
1 parent 12f1a4f3bb
commit 34cbacc105
3 files changed
+68 -39

No files matched your search

+2 -2
View File
@@ -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 IAC/Image providers since they use Trivy directly
if provider in ("iac", "image"):
return set()
# Local subsets
+43 -36
View File
@@ -348,49 +348,56 @@ class Scan:
return
# Special handling for Image provider
if self._provider.type == "image":
elif self._provider.type == "image":
if isinstance(self._provider, ImageProvider):
logger.info("Running Image scan with Trivy...")
image_reports = self._provider.run()
findings = []
for report in image_reports:
finding_uid = f"{report.check_metadata.CheckID}-{report.resource_name}-{report.resource_id}"
total_images = len(self._provider.images)
self._number_of_checks_to_execute = max(total_images, 1)
status_enum = (
Status.FAIL if report.status == "FAIL" else Status.PASS
)
if report.muted:
status_enum = Status.MUTED
for i, (image_name, image_reports) in enumerate(
self._provider.scan_per_image()
):
findings = []
for report in image_reports:
finding_uid = (
f"{report.check_metadata.CheckID}"
f"-{report.resource_name}"
f"-{report.resource_id}"
)
status_enum = (
Status.FAIL if report.status == "FAIL" else Status.PASS
)
if report.muted:
status_enum = Status.MUTED
finding = Finding(
auth_method=self._provider.auth_method,
timestamp=datetime.datetime.now(timezone.utc),
account_uid=self._provider.audited_account,
account_name="Image Scan",
metadata=report.check_metadata,
uid=finding_uid,
status=status_enum,
status_extended=report.status_extended,
muted=report.muted,
resource_uid=report.resource_name,
resource_metadata=report.resource,
resource_name=report.resource_name,
resource_details=report.resource_details,
resource_tags={},
region=self._provider.region,
compliance={},
raw=report.resource,
)
findings.append(finding)
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=report.resource_id,
resource_metadata=report.resource,
resource_name=report.resource_name,
resource_details=report.resource_details,
resource_tags={},
region=report.region,
compliance={},
raw=report.resource,
)
findings.append(finding)
if self._status:
findings = [f for f in findings if f.status in self._status]
if self._status:
findings = [f for f in findings if f.status in self._status]
self._number_of_checks_completed = 1
self._number_of_checks_to_execute = 1
yield (100.0, findings)
self._number_of_checks_completed = i + 1
yield (self.progress, findings)
end_time = datetime.datetime.now()
self._duration = int((end_time - start_time).total_seconds())
+23 -1
View File
@@ -321,7 +321,6 @@ class ImageProvider(Provider):
def cleanup(self) -> None:
"""Clean up any resources after scanning."""
pass
def _process_finding(
self, finding: dict, image_name: str, finding_type: str
@@ -454,6 +453,29 @@ class ImageProvider(Provider):
finally:
self.cleanup()
def scan_per_image(
self,
) -> Generator[tuple[str, list[CheckReportImage]], None, None]:
"""Scan images one by one, yielding (image_name, findings) per image.
Unlike run() which returns all findings at once, this method yields
after each image completes, enabling progress tracking.
"""
try:
for image in self.images:
try:
image_findings = []
for batch in self._scan_single_image(image):
image_findings.extend(batch)
yield (image, image_findings)
except (ImageScanError, ImageTrivyBinaryNotFoundError):
raise
except Exception as error:
logger.error(f"Error scanning image {image}: {error}")
continue
finally:
self.cleanup()
def run_scan(self) -> Generator[list[CheckReportImage], None, None]:
"""
Run Trivy scan on all configured images.