771 lines
33 KiB
Python
771 lines
33 KiB
Python
"""Tests for scan.py.
|
|
|
|
The Docker Engine API is faked with a real Unix socket server, so image
|
|
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
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import scan
|
|
|
|
|
|
class _FakeDockerHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None: # noqa: N802 - required by the base class
|
|
if self.path == "/containers/json":
|
|
status, payload = 200, self.server.containers # type: ignore[attr-defined]
|
|
elif self.path.startswith("/containers/") and self.path.endswith("/json"):
|
|
container_id = self.path.split("/")[2]
|
|
configs = self.server.configs # type: ignore[attr-defined]
|
|
if container_id in configs:
|
|
status, payload = 200, configs[container_id]
|
|
else:
|
|
status, payload = 404, {"message": "no such container"}
|
|
else:
|
|
status, payload = 404, {"message": "not found"}
|
|
|
|
body = json.dumps(payload).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args: object) -> None:
|
|
pass
|
|
|
|
|
|
class FakeDocker:
|
|
"""A Docker API listening on a temporary Unix socket."""
|
|
|
|
def __init__(self, containers: list[dict], configs: dict[str, dict] | None = None):
|
|
self.directory = tempfile.mkdtemp(prefix="fakedocker-")
|
|
self.socket_path = os.path.join(self.directory, "docker.sock")
|
|
self.server = socketserver.UnixStreamServer(self.socket_path, _FakeDockerHandler)
|
|
self.server.containers = containers # type: ignore[attr-defined]
|
|
self.server.configs = configs or {} # type: ignore[attr-defined]
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
|
|
def __enter__(self) -> "FakeDocker":
|
|
self.thread.start()
|
|
return self
|
|
|
|
def __exit__(self, *exc_info: object) -> None:
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=5)
|
|
shutil.rmtree(self.directory, ignore_errors=True)
|
|
|
|
|
|
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):
|
|
def test_images_in_use_returns_running_images(self) -> None:
|
|
with FakeDocker([container("a" * 64, "testphp-php"), container("b" * 64, "redis:7")]) as docker:
|
|
self.assertEqual(scan.images_in_use(docker.socket_path), ["testphp-php", "redis:7"])
|
|
|
|
def test_images_in_use_dedupes_shared_images(self) -> None:
|
|
with FakeDocker([container("a" * 64, "nginx:1.27"), container("b" * 64, "nginx:1.27")]) as docker:
|
|
self.assertEqual(scan.images_in_use(docker.socket_path), ["nginx:1.27"])
|
|
|
|
def test_images_in_use_resolves_image_id_via_inspect(self) -> None:
|
|
image_id = "sha256:" + "c" * 64
|
|
configs = {"a" * 64: {"Config": {"Image": "ghcr.io/acme/app:1.2.3"}}}
|
|
with FakeDocker([container("a" * 64, image_id)], configs) as docker:
|
|
self.assertEqual(scan.images_in_use(docker.socket_path), ["ghcr.io/acme/app:1.2.3"])
|
|
|
|
def test_images_in_use_falls_back_to_image_id(self) -> None:
|
|
image_id = "sha256:" + "d" * 64
|
|
with FakeDocker([container("a" * 64, "", image_id)], {}) as docker:
|
|
self.assertEqual(scan.images_in_use(docker.socket_path), [image_id])
|
|
|
|
def test_images_in_use_ignores_non_dict_entries(self) -> None:
|
|
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")]
|
|
with FakeDocker(containers) as docker:
|
|
with mock.patch.dict(os.environ, {"HOSTNAME": scanner_id[:12]}):
|
|
images = scan.images_in_use(docker.socket_path)
|
|
self.assertEqual(images, ["testphp-php"])
|
|
|
|
def test_images_in_use_can_include_scanner_container(self) -> None:
|
|
scanner_id = "f" * 64
|
|
containers = [container(scanner_id, "docker-vuln-scan:latest")]
|
|
with FakeDocker(containers) as docker:
|
|
with mock.patch.dict(os.environ, {"HOSTNAME": scanner_id[:12]}):
|
|
images = scan.images_in_use(docker.socket_path, include_self=True)
|
|
self.assertEqual(images, ["docker-vuln-scan:latest"])
|
|
|
|
def test_candidate_sockets_includes_sockets_of_other_users(self) -> None:
|
|
other = "/run/user/1000/docker.sock"
|
|
with mock.patch.object(scan.glob, "glob", return_value=[other]):
|
|
self.assertIn(other, scan.candidate_sockets())
|
|
|
|
def test_docker_api_get_raises_when_socket_is_missing(self) -> None:
|
|
with self.assertRaises(scan.DockerError):
|
|
scan.docker_api_get("/nonexistent/docker.sock", "/containers/json")
|
|
|
|
def test_docker_api_get_raises_on_http_error(self) -> None:
|
|
with FakeDocker([]) as docker:
|
|
with self.assertRaises(scan.DockerError):
|
|
scan.docker_api_get(docker.socket_path, "/containers/nope/json")
|
|
|
|
def test_find_docker_socket_honours_docker_sock(self) -> None:
|
|
with FakeDocker([]) as docker:
|
|
with mock.patch.dict(os.environ, {"DOCKER_SOCK": docker.socket_path, "DOCKER_HOST": ""}):
|
|
self.assertEqual(scan.find_docker_socket(), docker.socket_path)
|
|
|
|
def test_find_docker_socket_honours_docker_host(self) -> None:
|
|
with FakeDocker([]) as docker:
|
|
env = {"DOCKER_SOCK": "", "DOCKER_HOST": f"unix://{docker.socket_path}"}
|
|
with mock.patch.dict(os.environ, env):
|
|
self.assertEqual(scan.find_docker_socket(), docker.socket_path)
|
|
|
|
def test_find_docker_socket_rejects_regular_files(self) -> None:
|
|
path = os.path.join(tempfile.mkdtemp(prefix="notasocket-"), "docker.sock")
|
|
Path(path).write_text("not a socket")
|
|
try:
|
|
with mock.patch.object(scan, "candidate_sockets", return_value=[path]):
|
|
self.assertIsNone(scan.find_docker_socket())
|
|
finally:
|
|
shutil.rmtree(os.path.dirname(path), ignore_errors=True)
|
|
|
|
def test_discover_images_uses_socket(self) -> None:
|
|
with FakeDocker([container("a" * 64, "testphp-php")]) as docker:
|
|
with mock.patch.dict(os.environ, {"DOCKER_SOCK": docker.socket_path, "DOCKER_HOST": ""}):
|
|
images, source = scan.discover_images()
|
|
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, "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, "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:
|
|
self.assertEqual(scan.split_words("a:1 b:2\n c:3\n"), ["a:1", "b:2", "c:3"])
|
|
|
|
def test_dedupe_keeps_order(self) -> None:
|
|
self.assertEqual(scan.dedupe(["b", "a", "b", "c", "a"]), ["b", "a", "c"])
|
|
|
|
def test_report_slug_is_filesystem_safe(self) -> None:
|
|
self.assertEqual(scan.report_slug("ghcr.io/acme/app:1.2.3"), "ghcr.io_acme_app_1.2.3")
|
|
|
|
def test_read_images_file_ignores_comments_and_blank_lines(self) -> None:
|
|
directory = tempfile.mkdtemp(prefix="images-")
|
|
path = os.path.join(directory, "images.txt")
|
|
Path(path).write_text("# comment\nnginx:1.27\n\nredis:7 # trailing\n")
|
|
try:
|
|
self.assertEqual(scan.read_images_file(path), ["nginx:1.27", "redis:7"])
|
|
finally:
|
|
shutil.rmtree(directory, ignore_errors=True)
|
|
|
|
def test_read_images_file_rejects_missing_file(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
scan.read_images_file("/nonexistent/images.txt")
|
|
|
|
def test_collect_images_prefers_explicit_list(self) -> None:
|
|
settings = scan.Settings(explicit_images=["nginx:1.27"])
|
|
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_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_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)"
|
|
)
|
|
self.assertEqual(
|
|
targets[0].notification_context, " [containers: web, web-worker]"
|
|
)
|
|
self.assertEqual(
|
|
targets[0].container_detail, "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:
|
|
env = {
|
|
"IMAGES": "",
|
|
"IMAGES_FILE": "",
|
|
"NOTIFY_MIN_SEVERITY": "CRITICAL",
|
|
"FAIL_ON": "NONE",
|
|
"IGNORE_UNFIXED": "true",
|
|
"MAX_FINDINGS_IN_BODY": "3",
|
|
"TRIVY_VEX": "/vex/app.json repo",
|
|
"IMPORTANT_CONTAINERS": "traefik, web-frontend\nworker",
|
|
}
|
|
with mock.patch.dict(os.environ, env, clear=False):
|
|
settings = scan.Settings.from_env([])
|
|
self.assertEqual(settings.notify_min_severity, "CRITICAL")
|
|
self.assertEqual(settings.fail_rank, 0)
|
|
self.assertTrue(settings.ignore_unfixed)
|
|
self.assertEqual(settings.max_findings, 3)
|
|
self.assertEqual(settings.trivy_vex, ["/vex/app.json", "repo"])
|
|
self.assertEqual(
|
|
settings.important_containers, ["traefik", "web-frontend", "worker"]
|
|
)
|
|
|
|
def test_settings_rejects_unknown_severity(self) -> None:
|
|
with mock.patch.dict(os.environ, {"NOTIFY_MIN_SEVERITY": "SEVERE"}, clear=False):
|
|
with self.assertRaises(ValueError):
|
|
scan.Settings.from_env([])
|
|
|
|
def test_settings_rejects_negative_max_findings(self) -> None:
|
|
with mock.patch.dict(os.environ, {"MAX_FINDINGS_IN_BODY": "-1"}, clear=False):
|
|
with self.assertRaises(ValueError):
|
|
scan.Settings.from_env([])
|
|
|
|
def test_env_bool_rejects_garbage(self) -> None:
|
|
with mock.patch.dict(os.environ, {"SOME_FLAG": "maybe"}):
|
|
with self.assertRaises(ValueError):
|
|
scan.env_bool("SOME_FLAG", False)
|
|
|
|
def test_env_bool_accepts_common_spellings(self) -> None:
|
|
for raw, expected in (("1", True), ("YES", True), ("off", False), ("0", False)):
|
|
with mock.patch.dict(os.environ, {"SOME_FLAG": raw}):
|
|
self.assertIs(scan.env_bool("SOME_FLAG", not expected), expected)
|
|
|
|
|
|
REPORT = {
|
|
"Results": [
|
|
{
|
|
"Vulnerabilities": [
|
|
{"Severity": "LOW", "VulnerabilityID": "CVE-3", "PkgName": "zlib"},
|
|
{"Severity": "HIGH", "VulnerabilityID": "CVE-2", "PkgName": "openssl",
|
|
"InstalledVersion": "3.0.1", "FixedVersion": "3.0.2"},
|
|
{"Severity": "CRITICAL", "VulnerabilityID": "CVE-1", "PkgName": "libc",
|
|
"InstalledVersion": "1.0", "FixedVersion": "1.1"},
|
|
{"Severity": "MEDIUM", "VulnerabilityID": "CVE-4", "PkgName": "curl"},
|
|
{"Severity": "HIGH", "VulnerabilityID": "CVE-5", "PkgName": "bash"},
|
|
]
|
|
},
|
|
{"Vulnerabilities": None},
|
|
{},
|
|
]
|
|
}
|
|
|
|
|
|
class ReportTests(unittest.TestCase):
|
|
def test_summarise_counts_each_severity(self) -> None:
|
|
result = scan.summarise(REPORT)
|
|
self.assertEqual(result.counts, {"CRITICAL": 1, "HIGH": 2, "MEDIUM": 1, "LOW": 1})
|
|
|
|
def test_summarise_counts_fixable_high_and_critical_only(self) -> None:
|
|
self.assertEqual(scan.summarise(REPORT).fixable, 2)
|
|
|
|
def test_summarise_sorts_critical_first(self) -> None:
|
|
findings = scan.summarise(REPORT).findings
|
|
self.assertEqual([f["VulnerabilityID"] for f in findings], ["CVE-1", "CVE-2", "CVE-5"])
|
|
|
|
def test_summarise_handles_empty_report(self) -> None:
|
|
result = scan.summarise({})
|
|
self.assertEqual(result.counts, {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0})
|
|
self.assertEqual(result.findings, [])
|
|
|
|
def test_threshold_hits_respects_rank(self) -> None:
|
|
result = scan.summarise(REPORT)
|
|
self.assertEqual(scan.threshold_hits(result, 0), 0)
|
|
self.assertEqual(scan.threshold_hits(result, scan.severity_rank("CRITICAL")), 1)
|
|
self.assertEqual(scan.threshold_hits(result, scan.severity_rank("HIGH")), 3)
|
|
self.assertEqual(scan.threshold_hits(result, scan.severity_rank("LOW")), 5)
|
|
|
|
def test_build_body_truncates_findings(self) -> None:
|
|
result = scan.summarise(REPORT)
|
|
body = scan.build_body("nginx:1.27", result, "/reports/x.json", max_findings=2)
|
|
self.assertIn("showing 2 of 3", body)
|
|
self.assertIn("CVE-1", body)
|
|
self.assertIn("CVE-2", body)
|
|
self.assertNotIn("CVE-5", body)
|
|
self.assertIn("/reports/x.json", body)
|
|
|
|
def test_build_body_lists_all_findings_when_under_limit(self) -> None:
|
|
result = scan.summarise(REPORT)
|
|
body = scan.build_body("nginx:1.27", result, "/reports/x.json", max_findings=15)
|
|
self.assertNotIn("showing", body)
|
|
self.assertIn("CVE-5", body)
|
|
self.assertIn("no fix available", body)
|
|
|
|
def test_build_body_includes_container_detail(self) -> None:
|
|
result = scan.summarise(REPORT)
|
|
body = scan.build_body(
|
|
"nginx:1.27",
|
|
result,
|
|
"/reports/x.json",
|
|
max_findings=15,
|
|
container_detail="Containers: web, web-worker",
|
|
)
|
|
self.assertIn("Image: nginx:1.27", body)
|
|
self.assertIn("Containers: web, web-worker", body)
|
|
|
|
def test_aggregated_body_truncates_large_notifications(self) -> None:
|
|
aggregated = scan.AggregatedReport()
|
|
for i in range(500):
|
|
vuln = {
|
|
"Severity": "HIGH",
|
|
"VulnerabilityID": f"CVE-2026-{i:05d}",
|
|
"PkgName": "libc",
|
|
"InstalledVersion": "1.0",
|
|
"FixedVersion": "1.1",
|
|
}
|
|
aggregated.add_vulnerability(vuln, f"img{i}:latest", [f"c{i}"])
|
|
body = scan.build_aggregated_body(aggregated, 200)
|
|
self.assertLessEqual(len(body), scan.MAX_NOTIFICATION_CHARS)
|
|
|
|
def test_aggregated_body_compact_for_discord(self) -> None:
|
|
aggregated = scan.AggregatedReport()
|
|
for i in range(50):
|
|
vuln = {
|
|
"Severity": "CRITICAL",
|
|
"VulnerabilityID": f"CVE-2026-{i:05d}",
|
|
"PkgName": "libc",
|
|
"InstalledVersion": "1.0",
|
|
"FixedVersion": "1.1",
|
|
}
|
|
aggregated.add_vulnerability(vuln, f"img{i}:latest", [f"c{i}"])
|
|
body = scan.build_aggregated_body(aggregated, 50, compact=True)
|
|
self.assertLessEqual(len(body), scan.DISCORD_MAX_NOTIFICATION_CHARS)
|
|
|
|
def test_build_aggregated_body_groups_by_cve(self) -> None:
|
|
aggregated = scan.AggregatedReport()
|
|
vuln1 = {
|
|
"Severity": "CRITICAL",
|
|
"VulnerabilityID": "CVE-1",
|
|
"PkgName": "libc",
|
|
"InstalledVersion": "1.0",
|
|
"FixedVersion": "1.1",
|
|
}
|
|
vuln2 = {
|
|
"Severity": "HIGH",
|
|
"VulnerabilityID": "CVE-1",
|
|
"PkgName": "libc",
|
|
"InstalledVersion": "1.0",
|
|
"FixedVersion": "1.1",
|
|
}
|
|
aggregated.add_vulnerability(vuln1, "nginx:1.27", ["web"])
|
|
aggregated.add_vulnerability(vuln2, "redis:7", ["cache"])
|
|
body = scan.build_aggregated_body(aggregated, 15)
|
|
self.assertIn("CVE-1", body)
|
|
self.assertIn("web", body)
|
|
self.assertIn("nginx:1.27", body)
|
|
self.assertIn("cache", body)
|
|
self.assertIn("redis:7", body)
|
|
|
|
def test_cve_entries_split_fixable_and_unfixable(self) -> None:
|
|
entries = scan.cve_entries(scan.summarise(REPORT), scan.severity_rank("HIGH"))
|
|
self.assertEqual(scan.split_counts(entries), (3, 2, 1))
|
|
self.assertEqual([entry.cve for entry in entries], ["CVE-1", "CVE-2", "CVE-5"])
|
|
self.assertEqual(scan.severity_counts(entries), {
|
|
"CRITICAL": 1,
|
|
"HIGH": 2,
|
|
"MEDIUM": 0,
|
|
"LOW": 0,
|
|
})
|
|
|
|
def test_build_important_body_lists_every_cve(self) -> None:
|
|
result = scan.summarise(REPORT)
|
|
target = scan.ScanTarget(image="nginx:1.27", containers=["web"])
|
|
body = scan.build_important_body(
|
|
"web", target, result, "/reports/x.json", "HIGH"
|
|
)
|
|
self.assertIn("Image: nginx:1.27", body)
|
|
self.assertIn("Container: web", body)
|
|
self.assertIn("Total: 3 (fixable: 2, unfixable: 1)", body)
|
|
self.assertIn("CVE-1", body)
|
|
self.assertIn("CVE-2", body)
|
|
self.assertIn("CVE-5", body)
|
|
self.assertIn("no fix available", body)
|
|
self.assertIn("/reports/x.json", body)
|
|
|
|
def test_build_important_body_respects_notification_threshold(self) -> None:
|
|
result = scan.summarise(REPORT)
|
|
target = scan.ScanTarget(image="nginx:1.27", containers=["web"])
|
|
body = scan.build_important_body(
|
|
"web", target, result, "/reports/x.json", "CRITICAL"
|
|
)
|
|
self.assertIn("CVE-1", body)
|
|
self.assertNotIn("CVE-2", body)
|
|
self.assertNotIn("CVE-5", body)
|
|
self.assertIn("Total: 1 (fixable: 1, unfixable: 0)", body)
|
|
|
|
def test_build_other_containers_body_has_counts_per_container(self) -> None:
|
|
body = scan.build_other_containers_body(
|
|
[
|
|
("nginx:1.27", "web-worker", 3, 2, 1),
|
|
("redis:7", "cache", 4, 0, 4),
|
|
]
|
|
)
|
|
self.assertIn("* nginx:1.27 [web-worker]: 3 total (2 fixable, 1 unfixable)", body)
|
|
self.assertIn("* redis:7 [cache]: 4 total (0 fixable, 4 unfixable)", body)
|
|
self.assertIn("Total: 7 CVE(s) in 2 container(s)", body)
|
|
|
|
def test_build_missing_body_separates_missing_and_unscanned(self) -> None:
|
|
body = scan.build_missing_body(["gone"], ["web"])
|
|
self.assertIn("not running", body)
|
|
self.assertIn("* gone", body)
|
|
self.assertIn("image was not scanned", body)
|
|
self.assertIn("* web", body)
|
|
|
|
def test_split_body_keeps_every_line(self) -> None:
|
|
body = "\n".join(f"CVE-{index}" for index in range(10))
|
|
parts = scan._split_body(body, 20)
|
|
self.assertGreater(len(parts), 1)
|
|
self.assertTrue(all(len(part) <= 20 for part in parts))
|
|
self.assertEqual("\n".join(parts), body)
|
|
|
|
|
|
class TrivyCommandTests(unittest.TestCase):
|
|
def test_run_trivy_passes_vex_sources_in_order(self) -> None:
|
|
settings = scan.Settings(
|
|
explicit_images=[], trivy_vex=["/vex/app.json", "repo"]
|
|
)
|
|
completed = subprocess.CompletedProcess([], 0, stdout="", stderr="")
|
|
with mock.patch.object(scan.subprocess, "run", return_value=completed) as run:
|
|
status, _ = scan.run_trivy("nginx:1.27", settings)
|
|
|
|
command = run.call_args.args[0]
|
|
self.assertEqual(status, 0)
|
|
positions = [index for index, arg in enumerate(command) if arg == "--vex"]
|
|
self.assertEqual(
|
|
[command[index + 1] for index in positions],
|
|
["/vex/app.json", "repo"],
|
|
)
|
|
|
|
def test_run_trivy_omits_vex_when_not_configured(self) -> None:
|
|
settings = scan.Settings(explicit_images=[])
|
|
completed = subprocess.CompletedProcess([], 0, stdout="", stderr="")
|
|
with mock.patch.object(scan.subprocess, "run", return_value=completed) as run:
|
|
scan.run_trivy("nginx:1.27", settings)
|
|
|
|
self.assertNotIn("--vex", run.call_args.args[0])
|
|
|
|
|
|
class MainFlowTests(unittest.TestCase):
|
|
"""End-to-end runs of main() with fake trivy/apprise binaries."""
|
|
|
|
TRIVY_SCRIPT = """#!/usr/bin/env python3
|
|
import json, os, sys
|
|
args = sys.argv[1:]
|
|
if os.environ.get("FAKE_TRIVY_FAIL") == "1":
|
|
print("boom", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
report_path = args[args.index("--output") + 1]
|
|
image = args[-1]
|
|
with open(os.environ["FAKE_TRIVY_CALLS"], "a") as log:
|
|
log.write(image + "\\n")
|
|
severity = os.environ.get("FAKE_TRIVY_SEVERITY", "CRITICAL")
|
|
report = {"Results": [{"Vulnerabilities": [{
|
|
"Severity": severity,
|
|
"VulnerabilityID": "CVE-2026-0001",
|
|
"PkgName": "libdemo",
|
|
"InstalledVersion": "1.0.0",
|
|
"FixedVersion": "1.0.1",
|
|
}]}]}
|
|
with open(report_path, "w") as handle:
|
|
json.dump(report, handle)
|
|
"""
|
|
|
|
APPRISE_SCRIPT = """#!/bin/sh
|
|
printf '%s\\n' "$*" >> "$APPRISE_LOG"
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
self.workdir = tempfile.mkdtemp(prefix="scan-e2e-")
|
|
self.bin_dir = os.path.join(self.workdir, "bin")
|
|
os.makedirs(self.bin_dir)
|
|
self.trivy_calls = os.path.join(self.workdir, "trivy-calls.log")
|
|
self.apprise_log = os.path.join(self.workdir, "apprise.log")
|
|
Path(self.apprise_log).write_text("")
|
|
|
|
for name, script in (
|
|
("trivy", self.TRIVY_SCRIPT),
|
|
("apprise", self.APPRISE_SCRIPT),
|
|
):
|
|
path = os.path.join(self.bin_dir, name)
|
|
Path(path).write_text(script)
|
|
os.chmod(path, 0o755)
|
|
|
|
self.env = {
|
|
"PATH": self.bin_dir + os.pathsep + os.environ.get("PATH", ""),
|
|
"REPORT_DIR": os.path.join(self.workdir, "reports"),
|
|
"TRIVY_CACHE_DIR": os.path.join(self.workdir, "cache"),
|
|
"IMAGES": "",
|
|
"IMAGES_FILE": "",
|
|
"APPRISE_URLS": "json://localhost",
|
|
"APPRISE_CONFIG": "",
|
|
"FAIL_ON": "HIGH",
|
|
"NOTIFY_MIN_SEVERITY": "HIGH",
|
|
"NOTIFY_ON_SCAN_ERROR": "true",
|
|
"MAX_FINDINGS_IN_BODY": "15",
|
|
"IGNORE_UNFIXED": "",
|
|
"TRIVY_SEVERITIES": "",
|
|
"FAKE_TRIVY_CALLS": self.trivy_calls,
|
|
"APPRISE_LOG": self.apprise_log,
|
|
"LOG_LEVEL": "WARN",
|
|
}
|
|
|
|
def tearDown(self) -> None:
|
|
shutil.rmtree(self.workdir, ignore_errors=True)
|
|
|
|
def run_main(self, argv: list[str], docker_socket: str, **extra_env: str) -> int:
|
|
env = dict(self.env, DOCKER_SOCK=docker_socket, DOCKER_HOST="", **extra_env)
|
|
with mock.patch.dict(os.environ, env, clear=True):
|
|
return scan.main(argv)
|
|
|
|
def test_scans_images_from_running_containers(self) -> None:
|
|
containers = [container("a" * 64, "testphp-php")]
|
|
with FakeDocker(containers) as docker:
|
|
code = self.run_main([], docker.socket_path)
|
|
|
|
self.assertEqual(code, 1) # CRITICAL finding meets FAIL_ON=HIGH
|
|
self.assertEqual(Path(self.trivy_calls).read_text().split(), ["testphp-php"])
|
|
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_notification_includes_container(self) -> None:
|
|
containers = [container("a" * 64, "testphp-php", name="testphp")]
|
|
with FakeDocker(containers) as docker:
|
|
code = self.run_main([], docker.socket_path)
|
|
|
|
self.assertEqual(code, 1)
|
|
notification = Path(self.apprise_log).read_text()
|
|
self.assertIn("CVE-2026-0001", notification)
|
|
self.assertIn("testphp", notification)
|
|
|
|
def test_important_container_gets_full_list_and_others_get_summary(self) -> None:
|
|
containers = [
|
|
container("a" * 64, "testphp-php", name="web"),
|
|
container("b" * 64, "redis:7", name="cache"),
|
|
]
|
|
with FakeDocker(containers) as docker:
|
|
code = self.run_main(
|
|
[], docker.socket_path, IMPORTANT_CONTAINERS="web"
|
|
)
|
|
|
|
self.assertEqual(code, 1)
|
|
notification = Path(self.apprise_log).read_text()
|
|
self.assertIn("Vulnerability scan: web (testphp-php)", notification)
|
|
self.assertIn("Container: web", notification)
|
|
self.assertIn("CVE-2026-0001", notification)
|
|
self.assertIn("Vulnerability scan: other containers", notification)
|
|
self.assertIn("redis:7 [cache]", notification)
|
|
self.assertIn("(1 fixable, 0 unfixable)", notification)
|
|
# The important container must not also appear in the summary.
|
|
self.assertNotIn("testphp-php [web]", notification)
|
|
|
|
def test_missing_important_container_is_reported(self) -> None:
|
|
containers = [container("a" * 64, "testphp-php", name="web")]
|
|
with FakeDocker(containers) as docker:
|
|
code = self.run_main(
|
|
[], docker.socket_path, IMPORTANT_CONTAINERS="web stale"
|
|
)
|
|
|
|
self.assertEqual(code, 1)
|
|
notification = Path(self.apprise_log).read_text()
|
|
self.assertIn("Vulnerability scan: important containers", notification)
|
|
self.assertIn("Important containers are not running:", notification)
|
|
self.assertIn("* stale", notification)
|
|
|
|
def test_important_container_with_unscanned_image_is_reported(self) -> None:
|
|
containers = [container("a" * 64, "other:1", name="web")]
|
|
with FakeDocker(containers) as docker:
|
|
code = self.run_main(
|
|
["nginx:1.27"], docker.socket_path, IMPORTANT_CONTAINERS="web"
|
|
)
|
|
|
|
self.assertEqual(code, 1)
|
|
notification = Path(self.apprise_log).read_text()
|
|
self.assertIn("image was not scanned", notification)
|
|
self.assertIn("* web", notification)
|
|
|
|
def test_scan_failure_notification_includes_container(self) -> None:
|
|
containers = [container("a" * 64, "testphp-php", name="testphp")]
|
|
with FakeDocker(containers) as docker:
|
|
code = self.run_main([], docker.socket_path, FAKE_TRIVY_FAIL="1")
|
|
|
|
self.assertEqual(code, 2)
|
|
notification = Path(self.apprise_log).read_text()
|
|
self.assertIn("Scan failed: testphp-php [container: testphp]", notification)
|
|
self.assertIn("Container: testphp", notification)
|
|
self.assertIn("Trivy could not complete the scan", notification)
|
|
|
|
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")
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(Path(self.apprise_log).read_text(), "")
|
|
|
|
def test_explicit_image_skips_discovery(self) -> None:
|
|
with FakeDocker([container("a" * 64, "testphp-php")]) as docker:
|
|
code = self.run_main(["nginx:1.27"], docker.socket_path, FAIL_ON="NONE")
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(Path(self.trivy_calls).read_text().split(), ["nginx:1.27"])
|
|
|
|
def test_scan_failure_exits_two(self) -> None:
|
|
with FakeDocker([container("a" * 64, "testphp-php")]) as docker:
|
|
code = self.run_main([], docker.socket_path, FAKE_TRIVY_FAIL="1", NOTIFY_ON_SCAN_ERROR="false")
|
|
|
|
self.assertEqual(code, 2)
|
|
self.assertEqual(Path(self.apprise_log).read_text(), "")
|
|
|
|
def test_scan_failure_notifies(self) -> None:
|
|
with FakeDocker([container("a" * 64, "testphp-php")]) as docker:
|
|
code = self.run_main([], docker.socket_path, FAKE_TRIVY_FAIL="1")
|
|
|
|
self.assertEqual(code, 2)
|
|
self.assertIn("testphp-php", Path(self.apprise_log).read_text())
|
|
|
|
def test_nothing_to_scan_exits_zero(self) -> None:
|
|
with FakeDocker([]) as docker:
|
|
code = self.run_main([], docker.socket_path)
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertFalse(os.path.exists(self.trivy_calls))
|
|
|
|
def test_only_high_findings_fail_at_high_threshold(self) -> None:
|
|
with FakeDocker([container("a" * 64, "testphp-php")]) as docker:
|
|
code = self.run_main([], docker.socket_path, FAKE_TRIVY_SEVERITY="HIGH")
|
|
|
|
self.assertEqual(code, 1)
|
|
|
|
def test_medium_findings_pass_high_threshold(self) -> None:
|
|
with FakeDocker([container("a" * 64, "testphp-php")]) as docker:
|
|
code = self.run_main([], docker.socket_path, FAKE_TRIVY_SEVERITY="MEDIUM")
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(Path(self.apprise_log).read_text(), "")
|
|
|
|
def test_discovers_rootless_socket_without_configuration(self) -> None:
|
|
"""No DOCKER_SOCK: the socket is found under /run/user (as Compose mounts it)."""
|
|
env = dict(self.env, DOCKER_SOCK="", DOCKER_HOST="", FAIL_ON="NONE")
|
|
with FakeDocker([container("a" * 64, "testphp-php")]) as docker:
|
|
with mock.patch.dict(os.environ, env, clear=True):
|
|
with mock.patch.object(scan.glob, "glob", return_value=[docker.socket_path]):
|
|
code = scan.main([])
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(Path(self.trivy_calls).read_text().split(), ["testphp-php"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2) |