726 lines
24 KiB
Python
Executable File
726 lines
24 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_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 images_in_use(socket_path: str, include_self: bool = False) -> list[str]:
|
|
"""Images referenced by running containers, in Docker's listing order.
|
|
|
|
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: 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 image:
|
|
images.append(image)
|
|
return dedupe(images)
|
|
|
|
|
|
def images_from_docker_cli() -> 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}}"],
|
|
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 []
|
|
return dedupe(split_words(completed.stdout.replace("\n", " ")))
|
|
|
|
|
|
def discover_images(include_self: bool = False) -> tuple[list[str], str | None]:
|
|
"""Find images currently in use. Returns (images, source description)."""
|
|
socket_path = find_docker_socket()
|
|
if socket_path is None:
|
|
log("DEBUG", "no Docker socket found")
|
|
else:
|
|
try:
|
|
images = images_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 images:
|
|
return images, f"running containers (via {socket_path})"
|
|
|
|
images = images_from_docker_cli()
|
|
if images:
|
|
return images, "running containers (via docker CLI)"
|
|
|
|
return [], None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 = ""
|
|
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(),
|
|
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
|
|
|
|
|
|
def collect_images(settings: Settings) -> list[str]:
|
|
if settings.explicit_images:
|
|
return list(settings.explicit_images)
|
|
|
|
images, source = discover_images(include_self=settings.include_self)
|
|
if images:
|
|
log("INFO", f"No images configured; scanning {len(images)} image(s) from {source}")
|
|
return images
|
|
|
|
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 []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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"))
|
|
|
|
|
|
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) -> 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]
|
|
if settings.ignore_unfixed:
|
|
command.append("--ignore-unfixed")
|
|
command.append(image)
|
|
|
|
log("INFO", f"Scanning {image}")
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
capture_output=True,
|
|
text=True,
|
|
env=trivy_environment(settings),
|
|
)
|
|
except OSError as exc:
|
|
log("ERROR", f"Scan failed: {image}: {exc}")
|
|
return 1, ""
|
|
|
|
if completed.returncode != 0:
|
|
detail = completed.stderr.strip().splitlines()
|
|
if detail:
|
|
log("ERROR", detail[-1])
|
|
log("ERROR", f"Scan failed: {image}")
|
|
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}")
|
|
command = ["apprise", "--title", title, "--body", body]
|
|
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}")
|
|
return False
|
|
|
|
if completed.returncode != 0:
|
|
log("ERROR", f"Apprise notification failed: {title}")
|
|
if completed.stderr.strip():
|
|
log("ERROR", completed.stderr.strip())
|
|
return False
|
|
return True
|
|
|
|
|
|
def build_body(image: str, result: ScanResult, report_path: str, max_findings: int) -> str:
|
|
lines = [
|
|
f"Image: {image}",
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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(image: str, settings: Settings, stats: RunStats) -> None:
|
|
status, report_path = run_trivy(image, settings)
|
|
if status != 0:
|
|
stats.scan_errors += 1
|
|
if settings.notify_on_scan_error:
|
|
body = (
|
|
f"Image: {image}\n\n"
|
|
"Trivy could not complete the scan. Check the scanner logs for details.\n"
|
|
)
|
|
if not notify(settings, f"Scan failed: {image}", 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}: {result.critical} critical, {result.high} high, "
|
|
f"{result.medium} medium, {result.low} low "
|
|
f"({result.fixable} fixable high/critical)",
|
|
)
|
|
|
|
notify_hits = threshold_hits(result, settings.notify_rank)
|
|
fail_hits = threshold_hits(result, settings.fail_rank)
|
|
|
|
if notify_hits:
|
|
title = (
|
|
f"{settings.apprise_title_prefix}: {image} "
|
|
f"({result.critical} critical, {result.high} high)"
|
|
)
|
|
body = build_body(image, result, report_path, settings.max_findings)
|
|
if not notify(settings, title, body):
|
|
stats.notify_errors += 1
|
|
else:
|
|
log("INFO", f"{image}: no findings at or above {settings.notify_min_severity}")
|
|
|
|
if fail_hits:
|
|
log("WARN", f"{image} 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 a finding at or above
|
|
NOTIFY_MIN_SEVERITY (default: HIGH) is detected, a notification is sent
|
|
through Apprise.
|
|
|
|
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).
|
|
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()
|
|
|
|
images = collect_images(settings)
|
|
if not images:
|
|
return 0
|
|
|
|
os.makedirs(settings.report_dir, exist_ok=True)
|
|
os.makedirs(settings.trivy_cache_dir, exist_ok=True)
|
|
|
|
stats = RunStats()
|
|
for image in images:
|
|
scan_image(image, settings, stats)
|
|
|
|
log(
|
|
"INFO",
|
|
f"Scanned {len(images)} 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
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except KeyboardInterrupt:
|
|
raise SystemExit(130) from None
|