diff --git a/sagemaker-core/src/sagemaker/core/common_utils.py b/sagemaker-core/src/sagemaker/core/common_utils.py index f63136f2f3..0c8025174c 100644 --- a/sagemaker-core/src/sagemaker/core/common_utils.py +++ b/sagemaker-core/src/sagemaker/core/common_utils.py @@ -32,7 +32,7 @@ import abc import uuid from datetime import datetime -from os.path import abspath, realpath, dirname, normpath, join as joinpath +from os.path import abspath, realpath, dirname, isabs, normpath, join as joinpath from importlib import import_module @@ -1734,6 +1734,28 @@ def validate_path_within_directory(file_path, target_directory, source_descripti ) +def _is_within_base(resolved_path, base): + """Checks if an already resolved absolute path is contained within a base directory. + + Uses os.path.commonpath rather than a string prefix comparison, so that a sibling + directory which merely shares a textual prefix with the base directory (e.g. base + "/tmp/extract" and path "/tmp/extract-evil/f") is not treated as contained. + + Args: + resolved_path (str): An absolute, normalized path. + base (str): An absolute, normalized base directory. + + Returns: + bool: True if resolved_path is the base directory or nested under it. + """ + try: + return os.path.commonpath([resolved_path, base]) == base + except ValueError: + # Raised when the paths cannot be compared (e.g. different drives on Windows), + # in which case resolved_path cannot be inside base. + return False + + def _is_bad_path(path, base): """Checks if the joined path (base directory + file path) is rooted under the base directory @@ -1747,8 +1769,11 @@ def _is_bad_path(path, base): Returns: bool: True if the path is not rooted under the base directory, False otherwise. """ - # joinpath will ignore base if path is absolute - return not _get_resolved_path(joinpath(base, path)).startswith(base) + # joinpath would silently discard base for an absolute path, and an archive member + # targeting an absolute location is never legitimate, so reject it outright. + if isabs(path): + return True + return not _is_within_base(_get_resolved_path(joinpath(base, path)), base) def _is_bad_link(info, base): @@ -1768,19 +1793,20 @@ def _is_bad_link(info, base): return _is_bad_path(info.linkname, base=tip) -def _get_safe_members(members): +def _get_safe_members(members, base): """A generator that yields members that are safe to extract. It filters out bad paths and bad links. Args: members (list): A list of members to check. + base (str): The resolved base directory that members must stay within. This must + be the directory the archive is extracted into, since that is what the member + paths are resolved against at extraction time. Yields: tarfile.TarInfo: The tar file info. """ - base = _get_resolved_path("") - for file_info in members: if _is_bad_path(file_info.name, base): logger.error("%s is blocked (illegal path)", file_info.name) @@ -1811,7 +1837,7 @@ def _validate_extracted_paths(extract_path): for dir_name in dirs: dir_path = os.path.join(root, dir_name) resolved = _get_resolved_path(dir_path) - if not resolved.startswith(base): + if not _is_within_base(resolved, base): logger.error("Extracted directory escaped extraction path: %s", dir_path) raise ValueError(f"Extracted path outside expected directory: {dir_path}") @@ -1819,7 +1845,7 @@ def _validate_extracted_paths(extract_path): for file_name in files: file_path = os.path.join(root, file_name) resolved = _get_resolved_path(file_path) - if not resolved.startswith(base): + if not _is_within_base(resolved, base): logger.error("Extracted file escaped extraction path: %s", file_path) raise ValueError(f"Extracted path outside expected directory: {file_path}") @@ -1843,7 +1869,10 @@ def custom_extractall_tarfile(tar, extract_path): if hasattr(tarfile, "data_filter"): tar.extractall(path=extract_path, filter="data") else: - tar.extractall(path=extract_path, members=_get_safe_members(tar)) + # Members are resolved against the directory they are extracted into, so that is + # what containment has to be checked against. + base = _get_resolved_path(extract_path) + tar.extractall(path=extract_path, members=_get_safe_members(tar.getmembers(), base)) # Re-validate extracted paths to catch symlink race conditions _validate_extracted_paths(extract_path) diff --git a/sagemaker-core/tests/unit/test_common_utils.py b/sagemaker-core/tests/unit/test_common_utils.py index 3f7fc94f67..64114e8d55 100644 --- a/sagemaker-core/tests/unit/test_common_utils.py +++ b/sagemaker-core/tests/unit/test_common_utils.py @@ -1228,6 +1228,173 @@ def test_custom_extractall_tarfile_basic(self, tmp_path): assert (extract_path / "file.txt").exists() +class TestTarExtractionPathTraversal: + """Regression tests for path traversal in the pre-3.12 tar extraction fallback. + + The fallback branch of custom_extractall_tarfile runs only when + tarfile.data_filter is unavailable (Python < 3.12, before the 3.9.17 / 3.10.12 / + 3.11.4 backports). These tests force that branch so the member filtering is + exercised regardless of the interpreter the suite runs on. + """ + + @staticmethod + def _no_data_filter(): + """Patch the module's tarfile reference with one that has no data_filter.""" + from types import SimpleNamespace + + return patch("sagemaker.core.common_utils.tarfile", SimpleNamespace()) + + @staticmethod + def _tar_with_member(tar_path, member_name, content=b"pwned"): + """Write a tar archive containing a single member under an arbitrary name.""" + import io + + with tarfile.open(tar_path, "w:gz") as tar: + info = tarfile.TarInfo(name=member_name) + info.size = len(content) + tar.addfile(info, io.BytesIO(content)) + + def test_is_within_base_containment(self, tmp_path): + """_is_within_base accepts the base and nested paths, rejects outside paths.""" + from sagemaker.core.common_utils import _get_resolved_path, _is_within_base + + base = _get_resolved_path(str(tmp_path / "extract")) + + assert _is_within_base(base, base) is True + assert _is_within_base(_get_resolved_path(str(tmp_path / "extract" / "a")), base) is True + assert _is_within_base(_get_resolved_path(str(tmp_path / "other")), base) is False + + def test_is_within_base_rejects_sibling_prefix(self, tmp_path): + """A sibling directory sharing a textual prefix with base is not contained. + + A plain startswith() comparison would accept "-evil" because it is a + string prefix match. + """ + from sagemaker.core.common_utils import _get_resolved_path, _is_within_base + + base = _get_resolved_path(str(tmp_path / "extract")) + sibling = _get_resolved_path(str(tmp_path / "extract-evil" / "f.txt")) + + assert sibling.startswith(base) # the bug a prefix check would let through + assert _is_within_base(sibling, base) is False + + def test_is_bad_path_rejects_sibling_prefix(self, tmp_path): + """_is_bad_path blocks a member escaping into a prefix-sharing sibling dir.""" + from sagemaker.core.common_utils import _get_resolved_path, _is_bad_path + + base = _get_resolved_path(str(tmp_path / "extract")) + + assert _is_bad_path("../extract-evil/f.txt", base) is True + + def test_is_bad_path_rejects_absolute_member(self, tmp_path): + """_is_bad_path blocks absolute member paths outright.""" + from sagemaker.core.common_utils import _get_resolved_path, _is_bad_path + + base = _get_resolved_path(str(tmp_path / "extract")) + + assert _is_bad_path("/etc/passwd", base) is True + + def test_is_bad_path_allows_nested_member(self, tmp_path): + """_is_bad_path permits ordinary members nested under the base directory.""" + from sagemaker.core.common_utils import _get_resolved_path, _is_bad_path + + base = _get_resolved_path(str(tmp_path / "extract")) + + assert _is_bad_path("code/inference.py", base) is False + + def test_get_safe_members_filters_member_escaping_extract_path(self, tmp_path): + """_get_safe_members blocks a member that escapes the base it is given.""" + from sagemaker.core.common_utils import _get_resolved_path, _get_safe_members + + base = _get_resolved_path(str(tmp_path / "target" / "extract")) + escaping = tarfile.TarInfo(name="../extract-evil/escaped.txt") + benign = tarfile.TarInfo(name="model.tar") + + safe = list(_get_safe_members([escaping, benign], base)) + + assert [m.name for m in safe] == ["model.tar"] + + def test_members_are_validated_against_extract_path(self, tmp_path): + """Members must be validated against extract_path, not the working directory.""" + from sagemaker.core.common_utils import _get_resolved_path, custom_extractall_tarfile + + extract_path = tmp_path / "extract" + extract_path.mkdir() + mock_tar = Mock() + mock_tar.getmembers = Mock(return_value=[]) + + with self._no_data_filter(): + with patch("sagemaker.core.common_utils._get_safe_members") as mock_safe: + mock_safe.return_value = [] + custom_extractall_tarfile(mock_tar, str(extract_path)) + + assert mock_safe.call_args[0][1] == _get_resolved_path(str(extract_path)) + + def test_fallback_extraction_blocks_escape_outside_extract_path(self, tmp_path, monkeypatch): + """End-to-end: a crafted member must not be written outside extract_path. + + Mirrors the reported proof of concept. The member escapes into a directory whose + name shares a textual prefix with the working directory's path + ("/work" vs "/workevil"), so a startswith() check anchored to the + working directory accepts it while extraction still writes it outside + extract_path. _validate_extracted_paths only walks extract_path, so it does not + catch the escape either. + """ + from sagemaker.core.common_utils import custom_extractall_tarfile + + cwd = tmp_path / "work" + cwd.mkdir() + monkeypatch.chdir(cwd) + + extract_path = tmp_path / "target" / "extract" + extract_path.mkdir(parents=True) + + tar_path = tmp_path / "malicious.tar.gz" + self._tar_with_member(tar_path, "../workevil/escaped.txt") + + escaped = tmp_path / "target" / "workevil" / "escaped.txt" + + with tarfile.open(tar_path, "r:gz") as tar: + with self._no_data_filter(): + custom_extractall_tarfile(tar, str(extract_path)) + + assert not escaped.exists(), "member escaped the extraction directory" + assert list(extract_path.rglob("*")) == [] + + def test_fallback_extraction_blocks_absolute_member(self, tmp_path): + """An absolute member path must not be written to its absolute location.""" + from sagemaker.core.common_utils import custom_extractall_tarfile + + outside = tmp_path / "absolute_target.txt" + extract_path = tmp_path / "extract" + extract_path.mkdir() + + tar_path = tmp_path / "absolute.tar.gz" + self._tar_with_member(tar_path, str(outside)) + + with tarfile.open(tar_path, "r:gz") as tar: + with self._no_data_filter(): + custom_extractall_tarfile(tar, str(extract_path)) + + assert not outside.exists() + + def test_fallback_extraction_allows_benign_archive(self, tmp_path): + """The fallback branch still extracts legitimate nested members.""" + from sagemaker.core.common_utils import custom_extractall_tarfile + + extract_path = tmp_path / "extract" + extract_path.mkdir() + + tar_path = tmp_path / "benign.tar.gz" + self._tar_with_member(tar_path, "code/inference.py", content=b"print('hi')") + + with tarfile.open(tar_path, "r:gz") as tar: + with self._no_data_filter(): + custom_extractall_tarfile(tar, str(extract_path)) + + assert (extract_path / "code" / "inference.py").read_bytes() == b"print('hi')" + + class TestCanModelPackageSourceUriAutopopulate: """Test can_model_package_source_uri_autopopulate function."""