From 81961f268f042e13014681811bbbd52b3322a1f4 Mon Sep 17 00:00:00 2001 From: David Duran Rosich Date: Fri, 23 Oct 2026 22:31:56 +0200 Subject: [PATCH] add container info --- README.md | 4 +- scan.py | 157 ++++++++++++++++++++++++++++++++------------- tests/test_scan.py | 101 +++++++++++++++++++++++++++-- 3 files changed, 210 insertions(+), 52 deletions(-) diff --git a/README.md b/README.md index 883da19..df8455f 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,9 @@ For the last one the scanner asks the Docker Engine API on the Unix socket for `/containers/json` and scans each `Image` reference, de-duplicated and in listing order. Stopped containers are not included. The scanner's own container image is skipped so a scheduled run does not report on itself; set -`INCLUDE_SELF=true` to include it. +`INCLUDE_SELF=true` to include it. Scan log lines name the running container(s) +that use each image, for example +`Scanning nginx:1.27 (used by container web)`. The socket is looked up in this order: diff --git a/scan.py b/scan.py index 59ad64a..cb4cda3 100755 --- a/scan.py +++ b/scan.py @@ -217,6 +217,16 @@ def _looks_like_image_id(reference: str) -> bool: 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() @@ -237,11 +247,12 @@ def _container_image_name(socket_path: str, container: dict) -> str: 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. +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. - The scanner's own container is skipped unless ``include_self`` is true, so - a scheduled scan does not report on the scanner image itself. + 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): @@ -249,7 +260,7 @@ def images_in_use(socket_path: str, include_self: bool = False) -> list[str]: hostname = os.environ.get("HOSTNAME", "").strip() - images: list[str] = [] + images: dict[str, list[str]] = {} for container in containers: if not isinstance(container, dict): continue @@ -258,51 +269,84 @@ def images_in_use(socket_path: str, include_self: bool = False) -> list[str]: 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) + if not image: + continue + names = images.setdefault(image, []) + name = _container_name(container) + if name not in names: + names.append(name) + return images -def images_from_docker_cli() -> list[str]: +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 [] + return {} try: completed = subprocess.run( - ["docker", "ps", "--format", "{{.Image}}"], + ["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 [] + return {} if completed.returncode != 0: log("DEBUG", f"docker ps failed: {completed.stderr.strip()}") - return [] - return dedupe(split_words(completed.stdout.replace("\n", " "))) + 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 discover_images(include_self: bool = False) -> tuple[list[str], str | None]: - """Find images currently in use. Returns (images, source description).""" +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: - images = images_in_use(socket_path, include_self=include_self) + 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 images: - return images, f"running containers (via {socket_path})" + if containers: + return containers, f"running containers (via {socket_path})" - images = images_from_docker_cli() - if images: - return images, "running containers (via docker CLI)" + containers = containers_from_docker_cli() + if containers: + return containers, "running containers (via docker CLI)" - return [], None + 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 # --------------------------------------------------------------------------- @@ -386,14 +430,33 @@ def read_images_file(path: str) -> list[str]: return images -def collect_images(settings: Settings) -> list[str]: - if settings.explicit_images: - return list(settings.explicit_images) +@dataclass +class ScanTarget: + """An image to scan and the running containers that use it.""" - 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 + image: str + containers: list[str] = field(default_factory=list) + + @property + def container_suffix(self) -> str: + if not self.containers: + return "" + if len(self.containers) == 1: + return f" (used by container {self.containers[0]})" + return f" (used by containers {', '.join(self.containers)})" + + +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", @@ -404,6 +467,10 @@ def collect_images(settings: Settings) -> list[str]: return [] +def collect_images(settings: Settings) -> list[str]: + return [target.image for target in collect_targets(settings)] + + # --------------------------------------------------------------------------- # Trivy # --------------------------------------------------------------------------- @@ -463,7 +530,7 @@ def trivy_environment(settings: Settings) -> dict[str, str]: return env -def run_trivy(image: str, settings: Settings) -> tuple[int, str]: +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", @@ -486,7 +553,7 @@ def run_trivy(image: str, settings: Settings) -> tuple[int, str]: command.append("--ignore-unfixed") command.append(image) - log("INFO", f"Scanning {image}") + log("INFO", f"Scanning {image}{context}") try: completed = subprocess.run( command, @@ -495,14 +562,14 @@ def run_trivy(image: str, settings: Settings) -> tuple[int, str]: env=trivy_environment(settings), ) except OSError as exc: - log("ERROR", f"Scan failed: {image}: {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}") + log("ERROR", f"Scan failed: {image}{context}") return 1, "" return 0, report_path @@ -586,8 +653,10 @@ def threshold_hits(result: ScanResult, rank: int) -> int: 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) +def scan_image(target: ScanTarget, settings: Settings, stats: RunStats) -> 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: @@ -610,7 +679,7 @@ def scan_image(image: str, settings: Settings, stats: RunStats) -> None: result = summarise(report) log( "INFO", - f"{image}: {result.critical} critical, {result.high} high, " + f"{image}{context}: {result.critical} critical, {result.high} high, " f"{result.medium} medium, {result.low} low " f"({result.fixable} fixable high/critical)", ) @@ -627,10 +696,10 @@ def scan_image(image: str, settings: Settings, stats: RunStats) -> None: if not notify(settings, title, body): stats.notify_errors += 1 else: - log("INFO", f"{image}: no findings at or above {settings.notify_min_severity}") + log("INFO", f"{image}{context}: 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))") + log("WARN", f"{image}{context} meets FAIL_ON={settings.fail_on} ({fail_hits} finding(s))") stats.failed_images += 1 @@ -694,20 +763,20 @@ def main(argv: list[str] | None = None) -> int: settings.docker_socket = find_docker_socket() - images = collect_images(settings) - if not images: + 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() - for image in images: - scan_image(image, settings, stats) + for target in targets: + scan_image(target, settings, stats) log( "INFO", - f"Scanned {len(images)} image(s): {stats.failed_images} failed policy, " + f"Scanned {len(targets)} image(s): {stats.failed_images} failed policy, " f"{stats.scan_errors} scan error(s), {stats.notify_errors} notification error(s)", ) diff --git a/tests/test_scan.py b/tests/test_scan.py index db48223..b9bf508 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -6,14 +6,17 @@ discovery is exercised end to end without Docker being installed. from __future__ import annotations +import io import json import os import shutil import socketserver +import subprocess import sys import tempfile import threading import unittest +from contextlib import redirect_stderr from http.server import BaseHTTPRequestHandler from pathlib import Path from unittest import mock @@ -70,8 +73,16 @@ class FakeDocker: shutil.rmtree(self.directory, ignore_errors=True) -def container(container_id: str, image: str, image_id: str = "sha256:deadbeef") -> dict: - return {"Id": container_id, "Image": image, "ImageID": image_id} +def container( + container_id: str, + image: str, + image_id: str = "sha256:deadbeef", + name: str | None = None, +) -> dict: + entry = {"Id": container_id, "Image": image, "ImageID": image_id} + if name: + entry["Names"] = [f"/{name}"] + return entry class DockerDiscoveryTests(unittest.TestCase): @@ -98,6 +109,25 @@ class DockerDiscoveryTests(unittest.TestCase): with FakeDocker([container("a" * 64, "alpine:3.20"), "nonsense"]) as docker: # type: ignore[list-item] self.assertEqual(scan.images_in_use(docker.socket_path), ["alpine:3.20"]) + def test_containers_in_use_groups_containers_by_image(self) -> None: + containers = [ + container("a" * 64, "nginx:1.27", name="web"), + container("b" * 64, "nginx:1.27", name="web-worker"), + container("c" * 64, "redis:7", name="cache"), + ] + with FakeDocker(containers) as docker: + mapping = scan.containers_in_use(docker.socket_path) + self.assertEqual( + mapping, + {"nginx:1.27": ["web", "web-worker"], "redis:7": ["cache"]}, + ) + + def test_containers_in_use_falls_back_to_container_id(self) -> None: + container_id = "a" * 64 + with FakeDocker([container(container_id, "alpine:3.20")]) as docker: + mapping = scan.containers_in_use(docker.socket_path) + self.assertEqual(mapping, {"alpine:3.20": [container_id[:12]]}) + def test_images_in_use_skips_scanner_container(self) -> None: scanner_id = "f" * 64 containers = [container(scanner_id, "docker-vuln-scan:latest"), container("a" * 64, "testphp-php")] @@ -155,18 +185,43 @@ class DockerDiscoveryTests(unittest.TestCase): self.assertEqual(images, ["testphp-php"]) self.assertIn("running containers", source or "") + def test_discover_containers_uses_socket(self) -> None: + containers = [container("a" * 64, "testphp-php", name="testphp")] + with FakeDocker(containers) as docker: + with mock.patch.dict(os.environ, {"DOCKER_SOCK": docker.socket_path, "DOCKER_HOST": ""}): + discovered, source = scan.discover_containers() + self.assertEqual(discovered, {"testphp-php": ["testphp"]}) + self.assertIn("running containers", source or "") + def test_discover_images_falls_back_to_cli(self) -> None: with mock.patch.object(scan, "find_docker_socket", return_value=None): - with mock.patch.object(scan, "images_from_docker_cli", return_value=["redis:7"]): + with mock.patch.object( + scan, "containers_from_docker_cli", return_value={"redis:7": ["cache"]} + ): images, source = scan.discover_images() self.assertEqual(images, ["redis:7"]) self.assertEqual(source, "running containers (via docker CLI)") def test_discover_images_returns_empty_without_any_source(self) -> None: with mock.patch.object(scan, "find_docker_socket", return_value=None): - with mock.patch.object(scan, "images_from_docker_cli", return_value=[]): + with mock.patch.object(scan, "containers_from_docker_cli", return_value={}): self.assertEqual(scan.discover_images(), ([], None)) + def test_containers_from_docker_cli_parses_image_and_names(self) -> None: + completed = subprocess.CompletedProcess( + [], + 0, + stdout="nginx:1.27\tweb,web-worker\nredis:7\tcache\n", + stderr="", + ) + with mock.patch.object(scan.shutil, "which", return_value="/usr/bin/docker"): + with mock.patch.object(scan.subprocess, "run", return_value=completed): + mapping = scan.containers_from_docker_cli() + self.assertEqual( + mapping, + {"nginx:1.27": ["web", "web-worker"], "redis:7": ["cache"]}, + ) + class ImageListTests(unittest.TestCase): def test_split_words_handles_newlines_and_spaces(self) -> None: @@ -193,20 +248,39 @@ class ImageListTests(unittest.TestCase): def test_collect_images_prefers_explicit_list(self) -> None: settings = scan.Settings(explicit_images=["nginx:1.27"]) - with mock.patch.object(scan, "discover_images") as discover: + with mock.patch.object(scan, "discover_containers") as discover: self.assertEqual(scan.collect_images(settings), ["nginx:1.27"]) discover.assert_not_called() def test_collect_images_discovers_when_empty(self) -> None: settings = scan.Settings(explicit_images=[]) - with mock.patch.object(scan, "discover_images", return_value=(["redis:7"], "fake")): + with mock.patch.object( + scan, "discover_containers", return_value=({"redis:7": ["cache"]}, "fake") + ): self.assertEqual(scan.collect_images(settings), ["redis:7"]) def test_collect_images_returns_empty_when_nothing_found(self) -> None: settings = scan.Settings(explicit_images=[]) - with mock.patch.object(scan, "discover_images", return_value=([], None)): + with mock.patch.object(scan, "discover_containers", return_value=({}, None)): self.assertEqual(scan.collect_images(settings), []) + def test_collect_targets_keeps_container_names(self) -> None: + settings = scan.Settings(explicit_images=[]) + discovered = {"nginx:1.27": ["web", "web-worker"]} + with mock.patch.object(scan, "discover_containers", return_value=(discovered, "fake")): + targets = scan.collect_targets(settings) + self.assertEqual([target.image for target in targets], ["nginx:1.27"]) + self.assertEqual(targets[0].containers, ["web", "web-worker"]) + self.assertEqual( + targets[0].container_suffix, " (used by containers web, web-worker)" + ) + + def test_collect_targets_explicit_images_have_no_containers(self) -> None: + settings = scan.Settings(explicit_images=["nginx:1.27"]) + targets = scan.collect_targets(settings) + self.assertEqual(targets[0].containers, []) + self.assertEqual(targets[0].container_suffix, "") + class SettingsTests(unittest.TestCase): def test_settings_reads_environment(self) -> None: @@ -388,6 +462,19 @@ printf '%s\\n' "$*" >> "$APPRISE_LOG" self.assertTrue(os.path.exists(os.path.join(self.env["REPORT_DIR"], "testphp-php.json"))) self.assertIn("testphp-php", Path(self.apprise_log).read_text()) + def test_scan_logs_containers_using_the_image(self) -> None: + containers = [container("a" * 64, "testphp-php", name="testphp")] + with FakeDocker(containers) as docker: + stderr = io.StringIO() + with mock.patch.object(scan, "MIN_LOG_LEVEL", "INFO"): + with redirect_stderr(stderr): + code = self.run_main([], docker.socket_path) + + self.assertEqual(code, 1) + self.assertIn( + "Scanning testphp-php (used by container testphp)", stderr.getvalue() + ) + def test_exit_code_zero_when_policy_does_not_fail(self) -> None: with FakeDocker([container("a" * 64, "testphp-php")]) as docker: code = self.run_main([], docker.socket_path, FAIL_ON="NONE", NOTIFY_MIN_SEVERITY="NONE")