Quote arguments in the NNI trial command

monai/apps/auto3dseg/auto_runner.py

@@ -23,7 +23,7 @@
 from monai.apps.auto3dseg.bundle_gen import BundleGen
 from monai.apps.auto3dseg.data_analyzer import DataAnalyzer
 from monai.apps.auto3dseg.ensemble_builder import EnsembleRunner
-from monai.apps.auto3dseg.hpo_gen import NNIGen
+from monai.apps.auto3dseg.hpo_gen import NNIGen, _nni_trial_command
 from monai.apps.auto3dseg.utils import export_bundle_algo_history, import_bundle_algo_history
 from monai.apps.utils import get_logger
 from monai.auto3dseg.utils import algo_to_json
@@ -797,7 +797,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,8 @@
 from __future__ import annotations
 
 import os
+import shlex
+import sys
 from abc import abstractmethod
 from copy import deepcopy
 from typing import Any, cast
@@ -32,6 +34,58 @@
 __all__ = ["HPOGen", "NNIGen", "OptunaGen"]
 
 
+# cmd.exe expands or ends a quoted argument at these characters even inside double quotes
+_CMD_UNSAFE_CHARS = frozenset('"%!\r\n\x00')
+
+
+def _quote_trial_arg(arg: Any, windows: bool | None = None) -> str:
+    """
+    Quote one argument of an NNI ``trialCommand`` as a literal for the shell NNI uses.
+
+    NNI's local training service runs ``trialCommand`` with bash on Linux and macOS and with ``cmd.exe`` on
+    Windows. For ``cmd.exe`` the argument is wrapped in double quotes, which make ``&``, ``|``, ``<``, ``>``,
+    ``^`` and parentheses literal, and backslashes before the closing quote are doubled so the program receives
+    the argument unchanged. Characters that ``cmd.exe`` interprets even inside double quotes are rejected.
+
+    Args:
+        arg: the argument value.
+        windows: quote for ``cmd.exe`` instead of a POSIX shell. Defaults to ``sys.platform == "win32"``.
+
+    Returns:
+        the quoted argument.
+
+    Raises:
+        ValueError: on Windows, if the argument contains ``"``, ``%``, ``!`` or a line break or NUL character.
+    """
+    text = str(arg)
+    if sys.platform == "win32" if windows is None else windows:
+        unsafe = sorted(_CMD_UNSAFE_CHARS.intersection(text))
+        if unsafe:
+            raise ValueError(f"Cannot pass {text!r} to an NNI trial on Windows: it contains {unsafe}.")
+        trailing = len(text) - len(text.rstrip("\\"))
+        return '"' + text + "\\" * trailing + '"'
+    return shlex.quote(text)
+
+
+def _nni_trial_command(obj_filename: Any, work_dir: Any, windows: bool | None = None) -> str:
+    """
+    Build the NNI ``trialCommand`` that runs one algo trial.
+
+    The fixed part of the command is plain text; the two paths are quoted with ``_quote_trial_arg`` so they
+    are passed to ``NNIGen run_algo`` unchanged, whatever characters they contain.
+
+    Args:
+        obj_filename: the serialized algo object file.
+        work_dir: the Auto3DSeg working directory.
+        windows: quote for ``cmd.exe`` instead of a POSIX shell. Defaults to ``sys.platform == "win32"``.
+
+    Returns:
+        the command string to store as ``trialCommand``.
+    """
+    paths = " ".join(_quote_trial_arg(p, windows) for p in (obj_filename, work_dir))
+    return f"python -m monai.apps.auto3dseg NNIGen run_algo {paths}"
+
+
 class HPOGen(AlgoGen):
     """
     The base class for hyperparameter optimization (HPO) interfaces to generate algos in the Auto3Dseg pipeline.
@@ -143,7 +197,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} {_quote_trial_arg(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,109 @@
+# 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 json
+import os
+import shlex
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from monai.apps.auto3dseg.hpo_gen import _nni_trial_command, _quote_trial_arg
+from tests.test_utils import skip_if_windows
+
+PREFIX = ["python", "-m", "monai.apps.auto3dseg", "NNIGen", "run_algo"]
+SPECIAL_POSIX = ("my work", "a;b", "$(echo x)", "`x`", "a'b\"c", "a|b&c", "a\u2019b", "%PATH%", "a!b")
+SPECIAL_WINDOWS = ("my work", "a;b", "$(echo x)", "`x`", "a'b", "a|b&c", "a\u2019b", "a^b", "(x)", "a<b>c")
+ECHO_ARGS = "import json, sys; print(json.dumps(sys.argv[1:]))"
+
+
+def _run_through_shell(cmd: str, tempdir: str) -> list[str]:
+    """
+    Run ``cmd`` with the platform shell, as NNI does, with ``python`` swapped for a script printing its arguments.
+
+    ``subprocess.run(..., shell=True)`` uses ``/bin/sh`` on POSIX and ``cmd.exe`` on Windows, matching the shells
+    NNI's local training service runs ``trialCommand`` with.
+
+    Args:
+        cmd: a command produced by ``_nni_trial_command``.
+        tempdir: a directory for the helper script.
+
+    Returns:
+        the arguments the program received.
+    """
+    script = os.path.join(tempdir, "echo_args.py")
+    with open(script, "w") as f:
+        f.write(ECHO_ARGS)
+    if sys.platform == "win32":
+        program = subprocess.list2cmdline([sys.executable, script])
+    else:
+        program = shlex.join([sys.executable, script])
+    full_cmd = program + cmd[len("python") :]
+    out = subprocess.run(full_cmd, shell=True, capture_output=True, text=True, check=True, cwd=tempdir).stdout
+    return json.loads(out)
+
+
+class TestNNITrialCommandPosix(unittest.TestCase):
+    def test_plain_paths(self):
+        cmd = _nni_trial_command("/work/algo_0/algo_object.json", "/work", windows=False)
+        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 name in SPECIAL_POSIX:
+            with self.subTest(name=name):
+                work_dir = f"/tmp/{name}"
+                cmd = _nni_trial_command(f"{work_dir}/algo_object.json", work_dir, windows=False)
+                self.assertEqual(shlex.split(cmd), PREFIX + [f"{work_dir}/algo_object.json", work_dir])
+
+    @skip_if_windows
+    def test_shell_receives_the_literal_path(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            marker = os.path.join(tempdir, "marker")
+            for name in SPECIAL_POSIX + (f"w; touch {marker}", f"$(touch {marker})"):
+                with self.subTest(name=name):
+                    work_dir = os.path.join(tempdir, name)
+                    args = _run_through_shell(_nni_trial_command("obj.json", work_dir), tempdir)
+                    self.assertEqual(args, PREFIX[1:] + ["obj.json", work_dir])
+                    self.assertFalse(os.path.exists(marker))
+
+
+class TestNNITrialCommandWindows(unittest.TestCase):
+    def test_cmd_quoting(self):
+        self.assertEqual(_quote_trial_arg("C:\\work dir", windows=True), '"C:\\work dir"')
+        self.assertEqual(_quote_trial_arg("C:\\", windows=True), '"C:\\\\"')
+        self.assertEqual(
+            _nni_trial_command("C:\\w\\obj.json", "C:\\w", windows=True),
+            'python -m monai.apps.auto3dseg NNIGen run_algo "C:\\w\\obj.json" "C:\\w"',
+        )
+
+    def test_characters_cmd_cannot_quote_are_rejected(self):
+        for name in ('a"b', "%PATH%", "a!b", "a\nb", "a\x00b"):
+            with self.subTest(name=name):
+                with self.assertRaises(ValueError):
+                    _nni_trial_command("obj.json", f"C:\\{name}", windows=True)
+
+    @unittest.skipUnless(sys.platform == "win32", "Requires Windows cmd.exe")
+    def test_cmd_receives_the_literal_path(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            marker = os.path.join(tempdir, "marker")
+            for name in SPECIAL_WINDOWS + (f"w & type nul > {marker}", f"w | type nul > {marker}"):
+                with self.subTest(name=name):
+                    work_dir = os.path.join(tempdir, name)
+                    args = _run_through_shell(_nni_trial_command("obj.json", work_dir), tempdir)
+                    self.assertEqual(args, PREFIX[1:] + ["obj.json", work_dir])
+                    self.assertFalse(os.path.exists(marker))
+
+
+if __name__ == "__main__":
+    unittest.main()