aggregate by cve

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())