Path Traversal in zenml-io/zenml

@@ -29,6 +29,30 @@
 from zenml.utils.io_utils import is_path_within_directory
 
 
+def _is_safe_tar_member(member: tarfile.TarInfo, directory: str) -> bool:
+    """Check if a tar member is safe to extract.
+
+    This function validates that the member name and any link targets
+    are within the specified directory to prevent path traversal attacks.
+
+    Args:
+        member: The tar member to validate.
+        directory: The target extraction directory.
+
+    Returns:
+        True if the member is safe to extract, False otherwise.
+    """
+    # Check if the member name is within the directory
+    if not is_path_within_directory(member.name, directory):
+        return False
+
+    # For symbolic links and hard links, validate the target path
+    if member.issym() or member.islnk():
+        return is_path_within_directory(member.linkname, directory)
+
+    return True
+
+
 class PathMaterializer(BaseMaterializer):
     """Materializer for Path objects.
 
@@ -73,14 +97,14 @@ def load(self, data_type: Type[Any]) -> Any:
                 # Extract the archive to the temporary directory
                 with tarfile.open(archive_path_local, "r:gz") as tar:
                     # Validate archive members to prevent path traversal attacks
-                    # Filter members to only those with safe paths
+                    # Filter members to only those with safe paths and link targets
                     safe_members = []
                     for member in tar.getmembers():
-                        if is_path_within_directory(member.name, directory):
+                        if _is_safe_tar_member(member, directory):
                             safe_members.append(member)
 
                     # Extract only safe members
-                    tar.extractall(path=directory, members=safe_members)  # nosec B202 - members are filtered through is_path_within_directory
+                    tar.extractall(path=directory, members=safe_members)  # nosec B202 - members are filtered through _is_safe_tar_member
 
                 # Clean up the archive file
                 os.remove(archive_path_local)

@@ -13,11 +13,15 @@
 #  permissions and limitations under the License.
 """Tests for the PathMaterializer."""
 
+import tarfile
 import tempfile
 from pathlib import Path
 
 from tests.unit.test_general import _test_materializer
-from zenml.materializers.path_materializer import PathMaterializer
+from zenml.materializers.path_materializer import (
+    PathMaterializer,
+    _is_safe_tar_member,
+)
 
 
 def test_path_materializer():
@@ -120,3 +124,99 @@ def test_path_materializer_with_binary_file():
         with open(result, "rb") as f:
             content = f.read()
         assert content == b"\x00\x01\x02\x03\xff\xfe"
+
+
+def test_is_safe_tar_member_regular_files():
+    """Test that regular files are validated correctly."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        # Safe regular file
+        safe_file = tarfile.TarInfo("safe_file.txt")
+        safe_file.type = tarfile.REGTYPE
+        assert _is_safe_tar_member(safe_file, temp_dir)
+
+        # Unsafe regular file with path traversal
+        unsafe_file = tarfile.TarInfo("../../../etc/passwd")
+        unsafe_file.type = tarfile.REGTYPE
+        assert not _is_safe_tar_member(unsafe_file, temp_dir)
+
+        # Safe nested file
+        nested_file = tarfile.TarInfo("subdir/nested_file.txt")
+        nested_file.type = tarfile.REGTYPE
+        assert _is_safe_tar_member(nested_file, temp_dir)
+
+
+def test_is_safe_tar_member_symbolic_links():
+    """Test that symbolic links are validated correctly."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        # Safe symbolic link pointing within directory
+        safe_symlink = tarfile.TarInfo("safe_symlink.txt")
+        safe_symlink.type = tarfile.SYMTYPE
+        safe_symlink.linkname = "target_file.txt"
+        assert _is_safe_tar_member(safe_symlink, temp_dir)
+
+        # Safe symbolic link pointing to subdirectory file
+        safe_nested_symlink = tarfile.TarInfo("safe_nested_symlink.txt")
+        safe_nested_symlink.type = tarfile.SYMTYPE
+        safe_nested_symlink.linkname = "subdir/target.txt"
+        assert _is_safe_tar_member(safe_nested_symlink, temp_dir)
+
+        # Unsafe symbolic link pointing outside directory
+        unsafe_symlink = tarfile.TarInfo("unsafe_symlink.txt")
+        unsafe_symlink.type = tarfile.SYMTYPE
+        unsafe_symlink.linkname = "../../../etc/passwd"
+        assert not _is_safe_tar_member(unsafe_symlink, temp_dir)
+
+        # Unsafe symbolic link with absolute path
+        absolute_symlink = tarfile.TarInfo("absolute_symlink.txt")
+        absolute_symlink.type = tarfile.SYMTYPE
+        absolute_symlink.linkname = "/etc/passwd"
+        assert not _is_safe_tar_member(absolute_symlink, temp_dir)
+
+
+def test_is_safe_tar_member_hard_links():
+    """Test that hard links are validated correctly."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        # Safe hard link pointing within directory
+        safe_hardlink = tarfile.TarInfo("safe_hardlink.txt")
+        safe_hardlink.type = tarfile.LNKTYPE
+        safe_hardlink.linkname = "target_file.txt"
+        assert _is_safe_tar_member(safe_hardlink, temp_dir)
+
+        # Safe hard link pointing to subdirectory file
+        safe_nested_hardlink = tarfile.TarInfo("safe_nested_hardlink.txt")
+        safe_nested_hardlink.type = tarfile.LNKTYPE
+        safe_nested_hardlink.linkname = "subdir/target.txt"
+        assert _is_safe_tar_member(safe_nested_hardlink, temp_dir)
+
+        # Unsafe hard link pointing outside directory
+        unsafe_hardlink = tarfile.TarInfo("unsafe_hardlink.txt")
+        unsafe_hardlink.type = tarfile.LNKTYPE
+        unsafe_hardlink.linkname = "../../../etc/passwd"
+        assert not _is_safe_tar_member(unsafe_hardlink, temp_dir)
+
+        # Unsafe hard link with absolute path
+        absolute_hardlink = tarfile.TarInfo("absolute_hardlink.txt")
+        absolute_hardlink.type = tarfile.LNKTYPE
+        absolute_hardlink.linkname = "/etc/passwd"
+        assert not _is_safe_tar_member(absolute_hardlink, temp_dir)
+
+
+def test_is_safe_tar_member_mixed_scenarios():
+    """Test edge cases and mixed scenarios."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        # File with safe name but unsafe symlink name should be rejected
+        mixed_unsafe = tarfile.TarInfo("../unsafe_name.txt")
+        mixed_unsafe.type = tarfile.SYMTYPE
+        mixed_unsafe.linkname = "safe_target.txt"
+        assert not _is_safe_tar_member(mixed_unsafe, temp_dir)
+
+        # File with safe name but unsafe hardlink target should be rejected
+        safe_name_unsafe_target = tarfile.TarInfo("safe_name.txt")
+        safe_name_unsafe_target.type = tarfile.LNKTYPE
+        safe_name_unsafe_target.linkname = "../unsafe_target.txt"
+        assert not _is_safe_tar_member(safe_name_unsafe_target, temp_dir)
+
+        # Directory entry should be safe if name is safe
+        safe_directory = tarfile.TarInfo("safe_dir/")
+        safe_directory.type = tarfile.DIRTYPE
+        assert _is_safe_tar_member(safe_directory, temp_dir)