Further expand typing in ImageBuilder

backend/engine/oci/builder.py

@@ -70,8 +70,11 @@ def temporary_builder(name_prefix: str):
         ["docker", "buildx", "create", "--name", name],
         capture_output=True,
     )
+    errout = create_proc.stderr.decode("utf-8").strip()
     if create_proc.returncode != 0:
-        raise DockerException(f"Failed to create builder: {create_proc.stderr.decode('utf-8')}")
+        raise DockerException(f"Failed to create builder: {errout}")
+    elif errout != "":
+        log.warning("Docker builder create warning: %s", errout)
 
     try:
         yield name
@@ -82,12 +85,15 @@ def temporary_builder(name_prefix: str):
             ["docker", "buildx", "rm", "--builder", name, "-f"],
             capture_output=True,
         )
+        errout = rm_proc.stderr.decode("utf-8").strip()
         if rm_proc.returncode != 0:
-            log.error("Failed to remove builder %s: %s", name, rm_proc.stderr.decode("utf-8"))
+            log.error("Failed to remove builder %s: %s", name, errout)
+        elif errout != "":
+            log.warning("Docker builder remove warning: %s", errout)
 
 
 class ImageBuilder:
-    def __init__(self, path, repo_name, ignore_prefixes, engine_id):
+    def __init__(self, path: str, repo_name: str, ignore_prefixes: list[str] | None, engine_id: str):
         """
         Finds and builds any docker images within the path.
         :param path: path to images, typically the root of the repo.
@@ -97,7 +103,7 @@ def __init__(self, path, repo_name, ignore_prefixes, engine_id):
         """
         self.path = path
         self.repo_name = repo_name
-        self.ignore_prefixes = ignore_prefixes
+        self.ignore_prefixes = ignore_prefixes or []
         self.engine_id = engine_id
 
     def find_dockerfiles(self) -> list[str]:
@@ -161,8 +167,11 @@ def build_local_image(self, dockerfile: str, tag: str, builder: str | None = Non
         cmd += ["--pull", "--no-cache", "--force-rm", "-q", ".", "-f", dockerfile, "-t", tag_id]
 
         build_proc = subprocess.run(cmd, capture_output=True, cwd=self.path)
-        if build_proc.returncode != 0:
-            log.warning(build_proc.stderr.decode("utf-8"))
+
+        # Always log stderr, even if build succeeded.
+        # Since we invoke the build via the CLI, we want to capture CLI warnings.
+        if errout := build_proc.stderr.decode("utf-8").strip() != "":
+            log.warning("Docker build warnings: %s", errout)
 
         status = build_proc.returncode == 0
         log.info("Built %s from %s (success: %s)", tag_id, dockerfile_name, status)
@@ -178,9 +187,7 @@ def untag_base_images(self) -> None:
         pulled from the Artemis ECR should be excluded from this and are identified by the passed in prefix.
         :return: None
         """
-        if self.ignore_prefixes is None:
-            self.ignore_prefixes = []
-        to_remove = []
+        to_remove: list[str] = []
 
         # Get a list of all the image:tag and digests
         r = subprocess.run(
@@ -216,7 +223,7 @@ def untag_base_images(self) -> None:
                 # Log the error but keep going
                 log.error(r.stderr.decode("utf-8"))
 
-    def private_docker_repos_login(self, files) -> None:
+    def private_docker_repos_login(self, files: list[str]) -> None:
         """
         Gets Private Docker Repo Config/Credentials from Secrets Manager and login to the Docker Repo if needed.
         :param files: List of Dockerfiles to check
@@ -253,7 +260,7 @@ def private_docker_repos_login(self, files) -> None:
             else:
                 log.info("No Dockerfiles depend on %s", repo["url"])
 
-    def docker_login_needed(self, files: list, search: str, url: str) -> bool:
+    def docker_login_needed(self, files: list[str], search: str, url: str) -> bool:
         """
         Determine if any Dockerfiles in the list depend on the private repo
         :param files: List of Dockerfiles to check

backend/engine/oci/remover.py

@@ -1,35 +1,34 @@
-import subprocess
+import docker
+import docker.errors
 
 from artemislib.logging import Logger
 
 logger = Logger("oci_remover")
 
+docker_client = docker.from_env()
+
 
 def remove_docker_image(tag: str) -> bool:
     """
     Delete a container image from disk after the scan job is complete.
     :param tag: Image tag.
     :return: True if successful.
     """
-    logger.info("Removing image %s", tag)
-    return (
-        subprocess.run(
-            ["docker", "image", "rm", "-f", tag],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            check=False,
-        ).returncode
-        == 0
-    )
+    logger.info("Removing image: %s", tag)
+    try:
+        docker_client.images.remove(image=tag, force=True)
+    except docker.errors.DockerException:
+        logger.warning("Failed to remove image: %s", tag, exc_info=True)
+        return False
+    return True
 
 
 def prune_images() -> None:
     """
     Prune the docker images. This removes all the dangling images.
     """
     logger.info("Cleaning up unused images")
-    r = subprocess.run(
-        ["docker", "image", "prune", "--force"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False
-    )
-    if r.returncode != 0:
-        logger.error(r.stderr.decode("utf-8"))
+    try:
+        docker_client.images.prune()
+    except docker.errors.DockerException:
+        logger.warning("Failed to prune images", exc_info=True)

backend/engine/tests/test_docker_builder.py

@@ -92,12 +92,16 @@ def setUp(self) -> None:
             self.demo_results_dict = json.load(output_file)
 
     def test_find_dockerfiles(self):
-        image_builder = builder.ImageBuilder(os.path.join(TEST_ROOT, "Dockerfiles"), None, None, None)
+        image_builder = builder.ImageBuilder(
+            os.path.join(TEST_ROOT, "Dockerfiles"), "test_repo", None, "test_engine_id"
+        )
         result = image_builder.find_dockerfiles()
         self.assertGreater(len(result), 5)
 
     def test_private_docker_repos_login(self):
-        image_builder = builder.ImageBuilder(os.path.join(TEST_ROOT, "Dockerfiles"), None, None, None)
+        image_builder = builder.ImageBuilder(
+            os.path.join(TEST_ROOT, "Dockerfiles"), "test_repo", None, "test_engine_id"
+        )
         with patch("plugins.lib.utils.get_secret_with_status") as mock_get_secret_with_status:
             with patch("oci.builder.ImageBuilder.docker_login_needed") as mock_docker_login_needed:
                 mock_get_secret_with_status.return_value = TEST_GET_SECRET_WITH_STATUS_MOCK_OUTPUT
@@ -108,11 +112,11 @@ def test_private_docker_repos_login(self):
                 with patch("plugins.lib.utils.docker_login") as mock_docker_login:
                     mock_docker_login.return_value = True
 
-                    image_builder.private_docker_repos_login(os.path.join(TEST_ROOT, "Dockerfiles"))
+                    image_builder.private_docker_repos_login([os.path.join(TEST_ROOT, "Dockerfiles")])
                     mock_get_secret_with_status.assert_called_once()
 
                     mock_docker_login_needed.assert_called_once_with(
-                        os.path.join(TEST_ROOT, "Dockerfiles"),
+                        [os.path.join(TEST_ROOT, "Dockerfiles")],
                         TEST_PRIVATE_DOCKER_REPOS_CONFIGS[0]["search"],
                         TEST_PRIVATE_DOCKER_REPOS_CONFIGS[0]["url"],
                     )