fix: anchor tar member validation to extract_path in sagemaker-core

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,15 +1837,15 @@ 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}")
 
         # Check files
         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)
 

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 "<base>-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
+        ("<tmp>/work" vs "<tmp>/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."""