diff --git a/monai/apps/utils.py b/monai/apps/utils.py index b15b0c1969..710812c3ec 100644 --- a/monai/apps/utils.py +++ b/monai/apps/utils.py @@ -29,7 +29,7 @@ from urllib.request import urlopen, urlretrieve from monai.config.type_definitions import PathLike -from monai.utils import look_up_option, min_version, optional_import +from monai.utils import deprecated_arg_default, look_up_option, min_version, optional_import requests, has_requests = optional_import("requests") gdown, has_gdown = optional_import("gdown", "4.7.3") @@ -54,6 +54,12 @@ DEFAULT_FMT = "%(asctime)s - %(levelname)s - %(message)s" SUPPORTED_HASH_TYPES = {"md5": hashlib.md5, "sha1": hashlib.sha1, "sha256": hashlib.sha256, "sha512": hashlib.sha512} +_HASH_TYPE_DEFAULT_CHANGE_VERSION = "1.6.1" +_HASH_TYPE_DEFAULT_WARNING_REMOVAL_VERSION = "1.8" +_HASH_TYPE_DEFAULT_CHANGE_MSG = ( + "The default was updated to SHA-256 for stronger integrity verification. " + 'Pass `hash_type="md5"` explicitly only when you need backward compatibility with existing MD5 hashes.' +) class HashCheckError(ValueError): @@ -166,10 +172,22 @@ def safe_extract_member(member, extract_to): return full_path +@deprecated_arg_default( + "hash_type", + old_default='"md5"', + new_default='"sha256"', + since=_HASH_TYPE_DEFAULT_CHANGE_VERSION, + replaced=_HASH_TYPE_DEFAULT_WARNING_REMOVAL_VERSION, + msg_suffix=_HASH_TYPE_DEFAULT_CHANGE_MSG, +) def check_hash(filepath: PathLike, val: str | None = None, hash_type: str = "sha256") -> bool: """ Verify hash signature of specified file. + .. versionchanged:: 1.6.1 + The default ``hash_type`` changed from ``"md5"`` to ``"sha256"`` for stronger integrity verification. + Pass ``hash_type="md5"`` explicitly only when you need backward compatibility with existing MD5 hashes. + Args: filepath: path of source file to verify hash value. val: expected hash value of the file. @@ -200,6 +218,14 @@ def check_hash(filepath: PathLike, val: str | None = None, hash_type: str = "sha return True +@deprecated_arg_default( + "hash_type", + old_default='"md5"', + new_default='"sha256"', + since=_HASH_TYPE_DEFAULT_CHANGE_VERSION, + replaced=_HASH_TYPE_DEFAULT_WARNING_REMOVAL_VERSION, + msg_suffix=_HASH_TYPE_DEFAULT_CHANGE_MSG, +) def download_url( url: str, filepath: PathLike = "", @@ -211,6 +237,10 @@ def download_url( """ Download file from specified URL link, support process bar and hash check. + .. versionchanged:: 1.6.1 + The default ``hash_type`` changed from ``"md5"`` to ``"sha256"`` for stronger integrity verification. + Pass ``hash_type="md5"`` explicitly only when you need backward compatibility with existing MD5 hashes. + Args: url: source URL link to download file. filepath: target filepath to save the downloaded file (including the filename). @@ -312,6 +342,14 @@ def _extract_tar(filepath, output_dir): shutil.copyfileobj(source, target) +@deprecated_arg_default( + "hash_type", + old_default='"md5"', + new_default='"sha256"', + since=_HASH_TYPE_DEFAULT_CHANGE_VERSION, + replaced=_HASH_TYPE_DEFAULT_WARNING_REMOVAL_VERSION, + msg_suffix=_HASH_TYPE_DEFAULT_CHANGE_MSG, +) def extractall( filepath: PathLike, output_dir: PathLike = ".", @@ -324,6 +362,10 @@ def extractall( Extract file to the output directory. Expected file types are: `zip`, `tar.gz` and `tar`. + .. versionchanged:: 1.6.1 + The default ``hash_type`` changed from ``"md5"`` to ``"sha256"`` for stronger integrity verification. + Pass ``hash_type="md5"`` explicitly only when you need backward compatibility with existing MD5 hashes. + Args: filepath: the file path of compressed file. output_dir: target directory to save extracted files. @@ -390,6 +432,14 @@ def get_filename_from_url(data_url: str) -> str: raise Exception(f"Error processing URL: {e}") from e +@deprecated_arg_default( + "hash_type", + old_default='"md5"', + new_default='"sha256"', + since=_HASH_TYPE_DEFAULT_CHANGE_VERSION, + replaced=_HASH_TYPE_DEFAULT_WARNING_REMOVAL_VERSION, + msg_suffix=_HASH_TYPE_DEFAULT_CHANGE_MSG, +) def download_and_extract( url: str, filepath: PathLike = "", @@ -403,6 +453,10 @@ def download_and_extract( """ Download file from URL and extract it to the output directory. + .. versionchanged:: 1.6.1 + The default ``hash_type`` changed from ``"md5"`` to ``"sha256"`` for stronger integrity verification. + Pass ``hash_type="md5"`` explicitly only when you need backward compatibility with existing MD5 hashes. + Args: url: source URL link to download file. filepath: the file path of the downloaded compressed file. @@ -433,4 +487,6 @@ def download_and_extract( with tempfile.TemporaryDirectory() as tmp_dir: filename = filepath or Path(tmp_dir, get_filename_from_url(url)).resolve() download_url(url=url, filepath=filename, hash_val=hash_val, hash_type=hash_type, progress=progress) - extractall(filepath=filename, output_dir=output_dir, file_type=file_type, has_base=has_base) + extractall( + filepath=filename, output_dir=output_dir, hash_type=hash_type, file_type=file_type, has_base=has_base + ) diff --git a/tests/apps/test_check_hash.py b/tests/apps/test_check_hash.py index 75d768b0e5..09235b8cbf 100644 --- a/tests/apps/test_check_hash.py +++ b/tests/apps/test_check_hash.py @@ -13,13 +13,17 @@ import hashlib import os +import shutil import tempfile import unittest +import warnings +import zipfile +from unittest.mock import patch import numpy as np from parameterized import parameterized -from monai.apps import check_hash +from monai.apps import check_hash, download_and_extract, download_url, extractall TEST_CASE_1 = ["b94716452086a054208395e8c9d1ae2a", "md5", True] @@ -33,6 +37,31 @@ class TestCheckMD5(unittest.TestCase): + @staticmethod + def _hash_file(filename, hash_type): + hash_func = getattr(hashlib, hash_type) + with open(filename, "rb") as f: + return hash_func(f.read()).hexdigest() + + @staticmethod + def _write_file(filename, content=b"monai hash fixture"): + with open(filename, "wb") as f: + f.write(content) + + @classmethod + def _download_side_effect(cls, source): + def _copy(_url, destination, progress=True): + del progress + shutil.copyfile(source, destination) + + return _copy + + @classmethod + def _create_zip_fixture(cls, tempdir): + archive = os.path.join(tempdir, "fixture.zip") + with zipfile.ZipFile(archive, "w") as zip_file: + zip_file.writestr("fixture/data.txt", "monai") + return archive @parameterized.expand([TEST_CASE_1, TEST_CASE_2, TEST_CASE_3, TEST_CASE_4, TEST_CASE_5]) def test_result(self, md5_value, t, expected_result): @@ -55,7 +84,7 @@ def test_warns_when_val_is_none(self): filename = os.path.join(tempdir, "test_file.png") test_image.tofile(filename) with self.assertWarns(UserWarning): - result = check_hash(filename, None) + result = check_hash(filename, None, hash_type="sha256") self.assertTrue(result) def test_default_hash_type_is_sha256(self): @@ -64,7 +93,115 @@ def test_default_hash_type_is_sha256(self): 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)) + self.assertTrue(check_hash(filename, sha256, hash_type="sha256")) + + def test_omitting_hash_type_emits_future_warning(self): + def assert_single_future_warning(callable_obj, *args, **kwargs): + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + callable_obj(*args, **kwargs) + + future_warnings = [w for w in recorded if issubclass(w.category, FutureWarning)] + self.assertEqual(len(future_warnings), 1) + self.assertIn('hash_type="md5"', str(future_warnings[0].message)) + self.assertIn('hash_type="sha256"', str(future_warnings[0].message)) + + with tempfile.TemporaryDirectory() as tempdir: + filename = os.path.join(tempdir, "fixture.bin") + self._write_file(filename) + sha256 = self._hash_file(filename, "sha256") + + download_target = os.path.join(tempdir, "downloaded.bin") + archive = self._create_zip_fixture(tempdir) + archive_sha256 = self._hash_file(archive, "sha256") + + assert_single_future_warning(check_hash, filename, sha256) + with patch("monai.apps.utils._download_with_progress", side_effect=self._download_side_effect(filename)): + assert_single_future_warning( + download_url, "https://example.com/fixture.bin", download_target, hash_val=sha256, progress=False + ) + assert_single_future_warning(extractall, archive, os.path.join(tempdir, "extract"), hash_val=archive_sha256) + with ( + patch("monai.apps.utils._download_with_progress", side_effect=self._download_side_effect(archive)), + patch("monai.apps.utils.get_filename_from_url", return_value="fixture.zip"), + ): + assert_single_future_warning( + download_and_extract, + "https://example.com/fixture.zip", + filepath=os.path.join(tempdir, "downloaded.zip"), + output_dir=os.path.join(tempdir, "download_and_extract"), + hash_val=archive_sha256, + progress=False, + ) + + def test_explicit_sha256_does_not_emit_default_change_warning(self): + with tempfile.TemporaryDirectory() as tempdir: + filename = os.path.join(tempdir, "fixture.bin") + self._write_file(filename) + sha256 = self._hash_file(filename, "sha256") + + download_target = os.path.join(tempdir, "downloaded.bin") + archive = self._create_zip_fixture(tempdir) + archive_sha256 = self._hash_file(archive, "sha256") + + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + self.assertTrue(check_hash(filename, sha256, hash_type="sha256")) + with patch( + "monai.apps.utils._download_with_progress", side_effect=self._download_side_effect(filename) + ): + download_url( + "https://example.com/fixture.bin", + download_target, + hash_val=sha256, + hash_type="sha256", + progress=False, + ) + extractall(archive, os.path.join(tempdir, "extract"), hash_val=archive_sha256, hash_type="sha256") + with ( + patch("monai.apps.utils._download_with_progress", side_effect=self._download_side_effect(archive)), + patch("monai.apps.utils.get_filename_from_url", return_value="fixture.zip"), + ): + download_and_extract( + "https://example.com/fixture.zip", + filepath=os.path.join(tempdir, "downloaded.zip"), + output_dir=os.path.join(tempdir, "download_and_extract"), + hash_val=archive_sha256, + hash_type="sha256", + progress=False, + ) + + future_warnings = [w for w in recorded if issubclass(w.category, FutureWarning)] + self.assertEqual(future_warnings, []) + + def test_explicit_md5_still_works(self): + with tempfile.TemporaryDirectory() as tempdir: + filename = os.path.join(tempdir, "fixture.bin") + self._write_file(filename) + md5 = self._hash_file(filename, "md5") + + download_target = os.path.join(tempdir, "downloaded.bin") + archive = self._create_zip_fixture(tempdir) + archive_md5 = self._hash_file(archive, "md5") + + self.assertTrue(check_hash(filename, md5, hash_type="md5")) + with patch("monai.apps.utils._download_with_progress", side_effect=self._download_side_effect(filename)): + download_url( + "https://example.com/fixture.bin", download_target, hash_val=md5, hash_type="md5", progress=False + ) + extractall(archive, os.path.join(tempdir, "extract"), hash_val=archive_md5, hash_type="md5") + with ( + patch("monai.apps.utils._download_with_progress", side_effect=self._download_side_effect(archive)), + patch("monai.apps.utils.get_filename_from_url", return_value="fixture.zip"), + ): + download_and_extract( + "https://example.com/fixture.zip", + filepath=os.path.join(tempdir, "downloaded.zip"), + output_dir=os.path.join(tempdir, "download_and_extract"), + hash_val=archive_md5, + hash_type="md5", + progress=False, + ) if __name__ == "__main__":