more limited notification

This commit is contained in:
David Duran Rosich committed 2026-10-24 00:12:23 +02:00
1 parent 014a1d5563
commit 65b9c1fc84
2 files changed
+25 -22

No files matched your search

+23 -19
View File
@@ -708,7 +708,7 @@ def notify(settings: Settings, title: str, body: str) -> bool:
log("WARN", "No APPRISE_URLS or APPRISE_CONFIG configured - notification skipped") log("WARN", "No APPRISE_URLS or APPRISE_CONFIG configured - notification skipped")
return True 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] command = ["apprise", "--title", title, "--body", body]
if settings.apprise_config: if settings.apprise_config:
command += ["--config", settings.apprise_config] command += ["--config", settings.apprise_config]
@@ -717,13 +717,18 @@ def notify(settings: Settings, title: str, body: str) -> bool:
try: try:
completed = subprocess.run(command, capture_output=True, text=True) completed = subprocess.run(command, capture_output=True, text=True)
except OSError as exc: 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 return False
if completed.returncode != 0: 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(): 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 False
return True return True
@@ -766,7 +771,7 @@ def build_body(
return "\n".join(lines) return "\n".join(lines)
MAX_NOTIFICATION_CHARS = 3500 MAX_NOTIFICATION_CHARS = 2500
def _truncate_text(text: str, max_chars: int) -> str: def _truncate_text(text: str, max_chars: int) -> str:
@@ -798,6 +803,7 @@ def build_aggregated_body(
), ),
) )
total = len(cves) total = len(cves)
max_findings = min(max_findings, 20)
if total > max_findings: if total > max_findings:
lines.append(f"CVEs (showing {max_findings} of {total}):") lines.append(f"CVEs (showing {max_findings} of {total}):")
else: else:
@@ -805,35 +811,30 @@ def build_aggregated_body(
for cve in cves[:max_findings]: for cve in cves[:max_findings]:
packages = [] packages = []
for pkg, installed, fixed in cve.packages[:3]: for pkg, installed, fixed in cve.packages[:2]:
if pkg: if pkg:
packages.append(f"{pkg} {installed} -> {fixed or 'no fix available'}") packages.append(f"{pkg} {installed} -> {fixed or 'no fix available'}")
elif installed or fixed: elif installed or fixed:
packages.append(f"{installed} -> {fixed or 'no fix available'}") 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 "package unknown"
lines.append(f"* [{cve.severity}] {cve.cve} - {pkg_text}") 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" containers = ", ".join(affected.containers) if affected.containers else "unknown"
lines.append(f" - containers: {containers} (image: {affected.image})") lines.append(f" - containers: {containers} ({affected.image})")
if len(cve.affected) > 5: if len(cve.affected) > 3:
lines.append( lines.append(f" - ... +{len(cve.affected) - 3} more")
f" - ... and {len(cve.affected) - 5} more affected image/container(s)"
)
remaining = total - min(total, max_findings) remaining = total - min(total, max_findings)
if remaining > 0: if remaining > 0:
lines.append(f"... and {remaining} more CVE(s)") lines.append(f"... and {remaining} more CVE(s)")
if aggregated.reports: if aggregated.reports:
lines += ["", "Reports:"] lines += ["", "Reports:"] + [f"- {p}" for p in aggregated.reports[:5]]
lines += [f"- {path}" for path in aggregated.reports[:10]] if len(aggregated.reports) > 5:
if len(aggregated.reports) > 10: lines.append(f"- ... +{len(aggregated.reports) - 5} more")
lines.append(f"- ... and {len(aggregated.reports) - 10} more")
body = "\n".join(lines) body = "\n".join(lines)
if len(body) > MAX_NOTIFICATION_CHARS: return _truncate_text(body, MAX_NOTIFICATION_CHARS)
return _truncate_text(body, MAX_NOTIFICATION_CHARS)
return body
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -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): 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)" 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 = build_aggregated_body(aggregated, settings.max_findings)
body = _truncate_text(body, MAX_NOTIFICATION_CHARS) 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): if not notify(settings, title, body):
stats.notify_errors += 1 stats.notify_errors += 1
elif aggregated.cves: elif aggregated.cves:
+2 -3
View File
@@ -434,9 +434,9 @@ class ReportTests(unittest.TestCase):
body = scan.build_aggregated_body(aggregated, 15) body = scan.build_aggregated_body(aggregated, 15)
self.assertIn("CVE-1", body) self.assertIn("CVE-1", body)
self.assertIn("containers: web", 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("containers: cache", body)
self.assertIn("image: redis:7", body) self.assertIn("redis:7", body)
class TrivyCommandTests(unittest.TestCase): class TrivyCommandTests(unittest.TestCase):
@@ -569,7 +569,6 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG"
notification = Path(self.apprise_log).read_text() notification = Path(self.apprise_log).read_text()
self.assertIn("CVE-2026-0001", notification) self.assertIn("CVE-2026-0001", notification)
self.assertIn("containers: testphp", notification) self.assertIn("containers: testphp", notification)
self.assertIn("image: testphp-php", notification)
def test_scan_failure_notification_includes_container(self) -> None: def test_scan_failure_notification_includes_container(self) -> None:
containers = [container("a" * 64, "testphp-php", name="testphp")] containers = [container("a" * 64, "testphp-php", name="testphp")]