Harden download/deserialize integrity (weights_only, hash checks, path confinement) (#9088)

### Description

Harden the download/deserialization chain:

- Pass `weights_only=True` to pretrained weight loaders in `senet`,
`densenet`, `efficientnet`, and `text_embedding` so a substituted `.pth`
cannot unpickle code (GHSA-vm9c-7j6g-c7mm).
- `check_hash`: emit a `UserWarning` when no hash value is provided
instead of silently passing, and default `check_hash`/`download_url` to
`sha256` (GHSA-hhh4-h52m-fqh6).
- `download_large_files`: confine large-file targets to the bundle
directory, rejecting absolute and `../` traversal (GHSA-x4pc-gj5h-3pq7).

Note: SENet pretrained URLs remain `http://` because the upstream host
does not serve the files over HTTPS (verified unreachable);
`weights_only=True` closes the code-execution vector, leaving only a
transport-integrity gap.

### Types of changes
- [ ] Non-breaking change
- [x] Breaking change (default hash type changes from md5 to sha256 for
`check_hash`/`download_url`; callers that relied on the md5 default now
pass `hash_type` explicitly)
- [x] New tests added to cover the changes.

---------

Signed-off-by: R. Garcia-Dias <rafaelagd@gmail.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Eric Kerfoot <17726042+ericspod@users.noreply.github.com>

Author: 3 people

monai/apps/utils.py

@@ -166,20 +166,20 @@ def safe_extract_member(member, extract_to):
     return full_path
 
 
-def check_hash(filepath: PathLike, val: str | None = None, hash_type: str = "md5") -> bool:
+def check_hash(filepath: PathLike, val: str | None = None, hash_type: str = "sha256") -> bool:
     """
     Verify hash signature of specified file.
 
     Args:
         filepath: path of source file to verify hash value.
         val: expected hash value of the file.
-        hash_type: type of hash algorithm to use, default is `"md5"`.
+        hash_type: type of hash algorithm to use, default is `"sha256"`.
             The supported hash types are `"md5"`, `"sha1"`, `"sha256"`, `"sha512"`.
             See also: :py:data:`monai.apps.utils.SUPPORTED_HASH_TYPES`.
 
     """
     if val is None:
-        logger.info(f"Expected {hash_type} is None, skip {hash_type} check for file {filepath}.")
+        warnings.warn(f"No hash value provided for {filepath}; file integrity is NOT verified.", stacklevel=2)
         return True
     actual_hash_func = look_up_option(hash_type.lower(), SUPPORTED_HASH_TYPES)
 
@@ -204,7 +204,7 @@ def download_url(
     url: str,
     filepath: PathLike = "",
     hash_val: str | None = None,
-    hash_type: str = "md5",
+    hash_type: str = "sha256",
     progress: bool = True,
     **gdown_kwargs: Any,
 ) -> None:
@@ -217,7 +217,8 @@ def download_url(
             If undefined, `os.path.basename(url)` will be used.
         hash_val: expected hash value to validate the downloaded file.
             if None, skip hash validation.
-        hash_type: 'md5' or 'sha1', defaults to 'md5'.
+        hash_type: type of hash algorithm to use, default is `"sha256"`.
+            The supported hash types are `"md5"`, `"sha1"`, `"sha256"`, `"sha512"`.
         progress: whether to display a progress bar.
         gdown_kwargs: other args for `gdown` except for the `url`, `output` and `quiet`.
             these args will only be used if download from google drive.
@@ -315,7 +316,7 @@ def extractall(
     filepath: PathLike,
     output_dir: PathLike = ".",
     hash_val: str | None = None,
-    hash_type: str = "md5",
+    hash_type: str = "sha256",
     file_type: str = "",
     has_base: bool = True,
 ) -> None:
@@ -328,7 +329,7 @@ def extractall(
         output_dir: target directory to save extracted files.
         hash_val: expected hash value to validate the compressed file.
             if None, skip hash validation.
-        hash_type: 'md5' or 'sha1', defaults to 'md5'.
+        hash_type: type of hash algorithm to use, default is `"sha256"`.
         file_type: string of file type for decompressing. Leave it empty to infer the type from the filepath basename.
         has_base: whether the extracted files have a base folder. This flag is used when checking if the existing
             folder is a result of `extractall`, if it is, the extraction is skipped. For example, if A.zip is unzipped
@@ -394,7 +395,7 @@ def download_and_extract(
     filepath: PathLike = "",
     output_dir: PathLike = ".",
     hash_val: str | None = None,
-    hash_type: str = "md5",
+    hash_type: str = "sha256",
     file_type: str = "",
     has_base: bool = True,
     progress: bool = True,
@@ -410,7 +411,7 @@ def download_and_extract(
             default is the current directory.
         hash_val: expected hash value to validate the downloaded file.
             if None, skip hash validation.
-        hash_type: 'md5' or 'sha1', defaults to 'md5'.
+        hash_type: type of hash algorithm to use, default is `"sha256"`.
         file_type: string of file type for decompressing. Leave it empty to infer the type from url's base file name.
         has_base: whether the extracted files have a base folder. This flag is used when checking if the existing
             folder is a result of `extractall`, if it is, the extraction is skipped. For example, if A.zip is unzipped

monai/bundle/scripts.py

@@ -2008,6 +2008,17 @@ def create_workflow(
     return workflow_
 
 
+def _safe_large_file_path(bundle_path: PathLike, filepath: str) -> str:
+    """Securely resolve a large-file target path to prevent traversal outside the bundle directory."""
+    bundle_root = os.path.realpath(bundle_path)
+    target = os.path.normpath(os.path.join(bundle_path, filepath))
+    target_real = os.path.realpath(target)
+    # Ensure the resolved path stays within the bundle root
+    if os.path.commonpath([bundle_root, target_real]) != bundle_root:
+        raise ValueError(f"Unsafe path: path traversal {filepath} for bundle_path {bundle_path}")
+    return target
+
+
 def download_large_files(bundle_path: str | None = None, large_file_name: str | None = None) -> None:
     """
     This utility allows you to download large files from a bundle. It supports file suffixes like ".yml", ".yaml", and ".json".
@@ -2042,6 +2053,6 @@ def download_large_files(bundle_path: str | None = None, large_file_name: str |
             lf_data.pop("hash_val")
         if "hash_type" in lf_data and lf_data.get("hash_type", "") == "":
             lf_data.pop("hash_type")
-        lf_data["filepath"] = os.path.join(bundle_path, lf_data["path"])
+        lf_data["filepath"] = _safe_large_file_path(bundle_path, lf_data["path"])
         lf_data.pop("path")
         download_url(**lf_data)

monai/networks/blocks/text_embedding.py

@@ -67,7 +67,7 @@ def __init__(
 
             if pretrained:
                 model_url = url_map[self.encoding]
-                pretrain_state_dict = model_zoo.load_url(model_url, map_location="cpu")
+                pretrain_state_dict = model_zoo.load_url(model_url, map_location="cpu", weights_only=True)
                 self.text_embedding.data = pretrain_state_dict.float()  # type: ignore
             else:
                 print(f"{self.encoding} is not implemented, and can not be downloaded, please load your own")

monai/networks/nets/densenet.py

@@ -277,7 +277,7 @@ def _load_state_dict(model: nn.Module, arch: str, progress: bool):
         r"^(.*denselayer\d+)(\.(?:norm|relu|conv))\.((?:[12])\.(?:weight|bias|running_mean|running_var))$"
     )
 
-    state_dict = load_state_dict_from_url(model_url, progress=progress)
+    state_dict = load_state_dict_from_url(model_url, progress=progress, weights_only=True)
     for key in list(state_dict.keys()):
         res = pattern.match(key)
         if res:

monai/networks/nets/efficientnet.py

@@ -793,7 +793,7 @@ def _load_state_dict(model: nn.Module, arch: str, progress: bool, adv_prop: bool
     else:
         # load state dict from url
         model_url = url_map[arch]
-        pretrain_state_dict = model_zoo.load_url(model_url, progress=progress)
+        pretrain_state_dict = model_zoo.load_url(model_url, progress=progress, weights_only=True)
         model_state_dict = model.state_dict()
 
         pattern = re.compile(r"(.+)\.\d+(\.\d+\..+)")

monai/networks/nets/senet.py

@@ -304,7 +304,7 @@ def _load_state_dict(model: nn.Module, arch: str, progress: bool):
         download_url(model_url["url"], filepath=model_url["filename"])
         state_dict = torch.load(model_url["filename"], map_location=None, weights_only=True)
     else:
-        state_dict = load_state_dict_from_url(model_url, progress=progress)
+        state_dict = load_state_dict_from_url(model_url, progress=progress, weights_only=True)
     for key in list(state_dict.keys()):
         new_key = None
         if pattern_conv.match(key):

tests/apps/test_check_hash.py

@@ -11,6 +11,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import os
 import tempfile
 import unittest
@@ -48,6 +49,23 @@ def test_hash_type_error(self):
             with tempfile.TemporaryDirectory() as tempdir:
                 check_hash(tempdir, "test_hash", "test_type")
 
+    def test_warns_when_val_is_none(self):
+        test_image = np.ones((5, 5, 3))
+        with tempfile.TemporaryDirectory() as tempdir:
+            filename = os.path.join(tempdir, "test_file.png")
+            test_image.tofile(filename)
+            with self.assertWarns(UserWarning):
+                result = check_hash(filename, None)
+            self.assertTrue(result)
+
+    def test_default_hash_type_is_sha256(self):
+        test_image = np.ones((5, 5, 3))
+        with tempfile.TemporaryDirectory() as tempdir:
+            filename = os.path.join(tempdir, "test_file.png")
+            test_image.tofile(filename)
+            sha256 = hashlib.sha256(test_image.tobytes()).hexdigest()
+            self.assertTrue(check_hash(filename, sha256))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/bundle/test_bundle_download.py

@@ -26,7 +26,7 @@
 import monai.networks.nets as nets
 from monai.apps import check_hash
 from monai.bundle import ConfigParser, create_workflow, load, run
-from monai.bundle.scripts import _examine_monai_version, _list_latest_versions, download
+from monai.bundle.scripts import _examine_monai_version, _list_latest_versions, download, download_large_files
 from monai.utils import optional_import
 from tests.test_utils import (
     assert_allclose,
@@ -166,7 +166,7 @@ def test_github_download_bundle(self, bundle_name, version):
                     file_path = os.path.join(tempdir, "test_bundle", file)
                     self.assertTrue(os.path.exists(file_path))
                     if file == "network.json":
-                        self.assertTrue(check_hash(filepath=file_path, val=hash_val))
+                        self.assertTrue(check_hash(filepath=file_path, val=hash_val, hash_type="md5"))
 
     @parameterized.expand([TEST_CASE_3])
     @skip_if_quick
@@ -184,8 +184,8 @@ def test_url_download_bundle(self, bundle_files, bundle_name, url, hash_val):
                 for file in bundle_files:
                     file_path = os.path.join(tempdir, bundle_name, file)
                     self.assertTrue(os.path.exists(file_path))
-                if file == "network.json":
-                    self.assertTrue(check_hash(filepath=file_path, val=hash_val))
+                    if file == "network.json":
+                        self.assertTrue(check_hash(filepath=file_path, val=hash_val, hash_type="md5"))
 
     @parameterized.expand([TEST_CASE_4])
     @skip_if_quick
@@ -445,6 +445,15 @@ def test_load_ts_module(self, bundle_files, bundle_name, version, repo, device,
 
 
 class TestDownloadLargefiles(unittest.TestCase):
+
+    def test_large_files_rejects_path_traversal(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            large_files_path = os.path.join(tempdir, "large_files.yaml")
+            with open(large_files_path, "w") as f:
+                f.write("large_files:\n" "  - path: ../evil.pt\n" "    url: https://example.com/evil.pt\n")
+            with self.assertRaises(ValueError):
+                download_large_files(bundle_path=tempdir)
+
     @parameterized.expand([TEST_CASE_10])
     @skip_if_quick
     def test_url_download_large_files(self, bundle_files, bundle_name, url, hash_val):
@@ -469,7 +478,7 @@ def test_url_download_large_files(self, bundle_files, bundle_name, url, hash_val
                 command_line_tests(cmd)
                 for file in ["model.pt", "model.ts"]:
                     file_path = os.path.join(tempdir, bundle_name, f"models/{file}")
-                    self.assertTrue(check_hash(filepath=file_path, val=hash_val[file]))
+                    self.assertTrue(check_hash(filepath=file_path, val=hash_val[file], hash_type="md5"))
 
 
 @skip_if_windows
@@ -484,7 +493,7 @@ def test_ngc_download_bundle(self, bundle_name, version, remove_prefix, download
                 )
                 full_file_path = os.path.join(tempdir, download_name, file_path)
                 self.assertTrue(os.path.exists(full_file_path))
-                self.assertTrue(check_hash(filepath=full_file_path, val=hash_val))
+                self.assertTrue(check_hash(filepath=full_file_path, val=hash_val, hash_type="md5"))
 
                 model = load(
                     name=bundle_name, source="ngc", version=version, bundle_dir=tempdir, remove_prefix=remove_prefix