CVE-2026-40491

@@ -12,6 +12,7 @@
 import bs4
 
 from .download import _get_session
+from .download import _sanitize_filename
 from .download import download
 from .exceptions import FolderContentsMaximumLimitError
 from .parse_url import is_google_drive_url
@@ -182,7 +183,7 @@ def _get_directory_structure(gdrive_file, previous_path):
 
     directory_structure = []
     for file in gdrive_file.children:
-        file.name = file.name.replace(osp.sep, "_")
+        file.name = _sanitize_filename(file.name)
         if file.is_folder():
             directory_structure.append((None, osp.join(previous_path, file.name)))
             for i in _get_directory_structure(file, osp.join(previous_path, file.name)):
@@ -283,6 +284,8 @@ def download_folder(
         print("Failed to retrieve folder contents", file=sys.stderr)
         return None
 
+    gdrive_file.name = _sanitize_filename(gdrive_file.name)
+
     if not quiet:
         print("Retrieving folder contents completed", file=sys.stderr)
         print("Building directory structure", file=sys.stderr)

@@ -61,18 +61,24 @@ def get_url_from_gdrive_confirmation(contents):
     return url
 
 
+def _sanitize_filename(filename):
+    filename = filename.replace("\x00", "")
+    filename = filename.replace("/", "_").replace("\\", "_").strip()
+    if filename in ("", ".", ".."):
+        return "_"
+    return filename
+
+
 def _get_filename_from_response(response):
     content_disposition = urllib.parse.unquote(response.headers["Content-Disposition"])
 
     m = re.search(r"filename\*=UTF-8''(.*)", content_disposition)
     if m:
-        filename = m.groups()[0]
-        return filename.replace(osp.sep, "_")
+        return _sanitize_filename(m.groups()[0])
 
     m = re.search('attachment; filename="(.*?)"', content_disposition)
     if m:
-        filename = m.groups()[0]
-        return filename
+        return _sanitize_filename(m.groups()[0])
 
     return None
 
@@ -283,7 +289,7 @@ def download(
         filename_from_url = _get_filename_from_response(response=res)
         last_modified_time = _get_modified_time_from_response(response=res)
     if filename_from_url is None:
-        filename_from_url = osp.basename(url)
+        filename_from_url = _sanitize_filename(osp.basename(url))
 
     if output is None:
         output = filename_from_url

@@ -1,8 +1,16 @@
+import os
 import os.path as osp
+import sys
 import tarfile
 import zipfile
 
 
+def _is_within_directory(directory, target):
+    abs_directory = osp.realpath(directory)
+    abs_target = osp.realpath(target)
+    return abs_target.startswith(abs_directory + os.sep) or abs_target == abs_directory
+
+
 def extractall(path, to=None):
     """Extract archive file.
 
@@ -18,31 +26,61 @@ def extractall(path, to=None):
         to = osp.dirname(path)
 
     if path.endswith(".zip"):
-        opener, mode = zipfile.ZipFile, "r"
-    elif path.endswith(".tar"):
-        opener, mode = tarfile.open, "r"
+        return _extractall_zip(path, to)
+
+    if path.endswith(".tar"):
+        tar_mode = "r"
     elif path.endswith(".tar.gz") or path.endswith(".tgz"):
-        opener, mode = tarfile.open, "r:gz"
+        tar_mode = "r:gz"
     elif path.endswith(".tar.bz2") or path.endswith(".tbz"):
-        opener, mode = tarfile.open, "r:bz2"
+        tar_mode = "r:bz2"
     else:
         raise ValueError(
             "Could not extract '%s' as no appropriate extractor is found" % path
         )
 
-    def namelist(f):
-        if isinstance(f, zipfile.ZipFile):
-            return f.namelist()
-        return [m.path for m in f.members]
+    return _extractall_tar(path, to, tar_mode)
 
-    def filelist(f):
-        files = []
-        for fname in namelist(f):
-            fname = osp.join(to, fname)
-            files.append(fname)
-        return files
 
-    with opener(path, mode) as f:
+def _extractall_zip(path, to):
+    with zipfile.ZipFile(path, "r") as f:
+        names = f.namelist()
+        for member in names:
+            member_path = osp.join(to, member)
+            if not _is_within_directory(to, member_path):
+                raise ValueError(
+                    "Archive member '%s' would extract outside "
+                    "target directory: %s" % (member, to)
+                )
         f.extractall(path=to)
+    return [osp.join(to, name) for name in names]
+
+
+def _extractall_tar(path, to, tar_mode):
+    with tarfile.open(path, tar_mode) as f:
+        members = f.getmembers()
+        if sys.version_info >= (3, 12):
+            f.extractall(path=to, filter="data")
+        else:
+            for member in members:
+                if member.issym() or member.islnk():
+                    raise ValueError(
+                        "Archive member '%s' is a link, "
+                        "which is not allowed for security reasons"
+                        % member.name
+                    )
+                if member.ischr() or member.isblk() or member.isfifo():
+                    raise ValueError(
+                        "Archive member '%s' is a special file, "
+                        "which is not allowed for security reasons"
+                        % member.name
+                    )
+                member_path = osp.join(to, member.name)
+                if not _is_within_directory(to, member_path):
+                    raise ValueError(
+                        "Archive member '%s' would extract outside "
+                        "target directory: %s" % (member.name, to)
+                    )
+            f.extractall(path=to)
 
-    return filelist(f)
+    return [osp.join(to, m.path) for m in members]