Quote arguments in the NNI trial command

monai/apps/auto3dseg/auto_runner.py

@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import os
+import shlex
 import shutil
 import warnings
 from copy import deepcopy
@@ -37,6 +38,28 @@
 nni, has_nni = optional_import("nni")
 
 
+def _nni_trial_command(obj_filename: str, work_dir: str) -> str:
+    """
+    Build the NNI ``trialCommand`` that runs one algo trial.
+
+    NNI runs ``trialCommand`` through the platform shell, so each argument is quoted for that shell
+    and paths containing spaces or shell metacharacters are passed through unchanged.
+
+    Args:
+        obj_filename: the serialized algo object file.
+        work_dir: the Auto3DSeg working directory.
+
+    Returns:
+        the command string to store as ``trialCommand``.
+    """
+    args = ["python", "-m", "monai.apps.auto3dseg", "NNIGen", "run_algo", str(obj_filename), str(work_dir)]
+    if os.name == "nt":
+        from subprocess import list2cmdline
+
+        return list2cmdline(args)
+    return shlex.join(args)
+
+
 class AutoRunner:
     """
     An interface for handling Auto3Dseg with minimal inputs and understanding of the internal states in Auto3Dseg.
@@ -797,7 +820,7 @@ def _train_algo_in_nni(self, history: list[dict[str, Any]]) -> None:
                     nni_config[key] = self.hpo_params[key]
             nni_config.update({"experimentName": name})
             nni_config.update({"search_space": self.search_space})
-            trial_cmd = "python -m monai.apps.auto3dseg NNIGen run_algo " + obj_filename + " " + self.work_dir
+            trial_cmd = _nni_trial_command(obj_filename, self.work_dir)
             nni_config.update({"trialCommand": trial_cmd})
             nni_config_filename = os.path.abspath(os.path.join(self.work_dir, f"{name}_nni_config.yaml"))
             ConfigParser.export_config_file(nni_config, nni_config_filename, fmt="yaml", default_flow_style=None)

monai/apps/auto3dseg/hpo_gen.py

@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import os
+import shlex
 from abc import abstractmethod
 from copy import deepcopy
 from typing import Any, cast
@@ -143,7 +144,7 @@ def print_bundle_algo_instruction(self):
         logger.info("=" * 140)
         logger.info("If NNI will run in your local env: ")
         logger.info("1. Add the following line to the trialCommand in your NNI config: ")
-        logger.info(f"{hint} {self.obj_filename} {{result_dir}}")
+        logger.info(f"{hint} {shlex.quote(str(self.obj_filename))} {{result_dir}}")
         logger.info("-" * 140)
         logger.info("If NNI will run in a remote env: ")
         logger.info(

tests/auto3dseg/test_nni_trial_command.py

@@ -0,0 +1,52 @@
+# Copyright (c) MONAI Consortium
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import os
+import shlex
+import subprocess
+import tempfile
+import unittest
+
+from monai.apps.auto3dseg.auto_runner import _nni_trial_command
+
+PREFIX = ["python", "-m", "monai.apps.auto3dseg", "NNIGen", "run_algo"]
+
+
+@unittest.skipIf(os.name == "nt", "POSIX shell quoting")
+class TestNNITrialCommand(unittest.TestCase):
+    def test_plain_paths(self):
+        cmd = _nni_trial_command("/work/algo_0/algo_object.json", "/work")
+        self.assertEqual(cmd, "python -m monai.apps.auto3dseg NNIGen run_algo /work/algo_0/algo_object.json /work")
+
+    def test_paths_with_special_characters_are_single_arguments(self):
+        for work_dir in ("/tmp/my work", "/tmp/a;b", "/tmp/$(echo x)", "/tmp/`x`", "/tmp/a'b\"c", "/tmp/a|b&c"):
+            with self.subTest(work_dir=work_dir):
+                cmd = _nni_trial_command(os.path.join(work_dir, "algo_object.json"), work_dir)
+                self.assertEqual(shlex.split(cmd), PREFIX + [os.path.join(work_dir, "algo_object.json"), work_dir])
+
+    def test_shell_receives_the_literal_path(self):
+        """Run the command through ``sh`` with ``python`` swapped for ``printf`` and check the arguments it sees."""
+        with tempfile.TemporaryDirectory() as tempdir:
+            marker = os.path.join(tempdir, "marker")
+            work_dir = f"{tempdir}/w; touch {marker}; $(touch {marker})"
+            cmd = _nni_trial_command("obj.json", work_dir)
+            self.assertTrue(cmd.startswith("python "))
+            out = subprocess.run(
+                ["sh", "-c", "printf '%s\\n' " + cmd[len("python ") :]], capture_output=True, text=True, check=True
+            ).stdout
+            self.assertEqual(out.splitlines(), PREFIX[1:] + ["obj.json", work_dir])
+            self.assertFalse(os.path.exists(marker))
+
+
+if __name__ == "__main__":
+    unittest.main()