more strict limit
This commit is contained in:
1 parent
65b9c1fc84
commit
cdd8004a6f
2 files changed
+80
-22
No files matched your search
@@ -709,7 +709,7 @@ def notify(settings: Settings, title: str, body: str) -> bool:
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
log("INFO", f"Sending Apprise notification: {title} (title {len(title)} chars, body {len(body)} chars)")
|
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:
|
if settings.apprise_config:
|
||||||
command += ["--config", settings.apprise_config]
|
command += ["--config", settings.apprise_config]
|
||||||
command += split_words(settings.apprise_urls)
|
command += split_words(settings.apprise_urls)
|
||||||
@@ -772,6 +772,12 @@ def build_body(
|
|||||||
|
|
||||||
|
|
||||||
MAX_NOTIFICATION_CHARS = 2500
|
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:
|
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(
|
def build_aggregated_body(
|
||||||
aggregated: AggregatedReport, max_findings: int
|
aggregated: AggregatedReport, max_findings: int, compact: bool = False
|
||||||
) -> str:
|
) -> 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 = [
|
lines = [
|
||||||
f"Critical: {aggregated.critical} | High: {aggregated.high} | "
|
f"Critical: {aggregated.critical} | High: {aggregated.high} | "
|
||||||
f"Medium: {aggregated.medium} | Low: {aggregated.low}",
|
f"Medium: {aggregated.medium} | Low: {aggregated.low}",
|
||||||
@@ -803,7 +832,7 @@ def build_aggregated_body(
|
|||||||
),
|
),
|
||||||
)
|
)
|
||||||
total = len(cves)
|
total = len(cves)
|
||||||
max_findings = min(max_findings, 20)
|
max_findings = min(max_findings, 15)
|
||||||
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:
|
||||||
@@ -811,27 +840,27 @@ 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[:2]:
|
for pkg, installed, fixed in cve.packages[:1]:
|
||||||
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 "unknown"
|
||||||
lines.append(f"* [{cve.severity}] {cve.cve} - {pkg_text}")
|
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"
|
containers = ", ".join(affected.containers) if affected.containers else "unknown"
|
||||||
lines.append(f" - containers: {containers} ({affected.image})")
|
lines.append(f" - {containers} ({affected.image})")
|
||||||
if len(cve.affected) > 3:
|
if len(cve.affected) > 2:
|
||||||
lines.append(f" - ... +{len(cve.affected) - 3} more")
|
lines.append(f" - +{len(cve.affected) - 2} more")
|
||||||
|
|
||||||
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"... +{remaining} more CVE(s)")
|
||||||
|
|
||||||
if aggregated.reports:
|
if aggregated.reports:
|
||||||
lines += ["", "Reports:"] + [f"- {p}" for p in aggregated.reports[:5]]
|
lines += ["", "Reports:"] + [f"- {p}" for p in aggregated.reports[:3]]
|
||||||
if len(aggregated.reports) > 5:
|
if len(aggregated.reports) > 3:
|
||||||
lines.append(f"- ... +{len(aggregated.reports) - 5} more")
|
lines.append(f"- +{len(aggregated.reports) - 3} more")
|
||||||
|
|
||||||
body = "\n".join(lines)
|
body = "\n".join(lines)
|
||||||
return _truncate_text(body, MAX_NOTIFICATION_CHARS)
|
return _truncate_text(body, MAX_NOTIFICATION_CHARS)
|
||||||
@@ -876,8 +905,11 @@ def scan_image(
|
|||||||
"",
|
"",
|
||||||
]
|
]
|
||||||
body = "\n".join(lines)
|
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}"
|
title = f"Scan failed: {image}{target.notification_context}"
|
||||||
|
if len(title) > 200:
|
||||||
|
title = f"Scan failed: {image}"
|
||||||
if not notify(settings, title, body):
|
if not notify(settings, title, body):
|
||||||
stats.notify_errors += 1
|
stats.notify_errors += 1
|
||||||
return
|
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):
|
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:
|
if len(title) > 200:
|
||||||
title = f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s) across {len(targets)} img"
|
title = f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s)"
|
||||||
body = build_aggregated_body(aggregated, settings.max_findings)
|
discord = uses_discord(settings)
|
||||||
body = _truncate_text(body, MAX_NOTIFICATION_CHARS)
|
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")
|
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
|
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:
|
elif aggregated.cves:
|
||||||
log("INFO", f"{len(aggregated.cves)} CVE(s) tracked but below NOTIFY_MIN_SEVERITY={settings.notify_min_severity}")
|
log("INFO", f"{len(aggregated.cves)} CVE(s) tracked but below NOTIFY_MIN_SEVERITY={settings.notify_min_severity}")
|
||||||
|
|
||||||
|
|||||||
+17
-3
@@ -413,6 +413,20 @@ class ReportTests(unittest.TestCase):
|
|||||||
body = scan.build_aggregated_body(aggregated, 200)
|
body = scan.build_aggregated_body(aggregated, 200)
|
||||||
self.assertLessEqual(len(body), scan.MAX_NOTIFICATION_CHARS)
|
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:
|
def test_build_aggregated_body_groups_by_cve(self) -> None:
|
||||||
aggregated = scan.AggregatedReport()
|
aggregated = scan.AggregatedReport()
|
||||||
vuln1 = {
|
vuln1 = {
|
||||||
@@ -433,9 +447,9 @@ class ReportTests(unittest.TestCase):
|
|||||||
aggregated.add_vulnerability(vuln2, "redis:7", ["cache"])
|
aggregated.add_vulnerability(vuln2, "redis:7", ["cache"])
|
||||||
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("web", body)
|
||||||
self.assertIn("nginx:1.27", body)
|
self.assertIn("nginx:1.27", body)
|
||||||
self.assertIn("containers: cache", body)
|
self.assertIn("cache", body)
|
||||||
self.assertIn("redis:7", body)
|
self.assertIn("redis:7", body)
|
||||||
|
|
||||||
|
|
||||||
@@ -568,7 +582,7 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG"
|
|||||||
self.assertEqual(code, 1)
|
self.assertEqual(code, 1)
|
||||||
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("testphp", 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")]
|
||||||
|
|||||||
Reference in new issue
Block a user