Fix GHSA-873f-pvrv-4x83: warn before executing a bundle's config

monai.bundle.load(), with its default model=None, builds a bundle's
network by parsing the bundle's own config through create_workflow().
That parsing resolves any "_target_" value to an importable callable
with no allow list, and passes any "$"-prefixed value to Python
eval(). monai.bundle.run() reaches the same path via a caller-supplied
config_file. Either way, loading or running a bundle whose config
hasn't been reviewed can execute arbitrary code.

create_workflow() -- the shared path both load() and run() use to
parse a config file -- now raises a UserWarning immediately before
doing so, describing what "_target_"/"$"-expression content can do
and linking the advisory. This applies uniformly to every caller of
create_workflow(), not just load(). No behavior is blocked: the config
is still parsed and executed exactly as before, just with a warning
first. An earlier version of this fix added an opt-in
trust_remote_code flag to load(), but that was dropped after review:
MONAI has no way to establish whether a bundle is actually
trustworthy, so a flag like that would only teach callers to set it
once and forget about it.

Update docstrings on load(), run(), and create_workflow() to describe
the risk and point at the advisory.

Add TestLoadWarnsOnConfigExecution to tests/bundle/test_bundle_download.py:
default load() warns and still executes the config, explicit model=
skips config parsing entirely and warns about nothing, and run() warns
via the same create_workflow() path.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Signed-off-by: R. Garcia-Dias <rafaelagd@gmail.com>

Author: garciadias

monai/bundle/scripts.py

