diff --git a/monai/bundle/scripts.py b/monai/bundle/scripts.py index 63a774bfea..973f4a4dc2 100644 --- a/monai/bundle/scripts.py +++ b/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) diff --git a/tests/bundle/test_bundle_download.py b/tests/bundle/test_bundle_download.py index bb213cebd9..7314231550 100644 --- a/tests/bundle/test_bundle_download.py +++ b/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()