From 9d0fd2ba44b59f90b163112c7b34196d7a71cdb5 Mon Sep 17 00:00:00 2001 From: David Duran Rosich Date: Fri, 23 Oct 2026 23:59:30 +0200 Subject: [PATCH] aggregate by cve --- README.md | 20 +++-- scan.py | 203 ++++++++++++++++++++++++++++++++++++++++----- tests/test_scan.py | 31 ++++++- 3 files changed, 224 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 5914f37..7fbe19d 100644 --- a/README.md +++ b/README.md @@ -19,12 +19,14 @@ host only needs Docker. JSON reports are kept on disk for auditing. `/reports/.json`. Optional VEX sources are applied through Trivy's `--vex` flag, so findings marked as not affected are filtered out before counting and notifying (see [VEX](#vex)). -2. It counts findings per severity and checks them against two thresholds: - - `NOTIFY_MIN_SEVERITY` (default `HIGH`) – when to send an Apprise alert. +2. Findings are aggregated by CVE across all scanned images. Two thresholds + are checked against the aggregate counts: + - `NOTIFY_MIN_SEVERITY` (default `HIGH`) – when to send a single Apprise + alert. - `FAIL_ON` (default `HIGH`) – when the script should exit with code `1` (useful for CI). -3. On a match it sends one notification per image listing the most relevant - CVEs, the affected package, the installed version, and the fixed version. +3. If any CVE reaches `NOTIFY_MIN_SEVERITY`, one notification is sent listing + each CVE with the affected packages and the containers/images that use them. ## Quick start @@ -64,11 +66,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 and Apprise notifications name the running container(s) -that use each image, for example -`Scanning nginx:1.27 (used by container web)` and -`Vulnerability scan: nginx:1.27 [container: web]`. Explicitly passed images -omit the container context. +`INCLUDE_SELF=true` to include it. Scan logs still name the running container(s) that use each image (e.g. +`Scanning nginx:1.27 (used by container web)`). Notifications are aggregated +by CVE after all images are scanned: a single report lists each CVE with the +containers and images it affects. Scan failures still notify per failed image +with container context. Explicitly passed images omit container context. The socket is looked up in this order: diff --git a/scan.py b/scan.py index 921914a..0affe2c 100755 --- a/scan.py +++ b/scan.py @@ -518,6 +518,107 @@ class ScanResult: return sum(1 for finding in self.findings if finding.get("FixedVersion")) +@dataclass +class AffectedTarget: + image: str + containers: list[str] + package: str = "" + installed: str = "" + fixed: str = "" + + +@dataclass +class CveAggregation: + cve: str + severity: str + affected: list[AffectedTarget] = field(default_factory=list) + packages: list[tuple[str, str, str]] = field(default_factory=list) + fixable: bool = False + + +@dataclass +class AggregatedReport: + critical: int = 0 + high: int = 0 + medium: int = 0 + low: int = 0 + cves: dict[str, CveAggregation] = field(default_factory=dict) + reports: list[str] = field(default_factory=list) + + @property + def counts(self) -> dict[str, int]: + return { + "CRITICAL": self.critical, + "HIGH": self.high, + "MEDIUM": self.medium, + "LOW": self.low, + } + + @property + def fixable(self) -> int: + return sum(1 for cve in self.cves.values() if cve.fixable and severity_rank(cve.severity) >= 3) + + def add_vulnerability( + self, vuln: dict, image: str, containers: list[str] + ) -> None: + severity = str(vuln.get("Severity", "")).upper() + if severity == "CRITICAL": + self.critical += 1 + elif severity == "HIGH": + self.high += 1 + elif severity == "MEDIUM": + self.medium += 1 + elif severity == "LOW": + self.low += 1 + else: + return + + cve_id = str(vuln.get("VulnerabilityID", "")) + if not cve_id: + return + pkg = str(vuln.get("PkgName", "")) + installed = str(vuln.get("InstalledVersion", "")) + fixed = str(vuln.get("FixedVersion") or "") + fixable = bool(fixed) + + if cve_id not in self.cves: + self.cves[cve_id] = CveAggregation( + cve=cve_id, severity=severity, fixable=fixable + ) + else: + existing = self.cves[cve_id] + if severity_rank(severity) > severity_rank(existing.severity): + existing.severity = severity + if fixable: + existing.fixable = True + + agg = self.cves[cve_id] + pkg_key = (pkg, installed, fixed) + if pkg_key not in agg.packages: + agg.packages.append(pkg_key) + + affected_key = (image, tuple(sorted(containers))) + for affected in agg.affected: + if affected.image == image and tuple(sorted(affected.containers)) == tuple( + sorted(containers) + ): + if pkg and not affected.package: + affected.package = pkg + affected.installed = installed + affected.fixed = fixed + break + else: + agg.affected.append( + AffectedTarget( + image=image, + containers=list(containers), + package=pkg, + installed=installed, + fixed=fixed, + ) + ) + + def summarise(report: dict) -> ScanResult: result = ScanResult() for scan in report.get("Results") or []: @@ -665,6 +766,51 @@ def build_body( return "\n".join(lines) +def build_aggregated_body( + aggregated: AggregatedReport, max_findings: int +) -> str: + lines = [ + f"Critical: {aggregated.critical} | High: {aggregated.high} | " + f"Medium: {aggregated.medium} | Low: {aggregated.low}", + f"Fixable (critical/high): {aggregated.fixable}", + "", + ] + + cves = sorted( + aggregated.cves.values(), + key=lambda item: ( + -severity_rank(item.severity), + item.cve, + ), + ) + total = len(cves) + if total > max_findings: + lines.append(f"CVEs (showing {max_findings} of {total}):") + else: + lines.append("CVEs:") + + for cve in cves[:max_findings]: + packages = [] + for pkg, installed, fixed in cve.packages[:3]: + if pkg: + packages.append(f"{pkg} {installed} -> {fixed or 'no fix available'}") + elif installed or fixed: + packages.append(f"{installed} -> {fixed or 'no fix available'}") + pkg_text = "; ".join(packages) if packages else "package unknown" + lines.append( + f"* [{cve.severity}] {cve.cve} - {pkg_text}" + ) + for affected in cve.affected: + containers = ", ".join(affected.containers) if affected.containers else "unknown" + lines.append(f" - containers: {containers} (image: {affected.image})") + + if aggregated.reports: + lines += ["", "Reports:"] + lines += [f"- {path}" for path in aggregated.reports[:20]] + + return "\n".join(lines) + + # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- @@ -683,7 +829,12 @@ def threshold_hits(result: ScanResult, rank: int) -> int: return sum(count for severity, count in result.counts.items() if severity_rank(severity) >= rank) -def scan_image(target: ScanTarget, settings: Settings, stats: RunStats) -> None: +def scan_image( + target: ScanTarget, + settings: Settings, + stats: RunStats, + aggregated: AggregatedReport, +) -> None: image = target.image context = target.container_suffix status, report_path = run_trivy(image, settings, context) @@ -720,23 +871,17 @@ def scan_image(target: ScanTarget, settings: Settings, stats: RunStats) -> None: f"({result.fixable} fixable high/critical)", ) + if report_path and report_path not in aggregated.reports: + aggregated.reports.append(report_path) + + for vuln in result.findings: + aggregated.add_vulnerability(vuln, image, target.containers) + notify_hits = threshold_hits(result, settings.notify_rank) fail_hits = threshold_hits(result, settings.fail_rank) if notify_hits: - title = ( - 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, - target.container_detail, - ) - if not notify(settings, title, body): - stats.notify_errors += 1 + log("DEBUG", f"{image}{context}: findings above {settings.notify_min_severity} tracked for aggregated report") else: log("INFO", f"{image}{context}: no findings at or above {settings.notify_min_severity}") @@ -747,10 +892,11 @@ def scan_image(target: ScanTarget, settings: Settings, stats: RunStats) -> None: def usage() -> str: return f"""Usage: {PROG} [IMAGE ...] - -Scan each Docker image with Trivy. When a finding at or above -NOTIFY_MIN_SEVERITY (default: HIGH) is detected, a notification is sent -through Apprise. + +Scan each Docker image with Trivy. Findings are aggregated by CVE across all +images and a single notification is sent when any reach +NOTIFY_MIN_SEVERITY (default: HIGH). Each CVE lists the containers and images +that are affected. Images can be passed as arguments or with IMAGES (space or newline separated) or IMAGES_FILE (one image per line, '#' starts a comment). @@ -819,8 +965,17 @@ def main(argv: list[str] | None = None) -> int: os.makedirs(settings.trivy_cache_dir, exist_ok=True) stats = RunStats() + aggregated = AggregatedReport() for target in targets: - scan_image(target, settings, stats) + scan_image(target, settings, stats, aggregated) + + if aggregated.cves and threshold_hits_from_counts(aggregated.counts, settings.notify_rank): + title = f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s) across {len(targets)} image(s)" + body = build_aggregated_body(aggregated, settings.max_findings) + if not notify(settings, title, body): + stats.notify_errors += 1 + elif aggregated.cves: + log("INFO", f"{len(aggregated.cves)} CVE(s) tracked but below NOTIFY_MIN_SEVERITY={settings.notify_min_severity}") log( "INFO", @@ -835,6 +990,16 @@ def main(argv: list[str] | None = None) -> int: return 0 +def threshold_hits_from_counts(counts: dict[str, int], rank: int) -> int: + if rank <= 0: + return 0 + total = 0 + for severity, count in counts.items(): + if severity_rank(severity) >= rank: + total += count + return total + + if __name__ == "__main__": try: raise SystemExit(main()) diff --git a/tests/test_scan.py b/tests/test_scan.py index 8d407f0..a288d45 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -399,6 +399,31 @@ class ReportTests(unittest.TestCase): self.assertIn("Image: nginx:1.27", body) self.assertIn("Containers: web, web-worker", body) + def test_build_aggregated_body_groups_by_cve(self) -> None: + aggregated = scan.AggregatedReport() + vuln1 = { + "Severity": "CRITICAL", + "VulnerabilityID": "CVE-1", + "PkgName": "libc", + "InstalledVersion": "1.0", + "FixedVersion": "1.1", + } + vuln2 = { + "Severity": "HIGH", + "VulnerabilityID": "CVE-1", + "PkgName": "libc", + "InstalledVersion": "1.0", + "FixedVersion": "1.1", + } + aggregated.add_vulnerability(vuln1, "nginx:1.27", ["web"]) + aggregated.add_vulnerability(vuln2, "redis:7", ["cache"]) + body = scan.build_aggregated_body(aggregated, 15) + self.assertIn("CVE-1", body) + self.assertIn("containers: web", body) + self.assertIn("image: nginx:1.27", body) + self.assertIn("containers: cache", body) + self.assertIn("image: redis:7", body) + class TrivyCommandTests(unittest.TestCase): def test_run_trivy_passes_vex_sources_in_order(self) -> None: @@ -528,8 +553,9 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG" self.assertEqual(code, 1) notification = Path(self.apprise_log).read_text() - self.assertIn("[container: testphp]", notification) - self.assertIn("Container: testphp", notification) + self.assertIn("CVE-2026-0001", notification) + self.assertIn("containers: testphp", notification) + self.assertIn("image: testphp-php", notification) def test_scan_failure_notification_includes_container(self) -> None: containers = [container("a" * 64, "testphp-php", name="testphp")] @@ -540,6 +566,7 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG" notification = Path(self.apprise_log).read_text() self.assertIn("Scan failed: testphp-php [container: testphp]", notification) self.assertIn("Container: testphp", notification) + self.assertIn("Trivy could not complete the scan", notification) def test_exit_code_zero_when_policy_does_not_fail(self) -> None: with FakeDocker([container("a" * 64, "testphp-php")]) as docker: