1697 lines
58 KiB
Python
Executable File
1697 lines
58 KiB
Python
Executable File
#!/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/<uid>/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/<uid>/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 sorted(
|
|
rows, key=lambda row: (row[1].casefold(), row[0].casefold())
|
|
):
|
|
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 _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 image 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] = []
|
|
# Merge important containers that share an image into one row/package line.
|
|
by_image: dict[str, tuple[list[str], ScanTarget, ScanResult]] = {}
|
|
for name, target, result in important_rows:
|
|
if target.image in by_image:
|
|
by_image[target.image][0].append(name)
|
|
else:
|
|
by_image[target.image] = ([name], target, result)
|
|
groups = sorted(
|
|
(
|
|
(", ".join(sorted(names, key=str.casefold)), target, result)
|
|
for names, target, result in by_image.values()
|
|
),
|
|
key=lambda group: (group[0].casefold(), group[1].image.casefold()),
|
|
)
|
|
for name, target, result in groups:
|
|
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)} image(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/<uid>/.
|
|
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, containers, total, fixable, unfixable, counts, fixable_counts)`` rows.
|
|
|
|
One row per image: containers that share the image are merged into a single
|
|
row so its counts are not repeated.
|
|
"""
|
|
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
|
|
if target.containers:
|
|
containers = [name for name in target.containers if name not in skip]
|
|
if not containers:
|
|
continue
|
|
container = ", ".join(sorted(containers, key=str.casefold))
|
|
else:
|
|
container = ""
|
|
rows.append(
|
|
(target.image, container, total, fixable, unfixable, counts, fixable_counts)
|
|
)
|
|
rows.sort(key=lambda row: (row[1].casefold(), row[0].casefold()))
|
|
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)
|
|
if not entries:
|
|
# No individual notification for clean containers; the final
|
|
# summary table already lists them with 0/0 totals.
|
|
log(
|
|
"DEBUG",
|
|
f"Important container {name} ({short_image(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} ({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
|