Restore algo state values literally in algo_from_json

monai/auto3dseg/utils.py

@@ -63,13 +63,13 @@ def _require_pickle_allowed() -> None:
         raise RuntimeError(_PICKLE_DISABLED_MSG)
 
 
-def _reject_non_algo_target(target: Any, filename: str) -> None:
+def _reject_non_algo_target(target: Any, filename: str) -> type[Algo]:
     """
     Require that ``target`` names an ``Algo`` subclass before it is instantiated.
 
-    ``algo_from_json`` resolves ``_target_`` through ``ConfigParser``, which imports the dotted path
-    and calls it. ``algo_object.json`` has exactly one legitimate target type -- an ``Algo``
-    subclass -- so resolving the name and checking it here rejects payloads such as
+    ``algo_from_json`` resolves ``_target_`` by importing the dotted path and then calls it.
+    ``algo_object.json`` has exactly one legitimate target type -- an ``Algo`` subclass -- so
+    resolving the name and checking it here rejects payloads such as
     ``subprocess.call`` or ``os.system`` before they are ever invoked.
 
     Importing the named module still runs that module's top-level code, so this narrows the sink
@@ -80,6 +80,9 @@ def _reject_non_algo_target(target: Any, filename: str) -> None:
         target: the ``_target_`` value read from the file.
         filename: the file the value came from, used in the error message.
 
+    Returns:
+        the resolved ``Algo`` subclass.
+
     Raises:
         ValueError: if ``target`` is not a string, or does not resolve to a concrete ``Algo`` subclass.
         ModuleNotFoundError: if the module cannot be imported, so callers can try the next
@@ -97,6 +100,7 @@ def _reject_non_algo_target(target: Any, filename: str) -> None:
             "may be named in a configuration "
             "(see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-2wx3-8x3w-r8qv)."
         )
+    return resolved
 
 
 measure_np, has_measure = optional_import("skimage.measure", "0.14.2", min_version)
@@ -525,15 +529,12 @@ def algo_from_json(filename: str, template_path: PathLike | None = None, **kwarg
                 sys.path.insert(0, path)
                 path_added = True
 
-            algo_config: dict[str, Any] = {"_target_": target}
-            state_template_path = state.get("template_path")
-            if state_template_path:
-                algo_config["template_path"] = state_template_path
+            algo_cls: Any = _reject_non_algo_target(target, filename)  # subclasses may accept template_path
 
-            _reject_non_algo_target(target, filename)
-
-            parser = ConfigParser(algo_config)
-            algo = parser.get_parsed_content()
+            # Values read from the file are plain data: pass them to the constructor as literals
+            # rather than through ConfigParser, which would evaluate "$"/"@"/"%" prefixed strings.
+            state_template_path = state.get("template_path")
+            algo = algo_cls(template_path=state_template_path) if state_template_path else algo_cls()
             if not isinstance(algo, Algo):
                 raise ValueError(
                     f"refusing to return the object built from '{target}' in {filename}: it is a "

tests/auto3dseg/test_algo_target_allowlist.py

@@ -22,7 +22,7 @@
 class TestAlgoTargetAllowlist(unittest.TestCase):
     """Regression tests for GHSA-2wx3-8x3w-r8qv.
 
-    ``algo_from_json`` resolves the file's ``_target_`` through ``ConfigParser``, which imports the
+    ``algo_from_json`` resolves the file's ``_target_`` by importing the
     dotted path and calls it. ``algo_object.json`` has exactly one legitimate target type, so a
     ``_target_`` that is not an ``Algo`` subclass is rejected before it is instantiated.
     """

tests/auto3dseg/test_json_serialization.py

@@ -11,6 +11,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 import pickle
 import tempfile
@@ -115,5 +116,56 @@ def test_algo_from_json_legacy_pkl_disabled_by_default(self) -> None:
                 algo_from_json(pkl)
 
 
+class TestAlgoFromJsonStateLiterals(unittest.TestCase):
+    """Values in the saved ``_state_`` are restored as plain data, not interpreted as config syntax."""
+
+    def _load(self, template_path: str, **state: object) -> object:
+        """
+        Write an ``algo_object.json`` for a ``BundleAlgo`` and load it with ``algo_from_json``.
+
+        Args:
+            template_path: the ``template_path`` value stored in ``_state_``.
+            state: other ``_state_`` entries to store.
+
+        Returns:
+            the loaded Algo object.
+        """
+        with tempfile.TemporaryDirectory() as tempdir:
+            path = os.path.join(tempdir, "algo_object.json")
+            with open(path, "w", encoding="utf-8") as f:
+                json.dump(
+                    {
+                        "_target_": "monai.apps.auto3dseg.bundle_gen.BundleAlgo",
+                        "_state_": {"template_path": template_path, **state},
+                    },
+                    f,
+                )
+            algo, _ = algo_from_json(path)
+            return algo
+
+    def test_template_path_with_config_prefixes_is_not_evaluated(self) -> None:
+        """Strings that look like config syntax reach the constructor verbatim and are never evaluated."""
+        from monai.apps.auto3dseg.bundle_gen import BundleAlgo
+
+        for value in ("$len([1, 2])", "@some_id", "%other.json#key"):
+            with self.subTest(value=value):
+                with (
+                    mock.patch("monai.bundle.config_item.ConfigExpression.evaluate") as evaluate,
+                    mock.patch.object(BundleAlgo, "__init__", autospec=True, return_value=None) as init,
+                    mock.patch.object(BundleAlgo, "get_output_path", return_value=""),
+                ):
+                    algo = self._load(value)
+                evaluate.assert_not_called()
+                init.assert_called_once_with(algo, template_path=value)
+
+    def test_plain_state_is_restored(self) -> None:
+        """Saved state values that differ from the constructor defaults are restored exactly."""
+        records = {"hyper_parameters.yaml": {"num_epochs": 2}}
+        algo = self._load("/some/templates", name="segresnet_0", fill_records=records, best_metric=0.5)
+        self.assertEqual(algo.name, "segresnet_0")
+        self.assertEqual(algo.fill_records, records)
+        self.assertEqual(algo.best_metric, 0.5)
+
+
 if __name__ == "__main__":
     unittest.main()