Fix GHSA-873f-pvrv-4x83: warn before executing a bundle's config in load()/run()

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

@@ -15,6 +15,7 @@
 import os
 import tempfile
 import unittest
+import warnings
 from unittest.case import skipIf, skipUnless
 from unittest.mock import patch
 
@@ -24,7 +25,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 (
@@ -488,5 +489,65 @@ 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"))
+        # `marker` is embedded via `!r` (not raw-interpolated) since this string is itself later
+        # evaluated as Python source -- on Windows, a raw path's backslashes would otherwise be
+        # misparsed as escape sequences.
+        payload = f"$__import__('os').system({('echo pwned > ' + marker)!r})"
+        malicious_config = {"network_def": payload, "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.assertWarnsRegex(UserWarning, r"GHSA-873f-pvrv-4x83"):
+                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,))
+            with warnings.catch_warnings():
+                warnings.simplefilter("error", UserWarning)
+                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:
+                payload = f"$__import__('os').system({('echo pwned > ' + marker)!r})"
+                json.dump({"initialize": [payload]}, f)
+            with self.assertWarnsRegex(UserWarning, r"GHSA-873f-pvrv-4x83"):
+                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()