From 65b9c1fc846f898cb7203133c946507db55da2c3 Mon Sep 17 00:00:00 2001 From: David Duran Rosich Date: Sat, 24 Oct 2026 00:12:23 +0200 Subject: [PATCH] more limited notification --- scan.py | 42 +++++++++++++++++++++++------------------- tests/test_scan.py | 5 ++--- 2 files changed, 25 insertions(+), 22 deletions(-) diff --git a/scan.py b/scan.py index 60d6cec..290f203 100755 --- a/scan.py +++ b/scan.py @@ -708,7 +708,7 @@ def notify(settings: Settings, title: str, body: str) -> bool: log("WARN", "No APPRISE_URLS or APPRISE_CONFIG configured - notification skipped") return True - log("INFO", f"Sending Apprise notification: {title}") + log("INFO", f"Sending Apprise notification: {title} (title {len(title)} chars, body {len(body)} chars)") command = ["apprise", "--title", title, "--body", body] if settings.apprise_config: command += ["--config", settings.apprise_config] @@ -717,13 +717,18 @@ def notify(settings: Settings, title: str, body: str) -> bool: try: completed = subprocess.run(command, capture_output=True, text=True) except OSError as exc: - log("ERROR", f"Apprise notification failed: {title}: {exc}") + log("ERROR", f"Apprise notification failed: {title}: {exc} (title {len(title)} chars, body {len(body)} chars)") return False if completed.returncode != 0: - log("ERROR", f"Apprise notification failed: {title}") + log( + "ERROR", + f"Apprise notification failed: {title} (exit {completed.returncode}, title {len(title)} chars, body {len(body)} chars)", + ) if completed.stderr.strip(): - log("ERROR", completed.stderr.strip()) + log("ERROR", f"Apprise stderr: {completed.stderr.strip()}") + if completed.stdout.strip(): + log("ERROR", f"Apprise stdout: {completed.stdout.strip()}") return False return True @@ -766,7 +771,7 @@ def build_body( return "\n".join(lines) -MAX_NOTIFICATION_CHARS = 3500 +MAX_NOTIFICATION_CHARS = 2500 def _truncate_text(text: str, max_chars: int) -> str: @@ -798,6 +803,7 @@ def build_aggregated_body( ), ) total = len(cves) + max_findings = min(max_findings, 20) if total > max_findings: lines.append(f"CVEs (showing {max_findings} of {total}):") else: @@ -805,35 +811,30 @@ def build_aggregated_body( for cve in cves[:max_findings]: packages = [] - for pkg, installed, fixed in cve.packages[:3]: + for pkg, installed, fixed in cve.packages[:2]: 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[:5]: + for affected in cve.affected[:3]: 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)" - ) + lines.append(f" - containers: {containers} ({affected.image})") + if len(cve.affected) > 3: + lines.append(f" - ... +{len(cve.affected) - 3} more") 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[:10]] - if len(aggregated.reports) > 10: - lines.append(f"- ... and {len(aggregated.reports) - 10} more") + lines += ["", "Reports:"] + [f"- {p}" for p in aggregated.reports[:5]] + if len(aggregated.reports) > 5: + lines.append(f"- ... +{len(aggregated.reports) - 5} more") body = "\n".join(lines) - if len(body) > MAX_NOTIFICATION_CHARS: - return _truncate_text(body, MAX_NOTIFICATION_CHARS) - return body + return _truncate_text(body, MAX_NOTIFICATION_CHARS) # --------------------------------------------------------------------------- @@ -997,8 +998,11 @@ 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) + 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 elif aggregated.cves: diff --git a/tests/test_scan.py b/tests/test_scan.py index cb56273..187df4d 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -434,9 +434,9 @@ class ReportTests(unittest.TestCase): 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("nginx:1.27", body) self.assertIn("containers: cache", body) - self.assertIn("image: redis:7", body) + self.assertIn("redis:7", body) class TrivyCommandTests(unittest.TestCase): @@ -569,7 +569,6 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG" notification = Path(self.apprise_log).read_text() 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")]