#!/usr/bin/env python3 """Scan Docker images with Trivy and alert through Apprise. Images to scan come from (in order of precedence): 1. command line arguments, 2. the IMAGES environment variable, 3. the IMAGES_FILE environment variable, 4. the images used by running containers (via the Docker Engine API). Exit codes: 0 = clean, 1 = vulnerability policy failed, 2 = scan error. """ from __future__ import annotations import http.client import json import glob import os import re import shutil import socket import stat import subprocess import sys from dataclasses import dataclass, field from datetime import datetime, timezone PROG = os.path.basename(sys.argv[0]) SEVERITY_RANK = {"NONE": 0, "LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4} REPORTED_SEVERITIES = ("CRITICAL", "HIGH", "MEDIUM", "LOW") LOG_RANK = {"DEBUG": 10, "INFO": 20, "WARN": 30, "ERROR": 40} MIN_LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").strip().upper() if MIN_LOG_LEVEL not in LOG_RANK: MIN_LOG_LEVEL = "INFO" # --------------------------------------------------------------------------- # Logging and small helpers # --------------------------------------------------------------------------- def log(level: str, message: str) -> None: if LOG_RANK.get(level, 20) < LOG_RANK[MIN_LOG_LEVEL]: return timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") print(f"{timestamp} [{level:<5}] {message}", file=sys.stderr, flush=True) def die(message: str) -> None: log("ERROR", message) raise SystemExit(2) def severity_rank(severity: str) -> int: try: return SEVERITY_RANK[severity.upper()] except KeyError: raise ValueError(f"invalid severity {severity!r}") from None def env_bool(name: str, default: bool) -> bool: raw = os.environ.get(name) if raw is None or not raw.strip(): return default value = raw.strip().lower() if value in {"1", "true", "yes", "on"}: return True if value in {"0", "false", "no", "off"}: return False raise ValueError(f"{name} must be a boolean, got {raw!r}") def env_int(name: str, default: int) -> int: raw = os.environ.get(name) if raw is None or not raw.strip(): return default try: return int(raw.strip()) except ValueError: raise ValueError(f"{name} must be an integer, got {raw!r}") from None def split_words(raw: str) -> list[str]: """Split on whitespace/newlines, dropping empty entries.""" 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() unique: list[str] = [] for item in items: if item not in seen: seen.add(item) unique.append(item) return unique def report_slug(image: str) -> str: return "".join(c if c.isalnum() or c in "_.-" else "_" for c in image) _DIGEST_RE = re.compile(r"(sha[0-9]+:)([0-9a-fA-F]{7,})") def short_image(image: str) -> str: """Shorten image digest hashes to six characters for display.""" return _DIGEST_RE.sub( lambda match: f"{match.group(1)}{match.group(2)[:6]}", image ) # --------------------------------------------------------------------------- # Docker Engine API discovery # --------------------------------------------------------------------------- class DockerError(RuntimeError): pass class _UnixHTTPConnection(http.client.HTTPConnection): """Minimal HTTP client that talks over a Unix domain socket.""" def __init__(self, socket_path: str, timeout: float = 5.0) -> None: super().__init__("localhost", timeout=timeout) self._socket_path = socket_path def connect(self) -> None: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) try: sock.connect(self._socket_path) except OSError: sock.close() raise self.sock = sock def docker_api_get(socket_path: str, path: str) -> object: connection = _UnixHTTPConnection(socket_path) try: connection.request("GET", path, headers={"Host": "docker"}) response = connection.getresponse() body = response.read() if response.status >= 400: raise DockerError(f"GET {path} returned HTTP {response.status}: {body[:200]!r}") if not body: return {} try: return json.loads(body) except json.JSONDecodeError as exc: raise DockerError(f"GET {path} returned invalid JSON: {exc}") from None except OSError as exc: raise DockerError(f"cannot talk to Docker at {socket_path}: {exc}") from None finally: connection.close() def candidate_sockets() -> list[str]: """Unix sockets that may host the Docker (or Podman) API, in priority order.""" candidates: list[str] = [] configured = os.environ.get("DOCKER_SOCK", "").strip() if configured: candidates.append(configured) docker_host = os.environ.get("DOCKER_HOST", "").strip() if docker_host.startswith("unix://"): candidates.append(docker_host[len("unix://") :]) candidates += ["/var/run/docker.sock", "/run/docker.sock"] user_id = os.getuid() xdg_runtime_dir = os.environ.get("XDG_RUNTIME_DIR", "").strip() if xdg_runtime_dir: candidates.append(os.path.join(xdg_runtime_dir, "docker.sock")) # Rootless Docker / Podman put the socket under the user runtime directory. candidates += [ f"/run/user/{user_id}/docker.sock", f"/run/user/{user_id}/podman/podman.sock", "/run/podman/podman.sock", os.path.expanduser("~/.docker/run/docker.sock"), ] # In a container the runtime directory belongs to another uid, so also # consider sockets of every user (only useful if /run/user is mounted). candidates += sorted(glob.glob("/run/user/*/docker.sock")) candidates += sorted(glob.glob("/run/user/*/podman/podman.sock")) return dedupe([c for c in candidates if c]) def find_docker_socket() -> str | None: """Return the first candidate path that is a Unix socket.""" missing: list[str] = [] for path in candidate_sockets(): try: mode = os.stat(path).st_mode except OSError: missing.append(path) continue if stat.S_ISSOCK(mode): return path if stat.S_ISDIR(mode): log("DEBUG", f"{path} is a directory, not a Unix socket") if path == "/var/run/docker.sock": log( "WARN", f"{path} is a directory, not a Unix socket - the host socket was not " "mounted. Set DOCKER_SOCK to the real socket path " "(e.g. /run/user//docker.sock for rootless Docker).", ) else: log("DEBUG", f"{path} exists but is not a Unix socket") if missing: log("DEBUG", "Docker socket not found; checked: " + ", ".join(missing)) return None def _looks_like_image_id(reference: str) -> bool: if reference.startswith("sha256:"): return True return 12 <= len(reference) <= 64 and all(c in "0123456789abcdef" for c in reference) def _container_name(container: dict) -> str: """Human readable name for a container summary entry.""" for name in container.get("Names") or []: cleaned = str(name).strip().lstrip("/") if cleaned: return cleaned container_id = str(container.get("Id") or "") return container_id[:12] if container_id else "unknown" def _container_image_name(socket_path: str, container: dict) -> str: """Best available image reference for a container summary entry.""" name = (container.get("Image") or "").strip() if name and not _looks_like_image_id(name): return name container_id = container.get("Id") if container_id: try: details = docker_api_get(socket_path, f"/containers/{container_id}/json") except DockerError as exc: log("DEBUG", f"cannot inspect container {container_id[:12]}: {exc}") else: config_name = str((details.get("Config") or {}).get("Image") or "").strip() if config_name: return config_name return name or (container.get("ImageID") or "").strip() def containers_in_use(socket_path: str, include_self: bool = False) -> dict[str, list[str]]: """Map running-container image references to the containers using them. Images keep Docker's listing order and each container is listed once. The scanner's own container is skipped unless ``include_self`` is true, so a scheduled scan does not report on the scanner image itself. """ containers = docker_api_get(socket_path, "/containers/json") if not isinstance(containers, list): raise DockerError("/containers/json did not return a list") hostname = os.environ.get("HOSTNAME", "").strip() images: dict[str, list[str]] = {} for container in containers: if not isinstance(container, dict): continue container_id = str(container.get("Id") or "") if not include_self and hostname and container_id.startswith(hostname): log("DEBUG", f"skipping scanner's own container {container_id[:12]}") continue image = _container_image_name(socket_path, container) if not image: continue names = images.setdefault(image, []) name = _container_name(container) if name not in names: names.append(name) return images def images_in_use(socket_path: str, include_self: bool = False) -> list[str]: """Images referenced by running containers, in Docker's listing order.""" return list(containers_in_use(socket_path, include_self=include_self)) def containers_from_docker_cli() -> dict[str, list[str]]: """Fallback for environments where the socket is not reachable directly.""" if not shutil.which("docker"): log("DEBUG", "docker CLI not found") return {} try: completed = subprocess.run( ["docker", "ps", "--format", "{{.Image}}\t{{.Names}}"], capture_output=True, text=True, timeout=15, ) except OSError as exc: log("DEBUG", f"docker CLI failed: {exc}") return {} if completed.returncode != 0: log("DEBUG", f"docker ps failed: {completed.stderr.strip()}") return {} images: dict[str, list[str]] = {} for line in completed.stdout.splitlines(): image, _, raw_names = line.partition("\t") image = image.strip() if not image: continue names = images.setdefault(image, []) for name in raw_names.replace(",", " ").split(): if name not in names: names.append(name) return images def images_from_docker_cli() -> list[str]: """Images reported by the docker CLI, in listing order.""" return list(containers_from_docker_cli()) def discover_containers( include_self: bool = False, ) -> tuple[dict[str, list[str]], str | None]: """Find images in use and their containers. Returns (images, source).""" socket_path = find_docker_socket() if socket_path is None: log("DEBUG", "no Docker socket found") else: try: containers = containers_in_use(socket_path, include_self=include_self) except DockerError as exc: log("WARN", f"could not query Docker at {socket_path}: {exc}") else: if containers: return containers, f"running containers (via {socket_path})" containers = containers_from_docker_cli() if containers: return containers, "running containers (via docker CLI)" return {}, None def discover_images(include_self: bool = False) -> tuple[list[str], str | None]: """Find images currently in use. Returns (images, source description).""" containers, source = discover_containers(include_self=include_self) return list(containers), source # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- @dataclass class Settings: explicit_images: list[str] apprise_urls: str = "" apprise_config: str = "" apprise_title_prefix: str = "Vulnerability scan" notify_min_severity: str = "HIGH" fail_on: str = "HIGH" ignore_unfixed: bool = False trivy_severities: str = "" trivy_vex: list[str] = field(default_factory=list) notify_on_scan_error: bool = True report_dir: str = "/reports" 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 @classmethod def from_env(cls, argv: list[str]) -> "Settings": images = list(argv) images_env = os.environ.get("IMAGES", "") if images_env.strip(): images += split_words(images_env) images_file = os.environ.get("IMAGES_FILE", "").strip() if images_file: images += read_images_file(images_file) settings = cls( explicit_images=dedupe(images), apprise_urls=os.environ.get("APPRISE_URLS", "").strip(), apprise_config=os.environ.get("APPRISE_CONFIG", "").strip(), apprise_title_prefix=os.environ.get("APPRISE_TITLE_PREFIX", "Vulnerability scan"), notify_min_severity=os.environ.get("NOTIFY_MIN_SEVERITY", "HIGH"), fail_on=os.environ.get("FAIL_ON", "HIGH"), ignore_unfixed=env_bool("IGNORE_UNFIXED", False), trivy_severities=os.environ.get("TRIVY_SEVERITIES", "").strip(), trivy_vex=split_words(os.environ.get("TRIVY_VEX", "")), notify_on_scan_error=env_bool("NOTIFY_ON_SCAN_ERROR", True), report_dir=os.environ.get("REPORT_DIR", "/reports"), 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() return settings def validate(self) -> None: severity_rank(self.notify_min_severity) severity_rank(self.fail_on) if self.max_findings < 0: raise ValueError("MAX_FINDINGS_IN_BODY must be a non-negative integer") @property def notify_rank(self) -> int: return severity_rank(self.notify_min_severity) @property def fail_rank(self) -> int: return severity_rank(self.fail_on) def read_images_file(path: str) -> list[str]: if not os.path.isfile(path) or not os.access(path, os.R_OK): raise ValueError(f"IMAGES_FILE is not readable: {path}") images: list[str] = [] with open(path, encoding="utf-8") as handle: for line in handle: line = line.split("#", 1)[0] images += line.split() return images @dataclass class ScanTarget: """An image to scan and the running containers that use it.""" image: str containers: list[str] = field(default_factory=list) @property def container_names(self) -> str: return ", ".join(self.containers) @property def container_label(self) -> str: return "Container" if len(self.containers) == 1 else "Containers" @property def container_suffix(self) -> str: if not self.containers: return "" return f" (used by {self.container_label.lower()} {self.container_names})" @property def notification_context(self) -> str: if not self.containers: return "" return f" [{self.container_label.lower()}: {self.container_names}]" @property def container_detail(self) -> str: if not self.containers: return "" return f"{self.container_label}: {self.container_names}" def collect_targets(settings: Settings) -> list[ScanTarget]: if settings.explicit_images: return [ScanTarget(image=image) for image in settings.explicit_images] containers, source = discover_containers(include_self=settings.include_self) if containers: log("INFO", f"No images configured; scanning {len(containers)} image(s) from {source}") return [ ScanTarget(image=image, containers=names) for image, names in containers.items() ] log( "WARN", "No images configured and no running containers could be discovered; nothing to scan. " "Set IMAGES=... or DOCKER_SOCK to the host's Docker socket " "(/run/user//docker.sock for rootless Docker).", ) return [] def collect_images(settings: Settings) -> list[str]: return [target.image for target in collect_targets(settings)] # --------------------------------------------------------------------------- # Trivy # --------------------------------------------------------------------------- @dataclass class ScanResult: critical: int = 0 high: int = 0 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]: return { "CRITICAL": self.critical, "HIGH": self.high, "MEDIUM": self.medium, "LOW": self.low, } @property def fixable(self) -> int: return sum(1 for finding in self.findings if finding.get("FixedVersion")) @dataclass class AffectedTarget: image: str containers: list[str] package: str = "" installed: str = "" fixed: str = "" @dataclass class CveAggregation: cve: str severity: str affected: list[AffectedTarget] = field(default_factory=list) packages: list[tuple[str, str, str]] = field(default_factory=list) 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 high: int = 0 medium: int = 0 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]: return { "CRITICAL": self.critical, "HIGH": self.high, "MEDIUM": self.medium, "LOW": self.low, } @property def fixable(self) -> int: return sum(1 for cve in self.cves.values() if cve.fixable and severity_rank(cve.severity) >= 3) def add_vulnerability( self, vuln: dict, image: str, containers: list[str] ) -> None: severity = str(vuln.get("Severity", "")).upper() if severity == "CRITICAL": self.critical += 1 elif severity == "HIGH": self.high += 1 elif severity == "MEDIUM": self.medium += 1 elif severity == "LOW": self.low += 1 else: return cve_id = str(vuln.get("VulnerabilityID", "")) if not cve_id: return pkg = str(vuln.get("PkgName", "")) installed = str(vuln.get("InstalledVersion", "")) fixed = str(vuln.get("FixedVersion") or "") fixable = bool(fixed) if cve_id not in self.cves: self.cves[cve_id] = CveAggregation( cve=cve_id, severity=severity, fixable=fixable ) else: existing = self.cves[cve_id] if severity_rank(severity) > severity_rank(existing.severity): existing.severity = severity if fixable: existing.fixable = True agg = self.cves[cve_id] pkg_key = (pkg, installed, fixed) if pkg_key not in agg.packages: agg.packages.append(pkg_key) affected_key = (image, tuple(sorted(containers))) for affected in agg.affected: if affected.image == image and tuple(sorted(affected.containers)) == tuple( sorted(containers) ): if pkg and not affected.package: affected.package = pkg affected.installed = installed affected.fixed = fixed break else: agg.affected.append( AffectedTarget( image=image, containers=list(containers), package=pkg, installed=installed, fixed=fixed, ) ) def summarise(report: dict) -> ScanResult: result = ScanResult() for scan in report.get("Results") or []: for vuln in scan.get("Vulnerabilities") or []: severity = str(vuln.get("Severity", "")).upper() if severity == "CRITICAL": result.critical += 1 elif severity == "HIGH": result.high += 1 elif severity == "MEDIUM": 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) 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. if settings.docker_socket and not env.get("DOCKER_HOST"): env["DOCKER_HOST"] = f"unix://{settings.docker_socket}" return env def run_trivy(image: str, settings: Settings, context: str = "") -> tuple[int, str]: report_path = os.path.join(settings.report_dir, f"{report_slug(image)}.json") command = [ "trivy", "image", "--quiet", "--scanners", "vuln", "--format", "json", "--output", report_path, "--timeout", settings.trivy_timeout, "--cache-dir", settings.trivy_cache_dir, ] if settings.trivy_severities: command += ["--severity", settings.trivy_severities] for vex_source in settings.trivy_vex: command += ["--vex", vex_source] if settings.ignore_unfixed: command.append("--ignore-unfixed") command.append(image) log("INFO", f"Scanning {short_image(image)}{context}") try: completed = subprocess.run( command, capture_output=True, text=True, env=trivy_environment(settings), ) except OSError as exc: log("ERROR", f"Scan failed: {short_image(image)}{context}: {exc}") return 1, "" if completed.returncode != 0: detail = completed.stderr.strip().splitlines() if detail: log("ERROR", detail[-1]) log("ERROR", f"Scan failed: {short_image(image)}{context}") return 1, "" return 0, report_path # --------------------------------------------------------------------------- # Apprise notifications # --------------------------------------------------------------------------- def notify(settings: Settings, title: str, body: str) -> bool: if not settings.apprise_urls and not settings.apprise_config: log("WARN", "No APPRISE_URLS or APPRISE_CONFIG configured - notification skipped") return True if len(title) > 200: title = title[:197] + "..." log("INFO", f"Sending Apprise notification: {title} (title {len(title)} chars, body {len(body)} chars)") command = ["apprise", "-t", title, "-b", body, "-vv"] if settings.apprise_config: command += ["--config", settings.apprise_config] command += split_words(settings.apprise_urls) try: completed = subprocess.run(command, capture_output=True, text=True) except OSError as 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} (exit {completed.returncode}, title {len(title)} chars, body {len(body)} chars)", ) if 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 def build_body( image: str, result: ScanResult, report_path: str, max_findings: int, container_detail: str = "", ) -> str: lines = [f"Image: {short_image(image)}"] if container_detail: lines.append(container_detail) lines += [ f"Critical: {result.critical} | High: {result.high} | " f"Medium: {result.medium} | Low: {result.low}", f"Fixable (critical/high): {result.fixable}", "", ] total = result.critical + result.high if total > max_findings: lines.append(f"Findings (showing {max_findings} of {total}):") else: lines.append("Findings:") for vuln in result.findings[:max_findings]: lines.append( "* [{severity}] {vulnerability} - {package} {installed} -> {fixed}".format( severity=vuln.get("Severity", ""), vulnerability=vuln.get("VulnerabilityID", ""), package=vuln.get("PkgName", ""), installed=vuln.get("InstalledVersion", ""), fixed=vuln.get("FixedVersion") or "no fix available", ) ) lines += ["", f"Report: {report_path}"] return "\n".join(lines) MAX_NOTIFICATION_CHARS = 2500 DISCORD_MAX_NOTIFICATION_CHARS = 1200 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: 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( aggregated: AggregatedReport, max_findings: int, compact: bool = False ) -> 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 = [ f"Critical: {aggregated.critical} | High: {aggregated.high} | " f"Medium: {aggregated.medium} | Low: {aggregated.low}", f"Fixable (critical/high): {aggregated.fixable}", "", ] cves = sorted( aggregated.cves.values(), key=lambda item: ( -severity_rank(item.severity), item.cve, ), ) total = len(cves) max_findings = min(max_findings, 15) if total > max_findings: lines.append(f"CVEs (showing {max_findings} of {total}):") else: lines.append("CVEs:") for cve in cves[:max_findings]: packages = [] for pkg, installed, fixed in cve.packages[:1]: 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 "unknown" lines.append(f"* [{cve.severity}] {cve.cve} - {pkg_text}") for affected in cve.affected[:2]: containers = ", ".join(affected.containers) if affected.containers else "unknown" lines.append(f" - {containers} ({short_image(affected.image)})") if len(cve.affected) > 2: lines.append(f" - +{len(cve.affected) - 2} more") remaining = total - min(total, max_findings) if remaining > 0: lines.append(f"... +{remaining} more CVE(s)") if aggregated.reports: lines += ["", "Reports:"] + [f"- {p}" for p in aggregated.reports[:3]] if len(aggregated.reports) > 3: lines.append(f"- +{len(aggregated.reports) - 3} more") body = "\n".join(lines) 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 severity_text( counts: dict[str, int], fixable_counts: dict[str, int], severities: tuple[str, ...] | list[str] = REPORTED_SEVERITIES, ) -> str: return " ".join( f"{name[0]} {fixable_counts.get(name, 0)}/{counts.get(name, 0)}" for name in severities ) def visible_severities(rank: int) -> list[str]: """Severity columns that are not filtered out by the notify threshold.""" if rank <= 0: return list(REPORTED_SEVERITIES) return [name for name in REPORTED_SEVERITIES if severity_rank(name) >= rank] def summary_table( rows: list[tuple[str, str, dict[str, int], dict[str, int]]], severities: list[str] | tuple[str, ...], ) -> str: """Render image/container/severity counts as a monospaced markdown table.""" headers = ["Image", "Container"] + [name[0] for name in severities] table_rows: list[list[str]] = [] for image, container, counts, fixable_counts in rows: cells = [short_image(image), container or "-"] cells += [ f"{fixable_counts.get(name, 0)}/{counts.get(name, 0)}" for name in severities ] table_rows.append(cells) widths = [max(3, len(header)) for header in headers] for cells in table_rows: for index, cell in enumerate(cells): widths[index] = max(widths[index], len(cell)) def render(cells: list[str]) -> str: return "| " + " | ".join( cell.ljust(widths[index]) for index, cell in enumerate(cells) ) + " |" lines = [ "```", render(headers), "| " + " | ".join("-" * width for width in widths) + " |", ] lines += [render(cells) for cells in table_rows] lines.append("```") return "\n".join(lines) 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)) counts = severity_counts(entries) fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES} for entry in entries: if entry.fixable: fixable_counts[entry.severity] += 1 severities = visible_severities(severity_rank(min_severity)) lines = [ summary_table( [(target.image, container, counts, fixable_counts)], severities ), "", "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_important_summary_body( container: str, target: ScanTarget, result: ScanResult, min_severity: str ) -> str: """Summary for an important container with no findings above threshold.""" entries = cve_entries(result, severity_rank(min_severity)) counts = severity_counts(entries) fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES} for entry in entries: if entry.fixable: fixable_counts[entry.severity] += 1 severities = visible_severities(severity_rank(min_severity)) lines = [ summary_table( [(target.image, container, counts, fixable_counts)], severities ), "", "No CVEs at or above the notification threshold.", ] return "\n".join(lines) def _chunk_list(items: list[object], size: int) -> list[list[object]]: if size <= 0: return [items] return [items[i : i + size] for i in range(0, len(items), size)] def build_other_containers_bodies( rows: list, rank: int, ) -> list[str]: """Split the summary by container groups so each message stays small.""" if not rows: return [] severities = visible_severities(rank) bodies: list[str] = [] for chunk in _chunk_list(rows, 12): table_rows = [] for row in chunk: image, container, _, _, _, counts, fixable_counts = row table_rows.append((image, container, counts, fixable_counts)) bodies.append( "\n".join( [ "Other containers with findings:", "", summary_table(table_rows, severities), ] ) ) return bodies def build_final_summary_body( important_rows: list[tuple[str, ScanTarget, ScanResult]], all_rows: list, min_severity: str, ) -> str: """One notification at the end: important packages and overall totals.""" severities = visible_severities(severity_rank(min_severity)) table_rows: list[tuple[str, str, dict[str, int], dict[str, int]]] = [] package_lines: list[str] = [] for name, target, result in important_rows: entries = cve_entries(result, severity_rank(min_severity)) counts = severity_counts(entries) fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES} for entry in entries: if entry.fixable: fixable_counts[entry.severity] += 1 table_rows.append((target.image, name, counts, fixable_counts)) packages: list[str] = [] for entry in entries: for pkg, installed, fixed in entry.packages: if pkg: text = f"{pkg} {installed} -> {fixed or 'no fix available'}" elif installed or fixed: text = f"{installed} -> {fixed or 'no fix available'}" else: continue if text not in packages: packages.append(text) if packages: package_lines.append(f"* {name}: " + "; ".join(packages)) lines = ["Important containers:", ""] if table_rows: lines.append(summary_table(table_rows, severities)) else: lines.append("* none") if package_lines: lines += ["", "Packages:"] + package_lines overall_total = sum(row[2] for row in all_rows) overall_fixable = sum(row[3] for row in all_rows) overall_counts = {severity: 0 for severity in REPORTED_SEVERITIES} overall_fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES} for row in all_rows: counts = row[5] if len(row) > 5 else {} fixable_counts = row[6] if len(row) > 6 else {} for severity in REPORTED_SEVERITIES: overall_counts[severity] += counts.get(severity, 0) overall_fixable_counts[severity] += fixable_counts.get(severity, 0) lines += [ "", f"Overall severity: {severity_text(overall_counts, overall_fixable_counts, severities)}", f"Overall: {overall_fixable}/{overall_total} fixable/total in {len(all_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 # --------------------------------------------------------------------------- @dataclass class RunStats: scan_errors: int = 0 notify_errors: int = 0 failed_images: int = 0 def threshold_hits(result: ScanResult, rank: int) -> int: if rank <= 0: return 0 return sum(count for severity, count in result.counts.items() if severity_rank(severity) >= rank) def scan_image( target: ScanTarget, settings: Settings, stats: RunStats, aggregated: AggregatedReport, ) -> None: image = target.image context = target.container_suffix status, report_path = run_trivy(image, settings, context) if status != 0: stats.scan_errors += 1 if settings.notify_on_scan_error: lines = [f"Image: {short_image(image)}"] if target.container_detail: lines.append(target.container_detail) lines += [ "", "Trivy could not complete the scan. Check the scanner logs for details.", "", ] body = "\n".join(lines) discord = uses_discord(settings) body = _truncate_text(body, DISCORD_MAX_NOTIFICATION_CHARS if discord else MAX_NOTIFICATION_CHARS) title = f"Scan failed: {short_image(image)}{target.notification_context}" if len(title) > 200: title = f"Scan failed: {short_image(image)}" if not notify(settings, title, body): stats.notify_errors += 1 return try: with open(report_path, encoding="utf-8") as handle: report = json.load(handle) except (OSError, json.JSONDecodeError) as exc: log("ERROR", f"Could not read Trivy report {report_path}: {exc}") stats.scan_errors += 1 return result = summarise(report) aggregated.results[image] = result log( "INFO", f"{short_image(image)}{context}: {result.critical} critical, {result.high} high, " f"{result.medium} medium, {result.low} low " f"({result.fixable} fixable high/critical)", ) if report_path and report_path not in aggregated.reports: aggregated.reports.append(report_path) for vuln in result.findings: aggregated.add_vulnerability(vuln, image, target.containers) notify_hits = threshold_hits(result, settings.notify_rank) fail_hits = threshold_hits(result, settings.fail_rank) if notify_hits: log("DEBUG", f"{short_image(image)}{context}: findings above {settings.notify_min_severity} tracked for aggregated report") else: log("INFO", f"{short_image(image)}{context}: no findings at or above {settings.notify_min_severity}") if fail_hits: log("WARN", f"{short_image(image)}{context} meets FAIL_ON={settings.fail_on} ({fail_hits} finding(s))") stats.failed_images += 1 def usage() -> str: return f"""Usage: {PROG} [IMAGE ...] 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, the rest are summarised together, and a final notification lists the important packages and overall totals; 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). When none are given, the images used by running containers are scanned. Environment: DOCKER_SOCK Unix socket of the Docker daemon. Defaults to DOCKER_HOST, /var/run/docker.sock, /run/docker.sock and the rootless socket under /run/user//. INCLUDE_SELF true/false - also scan the scanner's own container image when images are auto-discovered (default: false). LOG_LEVEL DEBUG, INFO, WARN or ERROR (default: INFO). APPRISE_URLS Space separated Apprise URLs (quote the value). APPRISE_CONFIG Path to an Apprise config file (YAML). APPRISE_TITLE_PREFIX Notification title prefix (default: "Vulnerability scan"). NOTIFY_MIN_SEVERITY Lowest severity that triggers an alert: NONE, LOW, MEDIUM, HIGH, CRITICAL (default: HIGH; NONE disables). FAIL_ON Lowest severity that makes the script exit 1: NONE, LOW, MEDIUM, HIGH, CRITICAL (default: HIGH; NONE never). IGNORE_UNFIXED true/false - exclude vulnerabilities without a fix (default: false). TRIVY_SEVERITIES Comma separated severities Trivy should report (default: empty = all severities). TRIVY_VEX Space separated VEX sources passed to Trivy as --vex: "repo", "oci", "sbom-ref" or file paths. Earlier sources take priority (default: none). NOTIFY_ON_SCAN_ERROR true/false - alert when Trivy fails (default: true). REPORT_DIR Directory for JSON reports (default: /reports). 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). 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(): if len(line) > max_chars: while len(line) > max_chars: if current: parts.append("\n".join(current)) current = [] size = 0 chunk = line[:max_chars] line = line[max_chars:] parts.append(chunk) continue 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.""" original_title = title if len(title) > 200: original_title = title[:197] + "..." limit = min(notification_limit(settings), 1100) parts = _split_body(body, limit) if len(parts) > 1: log("INFO", f"Sending {title} in {len(parts)} parts") ok = True for index, part in enumerate(parts, 1): if len(parts) == 1: part_title = original_title else: suffix = f" [{index}/{len(parts)}]" if len(original_title) + len(suffix) > 200: part_title = original_title[: 200 - len(suffix) - 3] + "..." + suffix else: part_title = original_title + suffix 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 build_container_rows( targets: list[ScanTarget], aggregated: AggregatedReport, rank: int, skip: set[str] | None = None, include_empty: bool = False, ) -> list: """Build ``(image, container, total, fixable, unfixable, counts, fixable_counts)`` rows.""" skip = skip or set() rows: list = [] for target in targets: result = aggregated.results.get(target.image) if result is None: continue entries = cve_entries(result, rank) if not entries and not include_empty: continue total, fixable, unfixable = split_counts(entries) counts = severity_counts(entries) fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES} for entry in entries: if entry.fixable: fixable_counts[entry.severity] += 1 for container in target.containers or [""]: if container and container in skip: continue rows.append( (target.image, container, total, fixable, unfixable, counts, fixable_counts) ) return rows 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] = [] important_rows: list[tuple[str, ScanTarget, ScanResult]] = [] # 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 important_rows.append((name, target, result)) entries = cve_entries(result, settings.notify_rank) report_path = os.path.join(settings.report_dir, f"{report_slug(image)}.json") if entries: body = build_important_body( name, target, result, report_path, settings.notify_min_severity ) else: log( "DEBUG", f"Important container {name} ({short_image(image)}) has no findings at or above " f"{settings.notify_min_severity}", ) body = build_important_summary_body( name, target, result, settings.notify_min_severity ) title = f"{settings.apprise_title_prefix}: {name} ({short_image(image)})" log( "DEBUG", f"Preparing important-container notification for {name}: " f"{len(entries)} CVE(s) at or above threshold", ) send_notification_parts(settings, stats, title, body) if missing or not_scanned: send_notification_parts( settings, stats, f"{settings.apprise_title_prefix}: important containers", build_missing_body(missing, not_scanned), ) # One summary notification for every other container. rows = build_container_rows( targets, aggregated, settings.notify_rank, skip=important ) if rows: bodies = build_other_containers_bodies(rows, settings.notify_rank) for index, body in enumerate(bodies, 1): if len(bodies) == 1: title = f"{settings.apprise_title_prefix}: other containers" else: title = f"{settings.apprise_title_prefix}: other containers [{index}/{len(bodies)}]" if len(title) > 200: original = f"{settings.apprise_title_prefix}: other containers" suffix = f" [{index}/{len(bodies)}]" title = original[: 200 - len(suffix) - 3] + "..." + suffix body_truncated = _truncate_text(body, notification_limit(settings)) if notify(settings, title, body_truncated): continue body_truncated = _truncate_text(body, min(notification_limit(settings), 1100)) if notify(settings, title, body_truncated): continue stats.notify_errors += 1 # One final notification with the important packages and overall totals. all_rows = build_container_rows( targets, aggregated, settings.notify_rank, include_empty=True ) summary_body = build_final_summary_body( important_rows, all_rows, settings.notify_min_severity ) send_notification_parts( settings, stats, f"{settings.apprise_title_prefix}: summary", summary_body, ) 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) if argv and argv[0] in ("-h", "--help"): print(usage(), end="", file=sys.stderr) return 0 try: settings = Settings.from_env(argv) except ValueError as exc: die(str(exc)) if not shutil.which("trivy"): die("trivy not found in PATH") if (settings.apprise_urls or settings.apprise_config) and not shutil.which("apprise"): die("APPRISE_URLS/APPRISE_CONFIG is set but apprise is not installed") settings.docker_socket = find_docker_socket() if settings.trivy_vex: log("INFO", f"Applying VEX source(s): {', '.join(settings.trivy_vex)}") 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) stats = RunStats() aggregated = AggregatedReport() for target in targets: scan_image(target, settings, stats, aggregated) send_notifications( settings, targets, aggregated, stats, container_images, discovery_known ) log( "INFO", f"Scanned {len(targets)} image(s): {stats.failed_images} failed policy, " f"{stats.scan_errors} scan error(s), {stats.notify_errors} notification error(s)", ) if stats.scan_errors or stats.notify_errors: return 2 if stats.failed_images: return 1 return 0 def threshold_hits_from_counts(counts: dict[str, int], rank: int) -> int: if rank <= 0: return 0 total = 0 for severity, count in counts.items(): if severity_rank(severity) >= rank: total += count return total if __name__ == "__main__": try: raise SystemExit(main()) except KeyboardInterrupt: raise SystemExit(130) from None