diff --git a/scan.py b/scan.py index 0affe2c..60d6cec 100755 --- a/scan.py +++ b/scan.py @@ -766,6 +766,20 @@ def build_body( return "\n".join(lines) +MAX_NOTIFICATION_CHARS = 3500 + + +def _truncate_text(text: str, max_chars: int) -> str: + if len(text) <= max_chars: + return text + if max_chars <= 0: + return "" + marker = "\n... (truncated)\n" + if max_chars <= len(marker): + return text[:max_chars] + return text[: max_chars - len(marker)] + marker + + def build_aggregated_body( aggregated: AggregatedReport, max_findings: int ) -> str: @@ -797,18 +811,29 @@ def build_aggregated_body( 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: + lines.append(f"* [{cve.severity}] {cve.cve} - {pkg_text}") + for affected in cve.affected[:5]: containers = ", ".join(affected.containers) if affected.containers else "unknown" lines.append(f" - containers: {containers} (image: {affected.image})") + if len(cve.affected) > 5: + lines.append( + f" - ... and {len(cve.affected) - 5} more affected image/container(s)" + ) + + remaining = total - min(total, max_findings) + if remaining > 0: + lines.append(f"... and {remaining} more CVE(s)") if aggregated.reports: lines += ["", "Reports:"] - lines += [f"- {path}" for path in aggregated.reports[:20]] + lines += [f"- {path}" for path in aggregated.reports[:10]] + if len(aggregated.reports) > 10: + lines.append(f"- ... and {len(aggregated.reports) - 10} more") - return "\n".join(lines) + body = "\n".join(lines) + if len(body) > MAX_NOTIFICATION_CHARS: + return _truncate_text(body, MAX_NOTIFICATION_CHARS) + return body # --------------------------------------------------------------------------- @@ -850,6 +875,7 @@ def scan_image( "", ] body = "\n".join(lines) + body = _truncate_text(body, MAX_NOTIFICATION_CHARS) title = f"Scan failed: {image}{target.notification_context}" if not notify(settings, title, body): stats.notify_errors += 1 @@ -972,6 +998,7 @@ def main(argv: list[str] | None = None) -> int: 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) + body = _truncate_text(body, MAX_NOTIFICATION_CHARS) if not notify(settings, title, body): stats.notify_errors += 1 elif aggregated.cves: diff --git a/tests/test_scan.py b/tests/test_scan.py index a288d45..cb56273 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -399,6 +399,20 @@ class ReportTests(unittest.TestCase): self.assertIn("Image: nginx:1.27", body) self.assertIn("Containers: web, web-worker", body) + def test_aggregated_body_truncates_large_notifications(self) -> None: + aggregated = scan.AggregatedReport() + for i in range(500): + vuln = { + "Severity": "HIGH", + "VulnerabilityID": f"CVE-2026-{i:05d}", + "PkgName": "libc", + "InstalledVersion": "1.0", + "FixedVersion": "1.1", + } + aggregated.add_vulnerability(vuln, f"img{i}:latest", [f"c{i}"]) + body = scan.build_aggregated_body(aggregated, 200) + self.assertLessEqual(len(body), scan.MAX_NOTIFICATION_CHARS) + def test_build_aggregated_body_groups_by_cve(self) -> None: aggregated = scan.AggregatedReport() vuln1 = {