CVE-2025-12060

@@ -943,7 +943,7 @@ def __init__(self, root_path, archive=None, mode=None):
         if self.archive:
             self.tmp_dir = get_temp_dir()
             if self.mode == "r":
-                self.archive.extractall(path=self.tmp_dir)
+                file_utils.extract_open_archive(self.archive, self.tmp_dir)
             self.working_dir = file_utils.join(
                 self.tmp_dir, self.root_path
             ).replace("\\", "/")

@@ -2,6 +2,7 @@
 import os
 import re
 import shutil
+import sys
 import tarfile
 import tempfile
 import urllib
@@ -52,17 +53,32 @@ def is_link_in_dir(info, base):
     return is_path_in_dir(info.linkname, base_dir=tip)
 
 
-def filter_safe_paths(members):
+def filter_safe_zipinfos(members):
     base_dir = resolve_path(".")
     for finfo in members:
         valid_path = False
-        if is_path_in_dir(finfo.name, base_dir):
+        if is_path_in_dir(finfo.filename, base_dir):
             valid_path = True
             yield finfo
-        elif finfo.issym() or finfo.islnk():
+        if not valid_path:
+            warnings.warn(
+                "Skipping invalid path during archive extraction: "
+                f"'{finfo.name}'.",
+                stacklevel=2,
+            )
+
+
+def filter_safe_tarinfos(members):
+    base_dir = resolve_path(".")
+    for finfo in members:
+        valid_path = False
+        if finfo.issym() or finfo.islnk():
             if is_link_in_dir(finfo, base_dir):
                 valid_path = True
                 yield finfo
+        elif is_path_in_dir(finfo.name, base_dir):
+            valid_path = True
+            yield finfo
         if not valid_path:
             warnings.warn(
                 "Skipping invalid path during archive extraction: "
@@ -71,6 +87,35 @@ def filter_safe_paths(members):
             )
 
 
+def extract_open_archive(archive, path="."):
+    """Extracts an open tar or zip archive to the provided directory.
+
+    This function filters unsafe paths during extraction.
+
+    Args:
+        archive: The archive object, either a `TarFile` or a `ZipFile`.
+        path: Where to extract the archive file.
+    """
+    if isinstance(archive, zipfile.ZipFile):
+        # Zip archive.
+        archive.extractall(
+            path, members=filter_safe_zipinfos(archive.infolist())
+        )
+    else:
+        # Tar archive.
+        extractall_kwargs = {}
+        # The `filter="data"` option was added in Python 3.12. It became the
+        # default starting from Python 3.14. So we only specify it between
+        # those two versions.
+        if sys.version_info >= (3, 12) and sys.version_info < (3, 14):
+            extractall_kwargs = {"filter": "data"}
+        archive.extractall(
+            path,
+            members=filter_safe_tarinfos(archive),
+            **extractall_kwargs,
+        )
+
+
 def extract_archive(file_path, path=".", archive_format="auto"):
     """Extracts an archive if it matches a support format.
 
@@ -112,14 +157,7 @@ def extract_archive(file_path, path=".", archive_format="auto"):
         if is_match_fn(file_path):
             with open_fn(file_path) as archive:
                 try:
-                    if zipfile.is_zipfile(file_path):
-                        # Zip archive.
-                        archive.extractall(path)
-                    else:
-                        # Tar archive, perhaps unsafe. Filter paths.
-                        archive.extractall(
-                            path, members=filter_safe_paths(archive)
-                        )
+                    extract_open_archive(archive, path)
                 except (tarfile.TarError, RuntimeError, KeyboardInterrupt):
                     if os.path.exists(path):
                         if os.path.isfile(path):

@@ -142,7 +142,7 @@ def test_member_within_base_dir(self):
         with tarfile.open(self.tar_path, "w") as tar:
             tar.add(__file__, arcname="safe_path.txt")
         with tarfile.open(self.tar_path, "r") as tar:
-            members = list(file_utils.filter_safe_paths(tar.getmembers()))
+            members = list(file_utils.filter_safe_tarinfos(tar.getmembers()))
             self.assertEqual(len(members), 1)
             self.assertEqual(members[0].name, "safe_path.txt")
 
@@ -156,7 +156,7 @@ def test_symlink_within_base_dir(self):
         with tarfile.open(self.tar_path, "w") as tar:
             tar.add(symlink_path, arcname="symlink.txt")
         with tarfile.open(self.tar_path, "r") as tar:
-            members = list(file_utils.filter_safe_paths(tar.getmembers()))
+            members = list(file_utils.filter_safe_tarinfos(tar.getmembers()))
             self.assertEqual(len(members), 1)
             self.assertEqual(members[0].name, "symlink.txt")
         os.remove(symlink_path)
@@ -173,7 +173,7 @@ def test_invalid_path_warning(self):
             )  # Path intended to be outside of base dir
         with tarfile.open(self.tar_path, "r") as tar:
             with patch("warnings.warn") as mock_warn:
-                _ = list(file_utils.filter_safe_paths(tar.getmembers()))
+                _ = list(file_utils.filter_safe_tarinfos(tar.getmembers()))
                 warning_msg = (
                     "Skipping invalid path during archive extraction: "
                     "'../../invalid.txt'."
@@ -196,7 +196,7 @@ def test_symbolic_link_in_base_dir(self):
             tar.add(symlink_path, arcname="symlink.txt")
 
         with tarfile.open(self.tar_path, "r") as tar:
-            members = list(file_utils.filter_safe_paths(tar.getmembers()))
+            members = list(file_utils.filter_safe_tarinfos(tar.getmembers()))
             self.assertEqual(len(members), 1)
             self.assertEqual(members[0].name, "symlink.txt")
             self.assertTrue(