diff --git a/scan.py b/scan.py index 290f203..4351118 100755 --- a/scan.py +++ b/scan.py @@ -709,7 +709,7 @@ def notify(settings: Settings, title: str, body: str) -> bool: return True log("INFO", f"Sending Apprise notification: {title} (title {len(title)} chars, body {len(body)} chars)") - command = ["apprise", "--title", title, "--body", body] + command = ["apprise", "-t", title, "-b", body, "-vv"] if settings.apprise_config: command += ["--config", settings.apprise_config] command += split_words(settings.apprise_urls) @@ -772,6 +772,12 @@ def build_body( MAX_NOTIFICATION_CHARS = 2500 +DISCORD_MAX_NOTIFICATION_CHARS = 1800 + + +def uses_discord(settings: Settings) -> bool: + combined = f"{settings.apprise_urls} {settings.apprise_config}" + return "discord://" in combined def _truncate_text(text: str, max_chars: int) -> str: @@ -786,8 +792,31 @@ def _truncate_text(text: str, max_chars: int) -> str: def build_aggregated_body( - aggregated: AggregatedReport, max_findings: int + aggregated: AggregatedReport, max_findings: int, compact: bool = False ) -> str: + if compact: + cves = sorted( + aggregated.cves.values(), + key=lambda item: ( + -severity_rank(item.severity), + item.cve, + ), + ) + total = len(cves) + max_findings = min(max_findings, 8) + lines = [ + f"C:{aggregated.critical} H:{aggregated.high} M:{aggregated.medium} L:{aggregated.low}", + f"Fixable:{aggregated.fixable}", + ] + for cve in cves[:max_findings]: + affected_count = len(cve.affected) + lines.append(f"- [{cve.severity}] {cve.cve} ({affected_count} tgt)") + remaining = total - min(total, max_findings) + if remaining > 0: + lines.append(f"+{remaining} more") + body = "\n".join(lines) + return _truncate_text(body, DISCORD_MAX_NOTIFICATION_CHARS) + lines = [ f"Critical: {aggregated.critical} | High: {aggregated.high} | " f"Medium: {aggregated.medium} | Low: {aggregated.low}", @@ -803,7 +832,7 @@ def build_aggregated_body( ), ) total = len(cves) - max_findings = min(max_findings, 20) + max_findings = min(max_findings, 15) if total > max_findings: lines.append(f"CVEs (showing {max_findings} of {total}):") else: @@ -811,27 +840,27 @@ def build_aggregated_body( for cve in cves[:max_findings]: packages = [] - for pkg, installed, fixed in cve.packages[:2]: + for pkg, installed, fixed in cve.packages[:1]: 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" + pkg_text = "; ".join(packages) if packages else "unknown" lines.append(f"* [{cve.severity}] {cve.cve} - {pkg_text}") - for affected in cve.affected[:3]: + for affected in cve.affected[:2]: containers = ", ".join(affected.containers) if affected.containers else "unknown" - lines.append(f" - containers: {containers} ({affected.image})") - if len(cve.affected) > 3: - lines.append(f" - ... +{len(cve.affected) - 3} more") + lines.append(f" - {containers} ({affected.image})") + if len(cve.affected) > 2: + lines.append(f" - +{len(cve.affected) - 2} more") remaining = total - min(total, max_findings) if remaining > 0: - lines.append(f"... and {remaining} more CVE(s)") + lines.append(f"... +{remaining} more CVE(s)") if aggregated.reports: - lines += ["", "Reports:"] + [f"- {p}" for p in aggregated.reports[:5]] - if len(aggregated.reports) > 5: - lines.append(f"- ... +{len(aggregated.reports) - 5} more") + lines += ["", "Reports:"] + [f"- {p}" for p in aggregated.reports[:3]] + if len(aggregated.reports) > 3: + lines.append(f"- +{len(aggregated.reports) - 3} more") body = "\n".join(lines) return _truncate_text(body, MAX_NOTIFICATION_CHARS) @@ -876,8 +905,11 @@ def scan_image( "", ] body = "\n".join(lines) - body = _truncate_text(body, MAX_NOTIFICATION_CHARS) + discord = uses_discord(settings) + body = _truncate_text(body, DISCORD_MAX_NOTIFICATION_CHARS if discord else MAX_NOTIFICATION_CHARS) title = f"Scan failed: {image}{target.notification_context}" + if len(title) > 200: + title = f"Scan failed: {image}" if not notify(settings, title, body): stats.notify_errors += 1 return @@ -998,13 +1030,25 @@ 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)" - if len(title) > 250: - title = f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s) across {len(targets)} img" - body = build_aggregated_body(aggregated, settings.max_findings) - body = _truncate_text(body, MAX_NOTIFICATION_CHARS) + if len(title) > 200: + title = f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s)" + discord = uses_discord(settings) + body = build_aggregated_body(aggregated, settings.max_findings, compact=discord) + if discord: + body = _truncate_text(body, DISCORD_MAX_NOTIFICATION_CHARS) + else: + body = _truncate_text(body, MAX_NOTIFICATION_CHARS) log("DEBUG", f"Preparing aggregated notification: title {len(title)} chars, body {len(body)} chars, {len(aggregated.cves)} CVEs") if not notify(settings, title, body): - stats.notify_errors += 1 + log("WARN", "Retrying aggregated notification with minimal summary") + minimal = ( + f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s) across {len(targets)} image(s)\n" + f"C:{aggregated.critical} H:{aggregated.high} M:{aggregated.medium} L:{aggregated.low}\n" + f"Full details in reports in {settings.report_dir}" + ) + minimal = _truncate_text(minimal, DISCORD_MAX_NOTIFICATION_CHARS if discord else MAX_NOTIFICATION_CHARS) + if not notify(settings, title, minimal): + 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}") diff --git a/tests/test_scan.py b/tests/test_scan.py index 187df4d..0dffe65 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -413,6 +413,20 @@ class ReportTests(unittest.TestCase): body = scan.build_aggregated_body(aggregated, 200) self.assertLessEqual(len(body), scan.MAX_NOTIFICATION_CHARS) + def test_aggregated_body_compact_for_discord(self) -> None: + aggregated = scan.AggregatedReport() + for i in range(50): + vuln = { + "Severity": "CRITICAL", + "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, 50, compact=True) + self.assertLessEqual(len(body), scan.DISCORD_MAX_NOTIFICATION_CHARS) + def test_build_aggregated_body_groups_by_cve(self) -> None: aggregated = scan.AggregatedReport() vuln1 = { @@ -433,9 +447,9 @@ class ReportTests(unittest.TestCase): 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("web", body) self.assertIn("nginx:1.27", body) - self.assertIn("containers: cache", body) + self.assertIn("cache", body) self.assertIn("redis:7", body) @@ -568,7 +582,7 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG" self.assertEqual(code, 1) notification = Path(self.apprise_log).read_text() self.assertIn("CVE-2026-0001", notification) - self.assertIn("containers: testphp", notification) + self.assertIn("testphp", notification) def test_scan_failure_notification_includes_container(self) -> None: containers = [container("a" * 64, "testphp-php", name="testphp")]