add container names

This commit is contained in:
David Duran Rosich committed 2026-10-23 22:47:04 +02:00
1 parent f6e723a63e
commit 3003585a04
3 files changed
+93 -15

No files matched your search

+4 -2
View File
@@ -64,9 +64,11 @@ For the last one the scanner asks the Docker Engine API on the Unix socket for
`/containers/json` and scans each `Image` reference, de-duplicated and in `/containers/json` and scans each `Image` reference, de-duplicated and in
listing order. Stopped containers are not included. The scanner's own container listing order. Stopped containers are not included. The scanner's own container
image is skipped so a scheduled run does not report on itself; set image is skipped so a scheduled run does not report on itself; set
`INCLUDE_SELF=true` to include it. Scan log lines name the running container(s) `INCLUDE_SELF=true` to include it. Scan log lines and Apprise notifications name the running container(s)
that use each image, for example that use each image, for example
`Scanning nginx:1.27 (used by container web)`. `Scanning nginx:1.27 (used by container web)` and
`Vulnerability scan: nginx:1.27 [container: web]`. Explicitly passed images
omit the container context.
The socket is looked up in this order: The socket is looked up in this order:
+51 -13
View File
@@ -439,13 +439,31 @@ class ScanTarget:
image: str image: str
containers: list[str] = field(default_factory=list) containers: list[str] = field(default_factory=list)
@property
def container_names(self) -> str:
return ", ".join(self.containers)
@property
def container_label(self) -> str:
return "Container" if len(self.containers) == 1 else "Containers"
@property @property
def container_suffix(self) -> str: def container_suffix(self) -> str:
if not self.containers: if not self.containers:
return "" return ""
if len(self.containers) == 1: return f" (used by {self.container_label.lower()} {self.container_names})"
return f" (used by container {self.containers[0]})"
return f" (used by containers {', '.join(self.containers)})" @property
def notification_context(self) -> str:
if not self.containers:
return ""
return f" [{self.container_label.lower()}: {self.container_names}]"
@property
def container_detail(self) -> str:
if not self.containers:
return ""
return f"{self.container_label}: {self.container_names}"
def collect_targets(settings: Settings) -> list[ScanTarget]: def collect_targets(settings: Settings) -> list[ScanTarget]:
@@ -609,9 +627,17 @@ def notify(settings: Settings, title: str, body: str) -> bool:
return True return True
def build_body(image: str, result: ScanResult, report_path: str, max_findings: int) -> str: def build_body(
lines = [ image: str,
f"Image: {image}", result: ScanResult,
report_path: str,
max_findings: int,
container_detail: str = "",
) -> str:
lines = [f"Image: {image}"]
if container_detail:
lines.append(container_detail)
lines += [
f"Critical: {result.critical} | High: {result.high} | " f"Critical: {result.critical} | High: {result.high} | "
f"Medium: {result.medium} | Low: {result.low}", f"Medium: {result.medium} | Low: {result.low}",
f"Fixable (critical/high): {result.fixable}", f"Fixable (critical/high): {result.fixable}",
@@ -664,11 +690,17 @@ def scan_image(target: ScanTarget, settings: Settings, stats: RunStats) -> None:
if status != 0: if status != 0:
stats.scan_errors += 1 stats.scan_errors += 1
if settings.notify_on_scan_error: if settings.notify_on_scan_error:
body = ( lines = [f"Image: {image}"]
f"Image: {image}\n\n" if target.container_detail:
"Trivy could not complete the scan. Check the scanner logs for details.\n" lines.append(target.container_detail)
) lines += [
if not notify(settings, f"Scan failed: {image}", body): "",
"Trivy could not complete the scan. Check the scanner logs for details.",
"",
]
body = "\n".join(lines)
title = f"Scan failed: {image}{target.notification_context}"
if not notify(settings, title, body):
stats.notify_errors += 1 stats.notify_errors += 1
return return
@@ -693,10 +725,16 @@ def scan_image(target: ScanTarget, settings: Settings, stats: RunStats) -> None:
if notify_hits: if notify_hits:
title = ( title = (
f"{settings.apprise_title_prefix}: {image} " f"{settings.apprise_title_prefix}: {image}{target.notification_context} "
f"({result.critical} critical, {result.high} high)" f"({result.critical} critical, {result.high} high)"
) )
body = build_body(image, result, report_path, settings.max_findings) body = build_body(
image,
result,
report_path,
settings.max_findings,
target.container_detail,
)
if not notify(settings, title, body): if not notify(settings, title, body):
stats.notify_errors += 1 stats.notify_errors += 1
else: else:
+38
View File
@@ -274,6 +274,12 @@ class ImageListTests(unittest.TestCase):
self.assertEqual( self.assertEqual(
targets[0].container_suffix, " (used by containers web, web-worker)" targets[0].container_suffix, " (used by containers web, web-worker)"
) )
self.assertEqual(
targets[0].notification_context, " [containers: web, web-worker]"
)
self.assertEqual(
targets[0].container_detail, "Containers: web, web-worker"
)
def test_collect_targets_explicit_images_have_no_containers(self) -> None: def test_collect_targets_explicit_images_have_no_containers(self) -> None:
settings = scan.Settings(explicit_images=["nginx:1.27"]) settings = scan.Settings(explicit_images=["nginx:1.27"])
@@ -381,6 +387,18 @@ class ReportTests(unittest.TestCase):
self.assertIn("CVE-5", body) self.assertIn("CVE-5", body)
self.assertIn("no fix available", body) self.assertIn("no fix available", body)
def test_build_body_includes_container_detail(self) -> None:
result = scan.summarise(REPORT)
body = scan.build_body(
"nginx:1.27",
result,
"/reports/x.json",
max_findings=15,
container_detail="Containers: web, web-worker",
)
self.assertIn("Image: nginx:1.27", body)
self.assertIn("Containers: web, web-worker", body)
class TrivyCommandTests(unittest.TestCase): class TrivyCommandTests(unittest.TestCase):
def test_run_trivy_passes_vex_sources_in_order(self) -> None: def test_run_trivy_passes_vex_sources_in_order(self) -> None:
@@ -503,6 +521,26 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG"
"Scanning testphp-php (used by container testphp)", stderr.getvalue() "Scanning testphp-php (used by container testphp)", stderr.getvalue()
) )
def test_notification_includes_container(self) -> None:
containers = [container("a" * 64, "testphp-php", name="testphp")]
with FakeDocker(containers) as docker:
code = self.run_main([], docker.socket_path)
self.assertEqual(code, 1)
notification = Path(self.apprise_log).read_text()
self.assertIn("[container: testphp]", notification)
self.assertIn("Container: testphp", notification)
def test_scan_failure_notification_includes_container(self) -> None:
containers = [container("a" * 64, "testphp-php", name="testphp")]
with FakeDocker(containers) as docker:
code = self.run_main([], docker.socket_path, FAKE_TRIVY_FAIL="1")
self.assertEqual(code, 2)
notification = Path(self.apprise_log).read_text()
self.assertIn("Scan failed: testphp-php [container: testphp]", notification)
self.assertIn("Container: testphp", notification)
def test_exit_code_zero_when_policy_does_not_fail(self) -> None: def test_exit_code_zero_when_policy_does_not_fail(self) -> None:
with FakeDocker([container("a" * 64, "testphp-php")]) as docker: with FakeDocker([container("a" * 64, "testphp-php")]) as docker:
code = self.run_main([], docker.socket_path, FAIL_ON="NONE", NOTIFY_MIN_SEVERITY="NONE") code = self.run_main([], docker.socket_path, FAIL_ON="NONE", NOTIFY_MIN_SEVERITY="NONE")