@@ -648,6 +648,14 @@ def load(
     """
     Load model weights or TorchScript module of a bundle.
 
+    Security note: if `model` is `None`, building `network_def` requires parsing the bundle's own
+    "{workflow_type}.json" config, which can define `"_target_"` components resolved to any importable
+    callable and `"$"`-prefixed expressions evaluated with Python `eval()`. Only call `load()` this way
+    for bundles from a source you trust; a warning is printed every time this happens
+    (see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-873f-pvrv-4x83). To skip parsing
+    the bundle's config entirely, pass an explicit `model=` — only the weights are then loaded, via
+    `torch.load(..., weights_only=True)`.
+
     Args:
         name: bundle name. If `None` and `url` is `None`, it must be provided in `args_file`.
             for example:
@@ -935,6 +943,12 @@ def run(
     """
     Specify `config_file` to run monai bundle components and workflows.
 
+    Security note: parsing `config_file` can run arbitrary code. Any `"_target_"` value is resolved to an
+    importable callable and invoked with no allow list, and any `"$"`-prefixed value is passed to Python
+    `eval()`. Only point this at config files you wrote or otherwise fully trust; never at a config
+    downloaded from, or otherwise sourced from, an untrusted party. A warning is printed every time this
+    happens (see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-873f-pvrv-4x83).
+
     Typical usage examples:
 
     .. code-block:: bash
@@ -1929,6 +1943,12 @@ def create_workflow(
     The workflow should be subclass of `BundleWorkflow` and be available to import.
     It can be MONAI existing bundle workflows or user customized workflows.
 
+    Security note: parsing `config_file` can run arbitrary code. Any `"_target_"` value is resolved to an
+    importable callable and invoked with no allow list, and any `"$"`-prefixed value is passed to Python
+    `eval()`. Only point this at config files you wrote or otherwise fully trust; never at a config
+    downloaded from, or otherwise sourced from, an untrusted party. A warning is printed every time this
+    happens (see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-873f-pvrv-4x83).
+
     Typical usage examples:
 
     .. code-block:: python
@@ -1966,6 +1986,13 @@ def create_workflow(
         )
 
     if config_file is not None:
+        warnings.warn(
+            f"parsing config_file {config_file}: any `\"_target_\"` value in it is resolved to an importable "
+            "callable and invoked with no allow list, and any `\"$\"`-prefixed value is passed to Python "
+            "`eval()`. Only proceed if this config is from a source you trust "
+            "(see https://github.com/Project-MONAI/MONAI/security/advisories/GHSA-873f-pvrv-4x83).",
+            stacklevel=2,
+        )
         workflow_ = workflow_class(config_file=config_file, **_args)
     else:
         workflow_ = workflow_class(**_args)

tests/bundle/test_bundle_download.py

@@ -24,7 +24,7 @@
 
 import monai.networks.nets as nets
 from monai.apps import check_hash
-from monai.bundle import ConfigParser, create_workflow, load
+from monai.bundle import ConfigParser, create_workflow, load, run
 from monai.bundle.scripts import _examine_monai_version, _list_latest_versions, download
 from monai.utils import optional_import
 from tests.test_utils import (
@@ -368,7 +368,13 @@ def test_load_weights_with_net_override(self, bundle_name, device, net_override)
             # download bundle, and load weights from the downloaded path
             with tempfile.TemporaryDirectory() as tempdir:
                 # load weights
-                model = load(name=bundle_name, bundle_dir=tempdir, source="monaihosting", progress=False, device=device)
+                model = load(
+                    name=bundle_name,
+                    bundle_dir=tempdir,
+                    source="monaihosting",
+                    progress=False,
+                    device=device,
+                )
 
                 # prepare data and test
                 input_tensor = torch.rand(1, 1, 96, 96, 96).to(device)
@@ -477,7 +483,11 @@ def test_ngc_download_bundle(self, bundle_name, version, remove_prefix, download
                 self.assertTrue(check_hash(filepath=full_file_path, val=hash_val))
 
                 model = load(
-                    name=bundle_name, source="ngc", version=version, bundle_dir=tempdir, remove_prefix=remove_prefix
+                    name=bundle_name,
+                    source="ngc",
+                    version=version,
+                    bundle_dir=tempdir,
+                    remove_prefix=remove_prefix,
                 )
                 assert_allclose(
                     model.state_dict()[TESTCASE_NGC_WEIGHTS["key"]],
@@ -488,5 +498,58 @@ def test_ngc_download_bundle(self, bundle_name, version, remove_prefix, download
                 )
 
 
+class TestLoadWarnsOnConfigExecution(unittest.TestCase):
+    """Regression tests for GHSA-873f-pvrv-4x83: `load()`/`create_workflow()` parse and execute a
+    bundle's own config (arbitrary `_target_`/`$`-expression content) whenever `model` is `None`.
+    There is no opt-in flag -- MONAI has no way to establish whether a bundle is actually
+    trustworthy, so a flag would only teach callers to always pass it and ignore the risk. Instead,
+    a `UserWarning` is raised every time this happens, in both `load()` (via `create_workflow()`)
+    and `run()` (also via `create_workflow()`)."""
+
+    def _stage_malicious_bundle(self, tempdir: str, marker: str) -> str:
+        name = "evil_bundle"
+        bundle_root = os.path.join(tempdir, name)
+        os.makedirs(os.path.join(bundle_root, "configs"))
+        os.makedirs(os.path.join(bundle_root, "models"))
+        torch.save({"state_dict": {}}, os.path.join(bundle_root, "models", "model.pt"))
+        malicious_config = {"network_def": f"$__import__('os').system('echo pwned > {marker}')", "initialize": []}
+        with open(os.path.join(bundle_root, "configs", "train.json"), "w") as f:
+            json.dump(malicious_config, f)
+        return name
+
+    def test_default_warns_and_executes_config(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            marker = os.path.join(tempdir, "PWNED")
+            name = self._stage_malicious_bundle(tempdir, marker)
+            with self.assertWarns(UserWarning):
+                with self.assertRaises(AttributeError):
+                    # the malicious config is missing metadata.json and returns a plain `int` for
+                    # `network_def`, so the workflow construction fails after the payload has already
+                    # run -- this mirrors the advisory's own PoC, where the failure happens *after* RCE.
+                    load(name=name, bundle_dir=tempdir, source="github", repo="attacker/repo")
+            self.assertTrue(os.path.exists(marker))
+
+    def test_explicit_model_skips_config_parsing(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            marker = os.path.join(tempdir, "PWNED")
+            name = self._stage_malicious_bundle(tempdir, marker)
+            model = nets.UNet(spatial_dims=2, in_channels=1, out_channels=1, channels=(4, 8), strides=(2,))
+            load(name=name, model=model, bundle_dir=tempdir, source="github", repo="attacker/repo")
+            self.assertFalse(os.path.exists(marker))
+
+    def test_run_warns_on_config_execution(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            marker = os.path.join(tempdir, "PWNED")
+            config_file = os.path.join(tempdir, "train.json")
+            with open(config_file, "w") as f:
+                json.dump({"initialize": [f"$__import__('os').system('echo pwned > {marker}')"]}, f)
+            with self.assertWarns(UserWarning):
+                with self.assertRaises(ValueError):
+                    # no "run" ID is defined, so `workflow.run()` fails after `initialize()` has
+                    # already evaluated the payload above.
+                    run(config_file=config_file)
+            self.assertTrue(os.path.exists(marker))
+
+
 if __name__ == "__main__":
     unittest.main()