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,7 @@
 from __future__ import annotations
 
 import os
+import shlex
 from abc import abstractmethod
 from copy import deepcopy
 from typing import Any, cast
@@ -32,6 +33,50 @@
 __all__ = ["HPOGen", "NNIGen", "OptunaGen"]
 
 
+# characters PowerShell accepts as single quotes; each is escaped by doubling it
+_POWERSHELL_SINGLE_QUOTES = ("'", "\u2018", "\u2019", "\u201a", "\u201b")
+
+
+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 runs ``trialCommand`` with bash on Linux and macOS and with PowerShell on Windows.
+
+    Args:
+        arg: the argument value.
+        windows: quote for PowerShell instead of a POSIX shell. Defaults to ``os.name == "nt"``.
+
+    Returns:
+        the quoted argument.
+    """
+    text = str(arg)
+    if windows if windows is not None else os.name == "nt":
+        for quote in _POWERSHELL_SINGLE_QUOTES:
+            text = text.replace(quote, quote * 2)
+        return f"'{text}'"
+    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 PowerShell instead of a POSIX shell. Defaults to ``os.name == "nt"``.
+
+    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 +188,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,86 @@
+# 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 shutil
+import subprocess
+import tempfile
+import unittest
+
+from monai.apps.auto3dseg.hpo_gen import _nni_trial_command, _quote_trial_arg
+
+PREFIX = ["python", "-m", "monai.apps.auto3dseg", "NNIGen", "run_algo"]
+SPECIAL = ("my work", "a;b", "$(echo x)", "`x`", "a'b\"c", "a|b&c", "a\u2019b", "$env:PATH", "@(1)")
+POWERSHELL = shutil.which("pwsh") or shutil.which("powershell")
+
+
+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:
+            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])
+
+    @unittest.skipIf(os.name == "nt" or not shutil.which("sh"), "Requires a POSIX sh")
+    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, windows=False)
+            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))
+
+
+class TestNNITrialCommandPowerShell(unittest.TestCase):
+    def test_powershell_literal_quoting(self):
+        self.assertEqual(_quote_trial_arg("C:\\work dir", windows=True), "'C:\\work dir'")
+        self.assertEqual(_quote_trial_arg("C:\\O'Neil", windows=True), "'C:\\O''Neil'")
+        self.assertEqual(_quote_trial_arg("a\u2019b", windows=True), "'a\u2019\u2019b'")
+        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'",
+        )
+
+    @unittest.skipUnless(POWERSHELL, "Requires PowerShell")
+    def test_powershell_receives_the_literal_path(self):
+        """Run the command through PowerShell with ``python`` swapped for a function that echoes its arguments."""
+        with tempfile.TemporaryDirectory() as tempdir:
+            marker = os.path.join(tempdir, "marker")
+            for name in SPECIAL + (f"w; New-Item {marker}", f"$(New-Item {marker})"):
+                with self.subTest(name=name):
+                    work_dir = os.path.join(tempdir, name)
+                    cmd = _nni_trial_command("obj.json", work_dir, windows=True)
+                    script = "function python { $args | ForEach-Object { Write-Output $_ } }; " + cmd
+                    out = subprocess.run(
+                        [str(POWERSHELL), "-NoProfile", "-NonInteractive", "-Command", script],
+                        capture_output=True,
+                        text=True,
+                        encoding="utf-8",
+                        check=True,
+                    ).stdout
+                    self.assertEqual(out.splitlines(), PREFIX[1:] + ["obj.json", work_dir])
+                    self.assertFalse(os.path.exists(marker))
+
+
+if __name__ == "__main__":
+    unittest.main()