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."""