diff --git a/README.md b/README.md index 75ff7b1..5914f37 100644 --- a/README.md +++ b/README.md @@ -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 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 -`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 -`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: diff --git a/scan.py b/scan.py index 0356e35..921914a 100755 --- a/scan.py +++ b/scan.py @@ -439,13 +439,31 @@ class ScanTarget: image: str 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 def container_suffix(self) -> str: if not self.containers: return "" - if len(self.containers) == 1: - return f" (used by container {self.containers[0]})" - return f" (used by containers {', '.join(self.containers)})" + return f" (used by {self.container_label.lower()} {self.container_names})" + + @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]: @@ -609,9 +627,17 @@ def notify(settings: Settings, title: str, body: str) -> bool: return True -def build_body(image: str, result: ScanResult, report_path: str, max_findings: int) -> str: - lines = [ - f"Image: {image}", +def build_body( + image: str, + 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"Medium: {result.medium} | Low: {result.low}", f"Fixable (critical/high): {result.fixable}", @@ -664,11 +690,17 @@ def scan_image(target: ScanTarget, settings: Settings, stats: RunStats) -> None: if status != 0: stats.scan_errors += 1 if settings.notify_on_scan_error: - body = ( - f"Image: {image}\n\n" - "Trivy could not complete the scan. Check the scanner logs for details.\n" - ) - if not notify(settings, f"Scan failed: {image}", body): + lines = [f"Image: {image}"] + if target.container_detail: + lines.append(target.container_detail) + lines += [ + "", + "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 return @@ -693,10 +725,16 @@ def scan_image(target: ScanTarget, settings: Settings, stats: RunStats) -> None: if notify_hits: title = ( - f"{settings.apprise_title_prefix}: {image} " + f"{settings.apprise_title_prefix}: {image}{target.notification_context} " 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): stats.notify_errors += 1 else: diff --git a/tests/test_scan.py b/tests/test_scan.py index b42ac3d..8d407f0 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -274,6 +274,12 @@ class ImageListTests(unittest.TestCase): self.assertEqual( 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: settings = scan.Settings(explicit_images=["nginx:1.27"]) @@ -381,6 +387,18 @@ class ReportTests(unittest.TestCase): self.assertIn("CVE-5", 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): 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() ) + 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: with FakeDocker([container("a" * 64, "testphp-php")]) as docker: code = self.run_main([], docker.socket_path, FAIL_ON="NONE", NOTIFY_MIN_SEVERITY="NONE")