truncate notification
This commit is contained in:
1 parent
9d0fd2ba44
commit
014a1d5563
2 files changed
+47
-6
No files matched your search
@@ -766,6 +766,20 @@ def build_body(
|
|||||||
return "\n".join(lines)
|
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(
|
def build_aggregated_body(
|
||||||
aggregated: AggregatedReport, max_findings: int
|
aggregated: AggregatedReport, max_findings: int
|
||||||
) -> str:
|
) -> str:
|
||||||
@@ -797,18 +811,29 @@ def build_aggregated_body(
|
|||||||
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(
|
lines.append(f"* [{cve.severity}] {cve.cve} - {pkg_text}")
|
||||||
f"* [{cve.severity}] {cve.cve} - {pkg_text}"
|
for affected in cve.affected[:5]:
|
||||||
)
|
|
||||||
for affected in cve.affected:
|
|
||||||
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} (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:
|
if aggregated.reports:
|
||||||
lines += ["", "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 = "\n".join(lines)
|
||||||
|
body = _truncate_text(body, MAX_NOTIFICATION_CHARS)
|
||||||
title = f"Scan failed: {image}{target.notification_context}"
|
title = f"Scan failed: {image}{target.notification_context}"
|
||||||
if not notify(settings, title, body):
|
if not notify(settings, title, body):
|
||||||
stats.notify_errors += 1
|
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):
|
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)"
|
||||||
body = build_aggregated_body(aggregated, settings.max_findings)
|
body = build_aggregated_body(aggregated, settings.max_findings)
|
||||||
|
body = _truncate_text(body, MAX_NOTIFICATION_CHARS)
|
||||||
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:
|
||||||
|
|||||||
@@ -399,6 +399,20 @@ class ReportTests(unittest.TestCase):
|
|||||||
self.assertIn("Image: nginx:1.27", body)
|
self.assertIn("Image: nginx:1.27", body)
|
||||||
self.assertIn("Containers: web, web-worker", 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:
|
def test_build_aggregated_body_groups_by_cve(self) -> None:
|
||||||
aggregated = scan.AggregatedReport()
|
aggregated = scan.AggregatedReport()
|
||||||
vuln1 = {
|
vuln1 = {
|
||||||
|
|||||||
Reference in new issue
Block a user