1083 lines
37 KiB
Python
Executable File
1083 lines
37 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 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 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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
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),
|
|
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)
|
|
|
|
@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 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)
|
|
|
|
@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 ("CRITICAL", "HIGH"):
|
|
result.findings.append(vuln)
|
|
result.findings.sort(
|
|
key=lambda v: (
|
|
0 if str(v.get("Severity", "")).upper() == "CRITICAL" else 1,
|
|
str(v.get("VulnerabilityID", "")),
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
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 {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: {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: {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
|
|
|
|
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: {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 = 1800
|
|
|
|
|
|
def uses_discord(settings: Settings) -> bool:
|
|
combined = f"{settings.apprise_urls} {settings.apprise_config}"
|
|
return "discord://" in combined
|
|
|
|
|
|
def _truncate_text(text: str, max_chars: int) -> str:
|
|
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} ({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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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: {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: {image}{target.notification_context}"
|
|
if len(title) > 200:
|
|
title = f"Scan failed: {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)
|
|
log(
|
|
"INFO",
|
|
f"{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"{image}{context}: findings above {settings.notify_min_severity} tracked for aggregated report")
|
|
else:
|
|
log("INFO", f"{image}{context}: no findings at or above {settings.notify_min_severity}")
|
|
|
|
if fail_hits:
|
|
log("WARN", f"{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. Findings are aggregated by CVE across all
|
|
images and a single notification is sent when any reach
|
|
NOTIFY_MIN_SEVERITY (default: HIGH). Each CVE lists the containers and images
|
|
that are affected.
|
|
|
|
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).
|
|
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 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
|
|
|
|
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)
|
|
|
|
if aggregated.cves and threshold_hits_from_counts(aggregated.counts, settings.notify_rank):
|
|
title = f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s) across {len(targets)} image(s)"
|
|
if len(title) > 200:
|
|
title = f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s)"
|
|
discord = uses_discord(settings)
|
|
body = build_aggregated_body(aggregated, settings.max_findings, compact=discord)
|
|
if discord:
|
|
body = _truncate_text(body, DISCORD_MAX_NOTIFICATION_CHARS)
|
|
else:
|
|
body = _truncate_text(body, MAX_NOTIFICATION_CHARS)
|
|
log("DEBUG", f"Preparing aggregated notification: title {len(title)} chars, body {len(body)} chars, {len(aggregated.cves)} CVEs")
|
|
if not notify(settings, title, body):
|
|
log("WARN", "Retrying aggregated notification with minimal summary")
|
|
minimal = (
|
|
f"{settings.apprise_title_prefix}: {len(aggregated.cves)} CVE(s) across {len(targets)} image(s)\n"
|
|
f"C:{aggregated.critical} H:{aggregated.high} M:{aggregated.medium} L:{aggregated.low}\n"
|
|
f"Full details in reports in {settings.report_dir}"
|
|
)
|
|
minimal = _truncate_text(minimal, DISCORD_MAX_NOTIFICATION_CHARS if discord else MAX_NOTIFICATION_CHARS)
|
|
if not notify(settings, title, minimal):
|
|
stats.notify_errors += 1
|
|
elif aggregated.cves:
|
|
log("INFO", f"{len(aggregated.cves)} CVE(s) tracked but below NOTIFY_MIN_SEVERITY={settings.notify_min_severity}")
|
|
|
|
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
|