add container info

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")