Add allowed_image_name and use it to verify the run function.

host_modules/docker_service.py

@@ -43,6 +43,22 @@ def is_allowed_container(container):
             return True
     return False
 
+def is_allowed_image(image):
+    """
+    Check if the image is allowed to be managed by this service.
+
+    Args:
+        image (str): The image name.
+
+    Returns:
+        bool: True if the image is allowed, False otherwise.
+    """
+    image_name = image.split(":")[0]  # Remove tag if present
+    for allowed_image, _ in ALLOWED_CONTAINERS:
+        if image_name == allowed_image:
+            return True
+    return False
+
 
 class DockerService(host_service.HostModule):
     """
@@ -159,30 +175,20 @@ def run(self, image, command, kwargs):
         """
         try:
             client = docker.from_env()
-            image_list = client.images.list(all=True)
-
-            def _validate_image_name(image_name):
-                base_name_list = list(
-                    set(
-                        image.tags[0].split(":")[0]
-                        for image in image_list
-                        if image.tags
-                    )
+
+            if not is_allowed_image(image):
+                return (
+                    errno.EPERM,
+                    "Image {} is not allowed to be managed by this service.".format(
+                        image
+                    ),
                 )
-                base_name = image_name.split(":")[0]
-                if base_name in base_name_list:
-                    return True
-                return False
-
-            if not _validate_image_name(image):
-                return errno.EPERM, "Image {} is not allowed.".format(image)
 
-            def _validate_command(command):
-                # Only allow empty command.
-                return command == ""
-
-            if not _validate_command(command):
-                return errno.EPERM, "Command {} is not allowed.".format(command)
+            if command:
+                return (
+                    errno.EPERM,
+                    "Only empty command is allowed to be managed by this service."
+                )
 
             # Semgrep cannot detect codes for validating image and command.
             # nosemgrep: python.docker.security.audit.docker-arbitrary-container-run.docker-arbitrary-container-run

tests/host_modules/docker_service_test.py

@@ -213,48 +213,45 @@ def test_docker_restart_fail_api_error(self, MockInit, MockBusName, MockSystemBu
     @mock.patch("dbus.service.Object.__init__")
     def test_docker_run_success(self, MockInit, MockBusName, MockSystemBus):
         mock_docker_client = mock.Mock()
-        mock_docker_client.images.list.return_value = [mock.Mock(tags=["image_name:tag"])]
-        mock_docker_client.containers.run.return_value.name = "container_name"
+        mock_docker_client.containers.run.return_value.name = "syncd"
 
         with mock.patch.object(docker, "from_env", return_value=mock_docker_client):
             docker_service = DockerService(MOD_NAME)
-            rc, msg = docker_service.run("image_name", "command", {})
+            rc, msg = docker_service.run("docker-syncd-brcm:latest", "", {})
 
         assert rc == 0, "Return code is wrong"
         assert "started" in msg, "Message should contain 'started'"
-        mock_docker_client.containers.run.assert_called_once_with("image_name", "command", **{})
+        mock_docker_client.containers.run.assert_called_once_with("docker-syncd-brcm:latest", "", **{})
 
     @mock.patch("dbus.SystemBus")
     @mock.patch("dbus.service.BusName")
     @mock.patch("dbus.service.Object.__init__")
     def test_docker_run_fail_image_not_found(self, MockInit, MockBusName, MockSystemBus):
         mock_docker_client = mock.Mock()
-        mock_docker_client.images.list.return_value = [mock.Mock(tags=["image_name:tag"])]
         mock_docker_client.containers.run.side_effect = docker.errors.ImageNotFound("Image not found")
 
         with mock.patch.object(docker, "from_env", return_value=mock_docker_client):
             docker_service = DockerService(MOD_NAME)
-            rc, msg = docker_service.run("image_name", "command", {})
+            rc, msg = docker_service.run("docker-syncd-brcm:latest", "", {})
 
         assert rc == errno.ENOENT, "Return code is wrong"
         assert "not found" in msg, "Message should contain 'not found'"
-        mock_docker_client.containers.run.assert_called_once_with("image_name", "command", **{})
+        mock_docker_client.containers.run.assert_called_once_with("docker-syncd-brcm:latest", "", **{})
 
     @mock.patch("dbus.SystemBus")
     @mock.patch("dbus.service.BusName")
     @mock.patch("dbus.service.Object.__init__")
     def test_docker_run_fail_api_error(self, MockInit, MockBusName, MockSystemBus):
         mock_docker_client = mock.Mock()
-        mock_docker_client.images.list.return_value = [mock.Mock(tags=["image_name:tag"])]
         mock_docker_client.containers.run.side_effect = docker.errors.APIError("API error")
 
         with mock.patch.object(docker, "from_env", return_value=mock_docker_client):
             docker_service = DockerService(MOD_NAME)
-            rc, msg = docker_service.run("image_name", "command", {})
+            rc, msg = docker_service.run("docker-syncd-brcm:latest", "", {})
 
         assert rc != 0, "Return code is wrong"
         assert "API error" in msg, "Message should contain 'API error'"
-        mock_docker_client.containers.run.assert_called_once_with("image_name", "command", **{})
+        mock_docker_client.containers.run.assert_called_once_with("docker-syncd-brcm:latest", "", **{})
 
     @mock.patch("dbus.SystemBus")
     @mock.patch("dbus.service.BusName")