Initial commit

tests/test_scan.py

@@ -0,0 +1,452 @@
+"""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 json
+import os
+import shutil
+import socketserver
+import sys
+import tempfile
+import threading
+import unittest
+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") -> dict:
+    return {"Id": container_id, "Image": image, "ImageID": image_id}
+
+
+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_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_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"]):
+                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=[]):
+                self.assertEqual(scan.discover_images(), ([], None))
+
+
+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_images") 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")):
+            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)):
+            self.assertEqual(scan.collect_images(settings), [])
+
+
+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",
+        }
+        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)
+
+    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)
+
+
+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_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)