[Storage] az storage blob download-batch: Ensure downloaded blobs stay within the destination directory

src/azure-cli/azure/cli/command_modules/storage/operations/blob.py

@@ -442,6 +442,23 @@ def action_file_copy(file_info):
     raise ValueError('Fail to find source. Neither blob container nor file share is specified')
 
 
+def _get_blob_download_path(destination, normalized_blob_name, blob_name):
+    """Build the local path for a blob, ensuring it cannot escape the destination directory."""
+    # Blob names are arbitrary server-controlled strings (e.g. '../../.ssh/authorized_keys', or 'C:/x' on Windows).
+    # Only accept names that normalize to a plain relative path (no drive, root or leading '.'/'..'); joining such a
+    # path onto the destination cannot leave it, without relying on case-insensitive path comparisons (Windows
+    # directories can be case-sensitive). Existing symlinks/junctions under the destination are followed on purpose,
+    # preserving user-created layouts.
+    relative_path = os.path.normpath(normalized_blob_name)
+    first_component = relative_path.split(os.path.sep)[0]
+    if os.path.splitdrive(relative_path)[0] or first_component in ('', os.path.curdir, os.path.pardir):
+        raise FileOperationError('Blob "{}" cannot be downloaded because its name resolves to a path outside the '
+                                 'destination directory "{}". Use the `--pattern` parameter to exclude it, or use '
+                                 'the `storage blob download` command to download it to an explicit file path.'
+                                 .format(blob_name, destination))
+    return os.path.join(destination, relative_path)
+
+
 # pylint: disable=unused-argument
 def storage_blob_download_batch(client, source, destination, source_container_name, pattern=None, dryrun=False,
                                 progress_callback=None, overwrite=False, **kwargs):
@@ -455,11 +472,13 @@ def _download_blob(*args, **kwargs):
     for blob_name in source_blobs:
         # remove starting path seperator and normalize
         normalized_blob_name = normalize_blob_file_path(None, blob_name)
+        # validate every blob before downloading any of them, so a malicious name fails the batch up front
+        destination_path = _get_blob_download_path(destination, normalized_blob_name, blob_name)
         if normalized_blob_name in blobs_to_download:
             raise CLIError('Multiple blobs with download path: `{}`. As a solution, use the `--pattern` parameter '
                            'to select for a subset of blobs to download OR utilize the `storage blob download` '
                            'command instead to download individual blobs.'.format(normalized_blob_name))
-        blobs_to_download[normalized_blob_name] = blob_name
+        blobs_to_download[normalized_blob_name] = (blob_name, destination_path)
 
     if dryrun:
         logger.warning('download action: from %s to %s', source, destination)
@@ -477,13 +496,12 @@ def _download_blob(*args, **kwargs):
 
     results = []
     for index, blob_normed in enumerate(blobs_to_download):
+        blob_name, destination_path = blobs_to_download[blob_normed]
         # add blob name and number to progress message
         if progress_callback:
             progress_callback.message = '{}/{}: "{}"'.format(
-                index + 1, len(blobs_to_download), blobs_to_download[blob_normed])
-        blob_client = client.get_blob_client(container=source_container_name,
-                                             blob=blobs_to_download[blob_normed])
-        destination_path = os.path.join(destination, os.path.normpath(blob_normed))
+                index + 1, len(blobs_to_download), blob_name)
+        blob_client = client.get_blob_client(container=source_container_name, blob=blob_name)
         destination_folder = os.path.dirname(destination_path)
         # Failed when there is same name for file and folder
         if os.path.isfile(destination_path) and os.path.exists(destination_folder) and not overwrite: