fix(transfer_manager): Prevent path traversal in download_many_to_path (#1768)

fix(transfer_manager): Prevent path traversal in download_many_to_path

This PR addresses a security vulnerability where `download_many_to_path`
could be exploited to write files outside the intended destination
directory.

The fix ensures that the resolved path for each blob download remains
within the bounds of the user-provided `destination_directory`. If a
blob name would result in a path outside this directory (e.g., by using
`../`), a warning is issued, and that specific blob download is skipped.
This prevents directory traversal attacks.

Absolute paths in blob names (e.g., `/etc/passwd`) are now treated as
relative to the `destination_directory`, so `/etc/passwd` will be
downloaded to `destination_directory/etc/passwd`.

See b/449616593 for more details.

BREAKING CHANGE: Blobs that would resolve to a path outside the
`destination_directory` are no longer downloaded. While this is a
security fix, users relying on the previous behavior to write files
outside the target directory will see a change.

Author: chandra-siri

google/cloud/storage/transfer_manager.py

@@ -25,6 +25,7 @@
 import struct
 import base64
 import functools
+from pathlib import Path
 
 from google.api_core import exceptions
 from google.cloud.storage import Client
@@ -231,9 +232,11 @@ def upload_many(
                 executor.submit(
                     _call_method_on_maybe_pickled_blob,
                     _pickle_client(blob) if needs_pickling else blob,
-                    "_handle_filename_and_upload"
-                    if isinstance(path_or_file, str)
-                    else "_prep_and_do_upload",
+                    (
+                        "_handle_filename_and_upload"
+                        if isinstance(path_or_file, str)
+                        else "_prep_and_do_upload"
+                    ),
                     path_or_file,
                     **upload_kwargs,
                 )
@@ -259,6 +262,16 @@ def upload_many(
     return results
 
 
+def _resolve_path(target_dir, blob_path):
+    target_dir = Path(target_dir)
+    blob_path = Path(blob_path)
+    # blob_path.anchor will be '/' if `blob_path` is full path else it'll empty.
+    # This is useful to concatnate target_dir = /local/target , and blob_path =
+    # /usr/local/mybin into /local/target/usr/local/mybin
+    concatenated_path = target_dir / blob_path.relative_to(blob_path.anchor)
+    return concatenated_path.resolve()
+
+
 @_deprecate_threads_param
 def download_many(
     blob_file_pairs,
@@ -384,9 +397,11 @@ def download_many(
                 executor.submit(
                     _call_method_on_maybe_pickled_blob,
                     _pickle_client(blob) if needs_pickling else blob,
-                    "_handle_filename_and_download"
-                    if isinstance(path_or_file, str)
-                    else "_prep_and_do_download",
+                    (
+                        "_handle_filename_and_download"
+                        if isinstance(path_or_file, str)
+                        else "_prep_and_do_download"
+                    ),
                     path_or_file,
                     **download_kwargs,
                 )
@@ -618,7 +633,8 @@ def download_many_to_path(
     """Download many files concurrently by their blob names.
 
     The destination files are automatically created, with paths based on the
-    source blob_names and the destination_directory.
+    source `blob_names` and the `destination_directory`.
+
 
     The destination files are not automatically deleted if their downloads fail,
     so please check the return value of this function for any exceptions, or
@@ -629,6 +645,50 @@ def download_many_to_path(
     "images/icon.jpg" will be downloaded to a file named
     "/home/myuser/icon.jpg".
 
+
+    Note1: if the path after combining `blob_name` and `destination_directory`
+    resolves outside `destination_directory` a warning will be issued and the
+    that particular blob will NOT be downloaded. This may happen in scenarios
+    where `blob_name` contains "../"
+
+    For example,
+        consider `destination_directory` is "downloads/gcs_blobs" and
+        `blob_name` is '../hello.blob'. This blob will not be downloaded
+        because the final resolved path would be "downloads/hello.blob"
+
+
+    To give further examples, the following blobs will not be downloaded because
+    it "escapes" the "destination_directory"
+
+    "../../local/target", # skips download
+    "../escape.txt", # skips download
+    "go/four/levels/deep/../../../../../somefile1", # skips download
+    "go/four/levels/deep/../some_dir/../../../../../invalid/path1" # skips download
+
+    however the following blobs will be downloaded because the final resolved
+    destination_directory is still child of given destination_directory
+
+    "data/../sibling.txt",
+    "dir/./file.txt",
+    "go/four/levels/deep/../somefile2",
+    "go/four/levels/deep/../some_dir/valid/path1",
+    "go/four/levels/deep/../some_dir/../../../../valid/path2",
+
+    It is adviced to use other APIs such as `transfer_manager.download_many` or
+    `Blob.download_to_filename` or `Blob.download_to_file` to download such blobs.
+
+
+    Note2:
+    The resolved download_directory will always be relative to user provided
+    `destination_directory`. For example,
+
+    a `blob_name` "/etc/passwd" will be downloaded into
+        "destination_directory/etc/passwd" instead of "/etc/passwd"
+    Similarly,
+        "/tmp/my_fav_blob"  downloads to "destination_directory/tmp/my_fav_blob"
+
+
+
     :type bucket: :class:`google.cloud.storage.bucket.Bucket`
     :param bucket:
         The bucket which contains the blobs to be downloaded
@@ -646,18 +706,18 @@ def download_many_to_path(
 
     :type destination_directory: str
     :param destination_directory:
-        A string that will be prepended (with os.path.join()) to each blob_name
-        in the input list, in order to determine the destination path for that
-        blob.
+        A string that will be prepended to each blob_name in the input list, in
+        order to determine the destination path for that blob.
 
         For instance, if the destination_directory string is "/tmp/img" and a
         blob_name is "0001.jpg", with an empty blob_name_prefix, then the source
         blob "0001.jpg" will be downloaded to destination "/tmp/img/0001.jpg" .
 
         This parameter can be an empty string.
 
-        Note that this parameter allows directory traversal (e.g. "/", "../")
-        and is not intended for unsanitized end user input.
+        Note directory traversal may be possible as long as the final
+        (e.g. "/", "../") resolved path is inside "destination_directory".
+        See examples above.
 
     :type blob_name_prefix: str
     :param blob_name_prefix:
@@ -755,11 +815,22 @@ def download_many_to_path(
 
     for blob_name in blob_names:
         full_blob_name = blob_name_prefix + blob_name
-        path = os.path.join(destination_directory, blob_name)
+        resolved_path = _resolve_path(destination_directory, blob_name)
+        if not resolved_path.parent.is_relative_to(
+            Path(destination_directory).resolve()
+        ):
+            warnings.warn(
+                f"The blob {blob_name} will **NOT** be downloaded. "
+                f"The resolved destination_directory - {resolved_path.parent} - is either invalid or "
+                f"escapes user provided {Path(destination_directory).resolve()} . Please download this file separately using `download_to_filename`"
+            )
+            continue
+
+        resolved_path = str(resolved_path)
         if create_directories:
-            directory, _ = os.path.split(path)
+            directory, _ = os.path.split(resolved_path)
             os.makedirs(directory, exist_ok=True)
-        blob_file_pairs.append((bucket.blob(full_blob_name), path))
+        blob_file_pairs.append((bucket.blob(full_blob_name), resolved_path))
 
     return download_many(
         blob_file_pairs,

tests/system/test_transfer_manager.py

@@ -121,6 +121,151 @@ def test_upload_many_from_filenames_with_attributes(
     assert blob.cache_control == "no-cache"
 
 
+@pytest.mark.parametrize(
+    "blobname",
+    [
+        "../../local/target",  # skips download
+        "../escape.txt",  # skips download
+        "go/four/levels/deep/../../../../../somefile1",  # skips download
+        "go/four/levels/deep/../some_dir/../../../../../invalid/path1",  # skips download
+    ],
+)
+def test_download_many_to_path_skips_download(
+    shared_bucket, file_data, blobs_to_delete, blobname
+):
+    """
+    Test downloading blobs with traversal skipped
+    """
+    # Setup
+    BLOBNAMES = [blobname]
+
+    FILE_BLOB_PAIRS = [
+        (
+            file_data["simple"]["path"],
+            shared_bucket.blob("folder_traversal/" + blob_name),
+        )
+        for blob_name in BLOBNAMES
+    ]
+
+    results = transfer_manager.upload_many(
+        FILE_BLOB_PAIRS,
+        skip_if_exists=True,
+        deadline=DEADLINE,
+    )
+    for result in results:
+        assert result is None
+
+    blobs = list(shared_bucket.list_blobs(prefix="folder_traversal/"))
+    blobs_to_delete.extend(blobs)
+
+    # We expect 1 blob uploaded for this test parametrization
+    assert len(list(b for b in blobs if b.name == "folder_traversal/" + blobname)) == 1
+
+    # Actual Test
+    with tempfile.TemporaryDirectory() as tempdir:
+        import warnings
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            results = transfer_manager.download_many_to_path(
+                shared_bucket,
+                BLOBNAMES,
+                destination_directory=tempdir,
+                blob_name_prefix="folder_traversal/",
+                deadline=DEADLINE,
+                create_directories=True,
+            )
+
+        path_traversal_warnings = [
+            warning
+            for warning in w
+            if str(warning.message).startswith("The blob ")
+            and "will **NOT** be downloaded. The resolved destination_directory"
+            in str(warning.message)
+        ]
+        assert len(path_traversal_warnings) == 1, "---".join(
+            [str(warning.message) for warning in w]
+        )
+
+        # 1 total - 1 skipped = 0 results
+        assert len(results) == 0
+
+
+@pytest.mark.parametrize(
+    "blobname",
+    [
+        "simple_blob",
+        "data/file.txt",
+        "data/../sibling.txt",
+        "/etc/passwd",
+        "/local/usr/a.txt",
+        "dir/./file.txt",
+        "go/four/levels/deep/../somefile2",
+        "go/four/levels/deep/../some_dir/valid/path1",
+        "go/four/levels/deep/../some_dir/../../../../valid/path2",
+    ],
+)
+def test_download_many_to_path_downloads_within_dest_dir(
+    shared_bucket, file_data, blobs_to_delete, blobname
+):
+    """
+    Test downloading blobs with valid traversal
+    """
+    # Setup
+    BLOBNAMES = [blobname]
+
+    FILE_BLOB_PAIRS = [
+        (
+            file_data["simple"]["path"],
+            shared_bucket.blob("folder_traversal/" + blob_name),
+        )
+        for blob_name in BLOBNAMES
+    ]
+
+    results = transfer_manager.upload_many(
+        FILE_BLOB_PAIRS,
+        skip_if_exists=True,
+        deadline=DEADLINE,
+    )
+    for result in results:
+        assert result is None
+
+    blobs = list(shared_bucket.list_blobs(prefix="folder_traversal/"))
+    blobs_to_delete.extend(blobs)
+
+    assert len(list(b for b in blobs if b.name == "folder_traversal/" + blobname)) == 1
+
+    # Actual Test
+    with tempfile.TemporaryDirectory() as tempdir:
+        results = transfer_manager.download_many_to_path(
+            shared_bucket,
+            BLOBNAMES,
+            destination_directory=tempdir,
+            blob_name_prefix="folder_traversal/",
+            deadline=DEADLINE,
+            create_directories=True,
+        )
+
+        assert len(results) == 1
+        for result in results:
+            assert result is None
+
+        # Verify the file exists and contents match
+        from google.cloud.storage.transfer_manager import _resolve_path
+        from pathlib import Path
+
+        expected_file_path = Path(_resolve_path(tempdir, blobname))
+        assert expected_file_path.is_file()
+
+        with open(file_data["simple"]["path"], "rb") as source_file:
+            source_contents = source_file.read()
+
+        with expected_file_path.open("rb") as downloaded_file:
+            downloaded_contents = downloaded_file.read()
+
+        assert downloaded_contents == source_contents
+
+
 def test_download_many(listable_bucket):
     blobs = list(listable_bucket.list_blobs())
     with tempfile.TemporaryDirectory() as tempdir: