From 18a1d0e863023eb270f1f48515b78ec033e5d2af Mon Sep 17 00:00:00 2001 From: David Duran Rosich Date: Sat, 24 Oct 2026 21:59:12 +0200 Subject: [PATCH] add IMPORTANT_CONTAINERS --- .env.example | 6 + README.md | 54 +++- docker-compose.yml | 2 + komodo/vuln-scan-files-on-host.toml | 2 + komodo/vuln-scan.toml | 2 + scan.py | 450 ++++++++++++++++++++++++++-- tests/test_scan.py | 112 +++++++ 7 files changed, 588 insertions(+), 40 deletions(-) diff --git a/.env.example b/.env.example index a77125e..5471004 100644 --- a/.env.example +++ b/.env.example @@ -15,6 +15,12 @@ # Also scan the scanner's own image when auto-discovering (default: false). # INCLUDE_SELF=false +# Containers that face the internet. Each one gets its own notification with +# every CVE at/above NOTIFY_MIN_SEVERITY, split into fixable and unfixable. +# All other containers are summarised in one notification. Space or comma +# separated. Leave empty to keep the single aggregated notification. +# IMPORTANT_CONTAINERS=traefik web-frontend + # One or more Apprise URLs, space separated. Quote the value. # Examples: # mailto://user:app-password@gmail.com diff --git a/README.md b/README.md index 7fbe19d..3e0c01e 100644 --- a/README.md +++ b/README.md @@ -25,8 +25,12 @@ host only needs Docker. JSON reports are kept on disk for auditing. alert. - `FAIL_ON` (default `HIGH`) – when the script should exit with code `1` (useful for CI). -3. If any CVE reaches `NOTIFY_MIN_SEVERITY`, one notification is sent listing - each CVE with the affected packages and the containers/images that use them. +3. If any CVE reaches `NOTIFY_MIN_SEVERITY`, notifications are sent. When + `IMPORTANT_CONTAINERS` is set, each named container gets its own complete + CVE list (split into fixable and unfixable), missing important containers + are reported, and every other container is summarised in one notification. + Without `IMPORTANT_CONTAINERS`, a single aggregated notification lists each + CVE with the affected packages, containers, and images. ## Quick start @@ -67,10 +71,15 @@ For the last one the scanner asks the Docker Engine API on the Unix socket for listing order. Stopped containers are not included. The scanner's own container image is skipped so a scheduled run does not report on itself; set `INCLUDE_SELF=true` to include it. Scan logs still name the running container(s) that use each image (e.g. -`Scanning nginx:1.27 (used by container web)`). Notifications are aggregated -by CVE after all images are scanned: a single report lists each CVE with the -containers and images it affects. Scan failures still notify per failed image -with container context. Explicitly passed images omit container context. +`Scanning nginx:1.27 (used by container web)`). Without `IMPORTANT_CONTAINERS`, +notifications are aggregated by CVE after all images are scanned: a single +report lists each CVE with the containers and images it affects. With +`IMPORTANT_CONTAINERS`, each named container gets its own full CVE list and the +remaining containers are summarised together (see +[Important containers](#important-containers)). Scan failures still notify per +failed image with container context. Explicitly passed images omit container +context unless `IMPORTANT_CONTAINERS` is set, in which case running containers +are looked up to match them to the scanned images. The socket is looked up in this order: @@ -141,6 +150,7 @@ All configuration is done through environment variables. | `DOCKER_SOCK` | auto | Unix socket of the Docker daemon (or Podman). Also the Compose mount source. | | `DOCKER_HOST` | – | Standard Docker variable; `unix://` paths are used for discovery. | | `INCLUDE_SELF` | `false` | Also scan the scanner's own image when auto-discovering. | +| `IMPORTANT_CONTAINERS` | – | Container names (space/comma separated) that get a full per-container CVE notification. Others are summarised together. | | `APPRISE_URLS` | – | Space separated Apprise URLs (quote the value). | | `APPRISE_CONFIG` | – | Alternative: path to an Apprise YAML config. | | `APPRISE_TITLE_PREFIX` | `Vulnerability scan` | Prefix of notification titles. | @@ -153,7 +163,7 @@ All configuration is done through environment variables. | `REPORT_DIR` | `/reports` | Where JSON reports are written. | | `TRIVY_CACHE_DIR` | `/var/lib/trivy` | Trivy DB cache (mount a volume to avoid re-downloading). | | `TRIVY_TIMEOUT` | `15m` | Passed to `trivy --timeout`. | -| `MAX_FINDINGS_IN_BODY` | `15` | How many findings are listed per notification. | +| `MAX_FINDINGS_IN_BODY` | `15` | How many findings the single aggregated notification lists. Important-container notifications are not capped. | | `LOG_LEVEL` | `INFO` | `DEBUG`, `INFO`, `WARN` or `ERROR`. | Images can be passed as arguments, via `IMAGES`, or via `IMAGES_FILE`; the @@ -233,7 +243,35 @@ docker run --rm \ docker-vuln-scan:latest nginx:1.27-alpine ``` -## Exit codes +## Important containers + +Set `IMPORTANT_CONTAINERS` to the containers that are reachable from outside +(your public entry points). Names are matched against running containers, so +leave `IMAGES` empty unless you also want to scan specific images explicitly. + +```dotenv +IMPORTANT_CONTAINERS=traefik web-frontend +``` + +When the list is set, notifications are split into three groups: + +1. **One notification per important container**, titled + `Vulnerability scan: ()`. It lists every CVE at or above + `NOTIFY_MIN_SEVERITY`, with a total split into fixable and unfixable. +2. **One notification for important containers that are missing** – either not + running, or running an image that was not part of the scan. +3. **One summary notification for every other container**, listing the image, + container, total CVE count, and the fixable/unfixable split. + +A container can appear in only the important list or the summary, never both. +If an important container shares its image with other containers, the others +still appear in the summary. Important-container lists are never truncated: if +a list is longer than the Apprise/Discord message limit it is sent in numbered +parts. `FAIL_ON` and the exit codes are unchanged. + +Without `IMPORTANT_CONTAINERS`, the previous single aggregated notification is +sent. + | Code | Meaning | | --- | --- | diff --git a/docker-compose.yml b/docker-compose.yml index 8f38e39..a34e76a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -19,6 +19,8 @@ services: TRIVY_TIMEOUT: ${TRIVY_TIMEOUT:-15m} # Optional: also scan the scanner's own image (default: false). INCLUDE_SELF: ${INCLUDE_SELF:-false} + # Optional: containers that get a full per-container CVE notification. + IMPORTANT_CONTAINERS: ${IMPORTANT_CONTAINERS:-} # Optional: scan a list of images instead of IMAGES (mount the file). IMAGES_FILE: ${IMAGES_FILE:-} # Path of the Docker socket *inside* the container (same as on the host, diff --git a/komodo/vuln-scan-files-on-host.toml b/komodo/vuln-scan-files-on-host.toml index 6c88b0c..624ab86 100644 --- a/komodo/vuln-scan-files-on-host.toml +++ b/komodo/vuln-scan-files-on-host.toml @@ -38,6 +38,8 @@ APPRISE_TITLE_PREFIX = Vulnerability scan NOTIFY_MIN_SEVERITY = HIGH FAIL_ON = HIGH IGNORE_UNFIXED = false +# Containers with outside access that get a full per-container notification. +# IMPORTANT_CONTAINERS = traefik web-frontend MAX_FINDINGS_IN_BODY = 15 """ diff --git a/komodo/vuln-scan.toml b/komodo/vuln-scan.toml index 12e2427..3833b26 100644 --- a/komodo/vuln-scan.toml +++ b/komodo/vuln-scan.toml @@ -38,6 +38,8 @@ APPRISE_TITLE_PREFIX = Vulnerability scan NOTIFY_MIN_SEVERITY = HIGH FAIL_ON = HIGH IGNORE_UNFIXED = false +# Containers with outside access that get a full per-container notification. +# IMPORTANT_CONTAINERS = traefik web-frontend MAX_FINDINGS_IN_BODY = 15 """ diff --git a/scan.py b/scan.py index 4351118..eeccc66 100755 --- a/scan.py +++ b/scan.py @@ -87,6 +87,11 @@ def split_words(raw: str) -> list[str]: return [word for word in raw.replace("\n", " ").split() if word] +def split_list(raw: str) -> list[str]: + """Split a whitespace/comma separated list, dropping empty and duplicate entries.""" + return dedupe(split_words(raw.replace(",", " "))) + + def dedupe(items: list[str]) -> list[str]: """Drop duplicates while keeping the original order.""" seen: set[str] = set() @@ -370,6 +375,7 @@ class Settings: trivy_cache_dir: str = "/var/lib/trivy" trivy_timeout: str = "15m" max_findings: int = 15 + important_containers: list[str] = field(default_factory=list) include_self: bool = False docker_socket: str | None = None @@ -400,6 +406,7 @@ class Settings: trivy_cache_dir=os.environ.get("TRIVY_CACHE_DIR", "/var/lib/trivy"), trivy_timeout=os.environ.get("TRIVY_TIMEOUT", "15m"), max_findings=env_int("MAX_FINDINGS_IN_BODY", 15), + important_containers=split_list(os.environ.get("IMPORTANT_CONTAINERS", "")), include_self=env_bool("INCLUDE_SELF", False), ) settings.validate() @@ -503,6 +510,7 @@ class ScanResult: medium: int = 0 low: int = 0 findings: list[dict] = field(default_factory=list) + all_findings: list[dict] = field(default_factory=list) @property def counts(self) -> dict[str, int]: @@ -536,6 +544,16 @@ class CveAggregation: fixable: bool = False +@dataclass +class CveEntry: + """A single CVE in one image, with every package that reports it.""" + + cve: str + severity: str + fixable: bool = False + packages: list[tuple[str, str, str]] = field(default_factory=list) + + @dataclass class AggregatedReport: critical: int = 0 @@ -544,6 +562,7 @@ class AggregatedReport: low: int = 0 cves: dict[str, CveAggregation] = field(default_factory=dict) reports: list[str] = field(default_factory=list) + results: dict[str, ScanResult] = field(default_factory=dict) @property def counts(self) -> dict[str, int]: @@ -632,17 +651,80 @@ def summarise(report: dict) -> ScanResult: result.medium += 1 elif severity == "LOW": result.low += 1 + if severity in REPORTED_SEVERITIES: + result.all_findings.append(vuln) if severity in ("CRITICAL", "HIGH"): result.findings.append(vuln) - result.findings.sort( - key=lambda v: ( - 0 if str(v.get("Severity", "")).upper() == "CRITICAL" else 1, - str(v.get("VulnerabilityID", "")), + + def sort_key(vuln: dict) -> tuple[int, str]: + return ( + 0 if str(vuln.get("Severity", "")).upper() == "CRITICAL" else 1, + str(vuln.get("VulnerabilityID", "")), ) - ) + + result.findings.sort(key=sort_key) + result.all_findings.sort(key=sort_key) return result +def findings_at_rank(result: ScanResult, rank: int) -> list[dict]: + """Every finding at or above the requested severity rank.""" + if rank <= 0: + return [] + return [ + vuln + for vuln in result.all_findings + if severity_rank(str(vuln.get("Severity", ""))) >= rank + ] + + +def group_cves(findings: list[dict]) -> list[CveEntry]: + """Collapse package-level findings into one entry per CVE.""" + entries: dict[str, CveEntry] = {} + for vuln in findings: + cve = str(vuln.get("VulnerabilityID", "")) + if not cve: + continue + severity = str(vuln.get("Severity", "")).upper() + entry = entries.get(cve) + if entry is None: + entry = CveEntry(cve=cve, severity=severity) + entries[cve] = entry + elif severity_rank(severity) > severity_rank(entry.severity): + entry.severity = severity + + pkg = str(vuln.get("PkgName", "")) + installed = str(vuln.get("InstalledVersion", "")) + fixed = str(vuln.get("FixedVersion") or "") + if fixed: + entry.fixable = True + package = (pkg, installed, fixed) + if package not in entry.packages: + entry.packages.append(package) + + return sorted( + entries.values(), + key=lambda item: (-severity_rank(item.severity), item.cve), + ) + + +def cve_entries(result: ScanResult, rank: int) -> list[CveEntry]: + return group_cves(findings_at_rank(result, rank)) + + +def split_counts(entries: list[CveEntry]) -> tuple[int, int, int]: + """Return (total, fixable, unfixable) for a list of CVEs.""" + fixable = sum(1 for entry in entries if entry.fixable) + return len(entries), fixable, len(entries) - fixable + + +def severity_counts(entries: list[CveEntry]) -> dict[str, int]: + counts = {severity: 0 for severity in REPORTED_SEVERITIES} + for entry in entries: + counts[entry.severity] += 1 + return counts + + def trivy_environment(settings: Settings) -> dict[str, str]: env = os.environ.copy() # Trivy needs DOCKER_HOST to read images that only exist in the daemon. @@ -866,6 +948,85 @@ def build_aggregated_body( return _truncate_text(body, MAX_NOTIFICATION_CHARS) +def notification_limit(settings: Settings) -> int: + return DISCORD_MAX_NOTIFICATION_CHARS if uses_discord(settings) else MAX_NOTIFICATION_CHARS + + +def package_text(entry: CveEntry) -> str: + parts = [] + for pkg, installed, fixed in entry.packages: + if pkg: + parts.append(f"{pkg} {installed} -> {fixed or 'no fix available'}") + elif installed or fixed: + parts.append(f"{installed} -> {fixed or 'no fix available'}") + return "; ".join(parts) if parts else "unknown" + + +def build_important_body( + container: str, + target: ScanTarget, + result: ScanResult, + report_path: str, + min_severity: str, +) -> str: + """Full CVE list for one important container, with fixable/unfixable counts.""" + entries = cve_entries(result, severity_rank(min_severity)) + total, fixable, unfixable = split_counts(entries) + counts = severity_counts(entries) + + lines = [ + f"Image: {target.image}", + f"Container: {container}", + " | ".join(f"{name.title()}: {counts[name]}" for name in REPORTED_SEVERITIES), + f"Total: {total} (fixable: {fixable}, unfixable: {unfixable})", + "", + "CVEs:", + ] + for entry in entries: + lines.append(f"* [{entry.severity}] {entry.cve} - {package_text(entry)}") + if not entries: + lines.append("None at or above the notification threshold.") + + if report_path: + lines += ["", f"Report: {report_path}"] + return "\n".join(lines) + + +def build_other_containers_body( + rows: list[tuple[str, str, int, int, int]], +) -> str: + """Summarise containers that are not marked important. + + Each row is ``(image, container, total, fixable, unfixable)``. + """ + lines = ["Other containers with findings:", ""] + total_all = 0 + for image, container, total, fixable, unfixable in rows: + total_all += total + label = f"{image} [{container}]" if container else image + lines.append( + f"* {label}: {total} total " + f"({fixable} fixable, {unfixable} unfixable)" + ) + lines += ["", f"Total: {total_all} CVE(s) in {len(rows)} container(s)"] + return "\n".join(lines) + + +def build_missing_body( + missing: list[str], not_scanned: list[str] | None = None +) -> str: + lines = [] + if missing: + lines.append("Important containers are not running:") + lines += [f"* {name}" for name in missing] + if not_scanned: + if lines: + lines.append("") + lines.append("Important containers are running but their image was not scanned:") + lines += [f"* {name}" for name in not_scanned] + return "\n".join(lines) + + # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- @@ -923,6 +1084,7 @@ def scan_image( return result = summarise(report) + aggregated.results[image] = result log( "INFO", f"{image}{context}: {result.critical} critical, {result.high} high, " @@ -952,10 +1114,10 @@ def scan_image( def usage() -> str: return f"""Usage: {PROG} [IMAGE ...] -Scan each Docker image with Trivy. Findings are aggregated by CVE across all -images and a single notification is sent when any reach -NOTIFY_MIN_SEVERITY (default: HIGH). Each CVE lists the containers and images -that are affected. +Scan each Docker image with Trivy. When any finding reaches +NOTIFY_MIN_SEVERITY (default: HIGH), a notification is sent. With +IMPORTANT_CONTAINERS, each named container gets a full CVE list and the rest +are summarised together; otherwise one aggregated notification is sent. Images can be passed as arguments or with IMAGES (space or newline separated) or IMAGES_FILE (one image per line, '#' starts a comment). @@ -987,6 +1149,9 @@ Environment: TRIVY_CACHE_DIR Trivy database/cache directory (default: /var/lib/trivy). TRIVY_TIMEOUT Scan timeout passed to Trivy (default: 15m). MAX_FINDINGS_IN_BODY Findings listed in a notification body (default: 15). + IMPORTANT_CONTAINERS Container names (space or comma separated) that get a + full per-container CVE notification. Every other + container is summarised in one notification. IMAGES Images to scan, space or newline separated. IMAGES_FILE File containing images to scan (one per line). @@ -994,6 +1159,246 @@ Exit codes: 0 = clean, 1 = vulnerability policy failed, 2 = scan error. """ +def _split_body(body: str, max_chars: int) -> list[str]: + """Split a body on line boundaries so every part fits the channel limit.""" + if max_chars <= 0 or len(body) <= max_chars: + return [body] + parts: list[str] = [] + current: list[str] = [] + size = 0 + for line in body.splitlines(): + while len(line) > max_chars: + if current: + parts.append("\n".join(current)) + current = [] + size = 0 + parts.append(line[:max_chars]) + line = line[max_chars:] + line_size = len(line) + 1 + if current and size + line_size > max_chars: + parts.append("\n".join(current)) + current = [] + size = 0 + current.append(line) + size += line_size + if current: + parts.append("\n".join(current)) + return parts + + +def send_notification( + settings: Settings, + stats: RunStats, + title: str, + body: str, + fallback: str | None = None, +) -> bool: + if len(title) > 200: + title = title[:197] + "..." + body = _truncate_text(body, notification_limit(settings)) + if notify(settings, title, body): + return True + if fallback is not None: + log("WARN", "Retrying notification with minimal summary") + fallback = _truncate_text(fallback, notification_limit(settings)) + if notify(settings, title, fallback): + return True + stats.notify_errors += 1 + return False + + +def send_notification_parts( + settings: Settings, stats: RunStats, title: str, body: str +) -> bool: + """Send a body in as many messages as needed, keeping every line.""" + if len(title) > 200: + title = title[:197] + "..." + parts = _split_body(body, notification_limit(settings)) + if len(parts) > 1: + log("INFO", f"Sending {title} in {len(parts)} parts") + ok = True + for index, part in enumerate(parts, 1): + part_title = title if len(parts) == 1 else f"{title} [{index}/{len(parts)}]" + if not notify(settings, part_title, part): + ok = False + if not ok: + stats.notify_errors += 1 + return ok + + +def send_aggregated_notification( + settings: Settings, + targets: list[ScanTarget], + aggregated: AggregatedReport, + stats: RunStats, +) -> None: + if not aggregated.cves or not threshold_hits_from_counts( + aggregated.counts, settings.notify_rank + ): + if aggregated.cves: + log( + "INFO", + f"{len(aggregated.cves)} CVE(s) tracked but below " + f"NOTIFY_MIN_SEVERITY={settings.notify_min_severity}", + ) + return + + title = ( + f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s) " + f"across {len(targets)} image(s)" + ) + 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, " + f"body {len(body)} chars, {len(aggregated.cves)} CVEs", + ) + minimal = ( + f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s) " + f"across {len(targets)} image(s)\n" + f"C:{aggregated.critical} H:{aggregated.high} " + f"M:{aggregated.medium} L:{aggregated.low}\n" + f"Full details in reports in {settings.report_dir}" + ) + send_notification(settings, stats, title, body, fallback=minimal) + + +def send_notifications( + settings: Settings, + targets: list[ScanTarget], + aggregated: AggregatedReport, + stats: RunStats, + container_images: dict[str, str], + discovery_known: bool, +) -> None: + if settings.notify_rank <= 0: + return + + if not settings.important_containers: + send_aggregated_notification(settings, targets, aggregated, stats) + return + + important = set(settings.important_containers) + scanned = {target.image: target for target in targets} + missing: list[str] = [] + not_scanned: list[str] = [] + + # One full notification per important container (container + image). + for name in settings.important_containers: + image = container_images.get(name) + if image is None: + if discovery_known: + missing.append(name) + continue + target = scanned.get(image) + if target is None: + not_scanned.append(name) + continue + result = aggregated.results.get(image) + if result is None: + continue # scan failure; already reported by scan_image + + entries = cve_entries(result, settings.notify_rank) + if not entries: + log( + "DEBUG", + f"Important container {name} ({image}) has no findings at or above " + f"{settings.notify_min_severity}", + ) + continue + report_path = os.path.join(settings.report_dir, f"{report_slug(image)}.json") + body = build_important_body( + name, target, result, report_path, settings.notify_min_severity + ) + title = f"{settings.apprise_title_prefix}: {name} ({image})" + log( + "DEBUG", + f"Preparing important-container notification for {name}: " + f"{len(entries)} CVE(s)", + ) + send_notification_parts(settings, stats, title, body) + + if missing or not_scanned: + send_notification( + settings, + stats, + f"{settings.apprise_title_prefix}: important containers", + build_missing_body(missing, not_scanned), + ) + + # One summary notification for every other container. + rows: list[tuple[str, str, int, int, int]] = [] + for target in targets: + result = aggregated.results.get(target.image) + if result is None: + continue + entries = cve_entries(result, settings.notify_rank) + if not entries: + continue + total, fixable, unfixable = split_counts(entries) + for container in target.containers or [""]: + if container and container in important: + continue + rows.append((target.image, container, total, fixable, unfixable)) + + if rows: + send_notification( + settings, + stats, + f"{settings.apprise_title_prefix}: other containers", + build_other_containers_body(rows), + ) + + +def resolve_container_images( + targets: list[ScanTarget], settings: Settings +) -> tuple[dict[str, str], bool]: + """Map container names to images and enrich explicit targets. + + Auto-discovery already attaches containers to targets. When images are + given explicitly, discovery is still used (only when important containers + are configured) so those names can be matched. The boolean reports whether + discovery actually worked, so a failed lookup is not mistaken for "container + not running". + """ + container_images = { + container: target.image + for target in targets + for container in target.containers + } + if not settings.important_containers: + return container_images, True + + if settings.explicit_images: + discovered, source = discover_containers(include_self=settings.include_self) + if source is None: + log( + "WARN", + "IMPORTANT_CONTAINERS is set but running containers could not be " + "discovered; important-container notifications are disabled", + ) + return {}, False + container_images = { + name: image for image, names in discovered.items() for name in names + } + for target in targets: + target.containers = [ + name + for name, image in container_images.items() + if image == target.image + ] + return container_images, True + + return container_images, True + + def main(argv: list[str] | None = None) -> int: argv = list(sys.argv[1:] if argv is None else argv) @@ -1019,6 +1424,7 @@ def main(argv: list[str] | None = None) -> int: targets = collect_targets(settings) if not targets: return 0 + container_images, discovery_known = resolve_container_images(targets, settings) os.makedirs(settings.report_dir, exist_ok=True) os.makedirs(settings.trivy_cache_dir, exist_ok=True) @@ -1028,29 +1434,9 @@ def main(argv: list[str] | None = None) -> int: for target in targets: scan_image(target, settings, stats, aggregated) - 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) > 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): - 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}") + send_notifications( + settings, targets, aggregated, stats, container_images, discovery_known + ) log( "INFO", diff --git a/tests/test_scan.py b/tests/test_scan.py index 0dffe65..f708b98 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -298,6 +298,7 @@ class SettingsTests(unittest.TestCase): "IGNORE_UNFIXED": "true", "MAX_FINDINGS_IN_BODY": "3", "TRIVY_VEX": "/vex/app.json repo", + "IMPORTANT_CONTAINERS": "traefik, web-frontend\nworker", } with mock.patch.dict(os.environ, env, clear=False): settings = scan.Settings.from_env([]) @@ -306,6 +307,9 @@ class SettingsTests(unittest.TestCase): self.assertTrue(settings.ignore_unfixed) self.assertEqual(settings.max_findings, 3) self.assertEqual(settings.trivy_vex, ["/vex/app.json", "repo"]) + self.assertEqual( + settings.important_containers, ["traefik", "web-frontend", "worker"] + ) def test_settings_rejects_unknown_severity(self) -> None: with mock.patch.dict(os.environ, {"NOTIFY_MIN_SEVERITY": "SEVERE"}, clear=False): @@ -452,6 +456,68 @@ class ReportTests(unittest.TestCase): self.assertIn("cache", body) self.assertIn("redis:7", body) + def test_cve_entries_split_fixable_and_unfixable(self) -> None: + entries = scan.cve_entries(scan.summarise(REPORT), scan.severity_rank("HIGH")) + self.assertEqual(scan.split_counts(entries), (3, 2, 1)) + self.assertEqual([entry.cve for entry in entries], ["CVE-1", "CVE-2", "CVE-5"]) + self.assertEqual(scan.severity_counts(entries), { + "CRITICAL": 1, + "HIGH": 2, + "MEDIUM": 0, + "LOW": 0, + }) + + def test_build_important_body_lists_every_cve(self) -> None: + result = scan.summarise(REPORT) + target = scan.ScanTarget(image="nginx:1.27", containers=["web"]) + body = scan.build_important_body( + "web", target, result, "/reports/x.json", "HIGH" + ) + self.assertIn("Image: nginx:1.27", body) + self.assertIn("Container: web", body) + self.assertIn("Total: 3 (fixable: 2, unfixable: 1)", body) + self.assertIn("CVE-1", body) + self.assertIn("CVE-2", body) + self.assertIn("CVE-5", body) + self.assertIn("no fix available", body) + self.assertIn("/reports/x.json", body) + + def test_build_important_body_respects_notification_threshold(self) -> None: + result = scan.summarise(REPORT) + target = scan.ScanTarget(image="nginx:1.27", containers=["web"]) + body = scan.build_important_body( + "web", target, result, "/reports/x.json", "CRITICAL" + ) + self.assertIn("CVE-1", body) + self.assertNotIn("CVE-2", body) + self.assertNotIn("CVE-5", body) + self.assertIn("Total: 1 (fixable: 1, unfixable: 0)", body) + + def test_build_other_containers_body_has_counts_per_container(self) -> None: + body = scan.build_other_containers_body( + [ + ("nginx:1.27", "web-worker", 3, 2, 1), + ("redis:7", "cache", 4, 0, 4), + ] + ) + self.assertIn("* nginx:1.27 [web-worker]: 3 total (2 fixable, 1 unfixable)", body) + self.assertIn("* redis:7 [cache]: 4 total (0 fixable, 4 unfixable)", body) + self.assertIn("Total: 7 CVE(s) in 2 container(s)", body) + + def test_build_missing_body_separates_missing_and_unscanned(self) -> None: + body = scan.build_missing_body(["gone"], ["web"]) + self.assertIn("not running", body) + self.assertIn("* gone", body) + self.assertIn("image was not scanned", body) + self.assertIn("* web", body) + + def test_split_body_keeps_every_line(self) -> None: + body = "\n".join(f"CVE-{index}" for index in range(10)) + parts = scan._split_body(body, 20) + self.assertGreater(len(parts), 1) + self.assertTrue(all(len(part) <= 20 for part in parts)) + self.assertEqual("\n".join(parts), body) + class TrivyCommandTests(unittest.TestCase): def test_run_trivy_passes_vex_sources_in_order(self) -> None: @@ -584,6 +650,52 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG" self.assertIn("CVE-2026-0001", notification) self.assertIn("testphp", notification) + def test_important_container_gets_full_list_and_others_get_summary(self) -> None: + containers = [ + container("a" * 64, "testphp-php", name="web"), + container("b" * 64, "redis:7", name="cache"), + ] + with FakeDocker(containers) as docker: + code = self.run_main( + [], docker.socket_path, IMPORTANT_CONTAINERS="web" + ) + + self.assertEqual(code, 1) + notification = Path(self.apprise_log).read_text() + self.assertIn("Vulnerability scan: web (testphp-php)", notification) + self.assertIn("Container: web", notification) + self.assertIn("CVE-2026-0001", notification) + self.assertIn("Vulnerability scan: other containers", notification) + self.assertIn("redis:7 [cache]", notification) + self.assertIn("(1 fixable, 0 unfixable)", notification) + # The important container must not also appear in the summary. + self.assertNotIn("testphp-php [web]", notification) + + def test_missing_important_container_is_reported(self) -> None: + containers = [container("a" * 64, "testphp-php", name="web")] + with FakeDocker(containers) as docker: + code = self.run_main( + [], docker.socket_path, IMPORTANT_CONTAINERS="web stale" + ) + + self.assertEqual(code, 1) + notification = Path(self.apprise_log).read_text() + self.assertIn("Vulnerability scan: important containers", notification) + self.assertIn("Important containers are not running:", notification) + self.assertIn("* stale", notification) + + def test_important_container_with_unscanned_image_is_reported(self) -> None: + containers = [container("a" * 64, "other:1", name="web")] + with FakeDocker(containers) as docker: + code = self.run_main( + ["nginx:1.27"], docker.socket_path, IMPORTANT_CONTAINERS="web" + ) + + self.assertEqual(code, 1) + notification = Path(self.apprise_log).read_text() + self.assertIn("image was not scanned", notification) + self.assertIn("* web", notification) + def test_scan_failure_notification_includes_container(self) -> None: containers = [container("a" * 64, "testphp-php", name="testphp")] with FakeDocker(containers) as docker: