[v2] Validate multipart downloads using object ETags (#9493)

Author: hssyoo

.changes/next-release/bugfix-s3transfer-31636.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "s3transfer",
+  "description": "Validate ETag of stored object during multipart downloads"
+}

awscli/customizations/s3/filegenerator.py

@@ -108,6 +108,7 @@ def __init__(
         dest_type=None,
         operation_name=None,
         response_data=None,
+        etag=None,
     ):
         self.src = src
         self.dest = dest
@@ -118,6 +119,7 @@ def __init__(
         self.dest_type = dest_type
         self.operation_name = operation_name
         self.response_data = response_data
+        self.etag = etag
 
 
 class FileGenerator:
@@ -177,6 +179,7 @@ def _inject_extra_information(self, file_stat_kwargs, extra_information):
         src_type = file_stat_kwargs['src_type']
         file_stat_kwargs['size'] = extra_information['Size']
         file_stat_kwargs['last_update'] = extra_information['LastModified']
+        file_stat_kwargs['etag'] = extra_information.get('ETag')
 
         # S3 objects require the response data retrieved from HeadObject
         # and ListObject
@@ -401,4 +404,5 @@ def _list_single_object(self, s3_path):
         response['Size'] = int(response.pop('ContentLength'))
         last_update = parse(response['LastModified'])
         response['LastModified'] = last_update.astimezone(tzlocal())
+        response['ETag'] = response.pop('ETag', None)
         return s3_path, response

awscli/customizations/s3/fileinfo.py

@@ -54,6 +54,7 @@ def __init__(
         source_client=None,
         is_stream=False,
         associated_response_data=None,
+        etag=None,
     ):
         self.src = src
         self.src_type = src_type
@@ -71,6 +72,7 @@ def __init__(
         self.source_client = source_client
         self.is_stream = is_stream
         self.associated_response_data = associated_response_data
+        self.etag = etag
 
     def is_glacier_compatible(self):
         """Determines if a file info object is glacier compatible

awscli/customizations/s3/fileinfobuilder.py

@@ -47,6 +47,7 @@ def _inject_info(self, file_base):
         file_info_attr['parameters'] = self._parameters
         file_info_attr['is_stream'] = self._is_stream
         file_info_attr['associated_response_data'] = file_base.response_data
+        file_info_attr['etag'] = file_base.etag
 
         # This is a bit quirky. The below conditional hinges on the --delete
         # flag being set, which only occurs during a sync command. The source

awscli/customizations/s3/s3handler.py

@@ -37,6 +37,7 @@
     DeleteSourceFileSubscriber,
     DeleteSourceObjectSubscriber,
     DirectoryCreatorSubscriber,
+    ProvideETagSubscriber,
     ProvideLastModifiedTimeSubscriber,
     ProvideSizeSubscriber,
     ProvideUploadContentTypeSubscriber,
@@ -243,6 +244,7 @@ def _get_subscribers(self, fileinfo):
         subscribers = []
         result_subscriber_kwargs = self._get_result_subscriber_kwargs(fileinfo)
         self._add_provide_size_subscriber(subscribers, fileinfo)
+        subscribers.append(ProvideETagSubscriber(fileinfo.etag))
         subscribers.append(QueuedResultSubscriber(**result_subscriber_kwargs))
         self._add_additional_subscribers(subscribers, fileinfo)
         subscribers.extend(

awscli/customizations/s3/subscribers.py

@@ -78,6 +78,27 @@ def on_queued(self, future, **kwargs):
             )
 
 
+class ProvideETagSubscriber(BaseSubscriber):
+    """
+    A subscriber which provides the object ETag before it's queued.
+    """
+
+    def __init__(self, etag):
+        self.etag = etag
+
+    def on_queued(self, future, **kwargs):
+        # The CRT transfer client does not support providing an expected
+        # ETag of an object. So only provide it if it is available.
+        if hasattr(future.meta, 'provide_object_etag'):
+            future.meta.provide_object_etag(self.etag)
+        else:
+            LOGGER.debug(
+                'Not providing object ETag. Future: %s does not offer'
+                'the capability to notify the ETag of an object',
+                future,
+            )
+
+
 class DeleteSourceSubscriber(OnDoneFilteredSubscriber):
     """A subscriber which deletes the source of the transfer."""
 

awscli/s3transfer/download.py

@@ -14,8 +14,9 @@
 import logging
 import threading
 
+from botocore.exceptions import ClientError
 from s3transfer.compat import seekable
-from s3transfer.exceptions import RetriesExceededError
+from s3transfer.exceptions import RetriesExceededError, S3DownloadFailedError
 from s3transfer.futures import IN_MEMORY_DOWNLOAD_TAG
 from s3transfer.tasks import SubmissionTask, Task
 from s3transfer.utils import (
@@ -346,17 +347,23 @@ def _submit(
         :param bandwidth_limiter: The bandwidth limiter to use when
             downloading streams
         """
-        if transfer_future.meta.size is None:
-            # If a size was not provided figure out the size for the
-            # user.
+        if (
+            transfer_future.meta.size is None
+            or transfer_future.meta.etag is None
+        ):
             response = client.head_object(
                 Bucket=transfer_future.meta.call_args.bucket,
                 Key=transfer_future.meta.call_args.key,
                 **transfer_future.meta.call_args.extra_args,
             )
+            # If a size was not provided figure out the size for the
+            # user.
             transfer_future.meta.provide_transfer_size(
                 response['ContentLength']
             )
+            # Provide an etag to ensure a stored object is not modified
+            # during a multipart download.
+            transfer_future.meta.provide_object_etag(response.get('ETag'))
 
         download_output_manager = self._get_download_output_manager_cls(
             transfer_future, osutil
@@ -479,9 +486,12 @@ def _submit_ranged_download_request(
                 part_size, i, num_parts
             )
 
-            # Inject the Range parameter to the parameters to be passed in
-            # as extra args
-            extra_args = {'Range': range_parameter}
+            # Inject extra parameters to be passed in as extra args
+            extra_args = {
+                'Range': range_parameter,
+            }
+            if transfer_future.meta.etag is not None:
+                extra_args['IfMatch'] = transfer_future.meta.etag
             extra_args.update(call_args.extra_args)
             finalize_download_invoker.increment()
             # Submit the ranged downloads
@@ -593,6 +603,15 @@ def _main(
                     else:
                         return
                 return
+            except ClientError as e:
+                error_code = e.response.get('Error', {}).get('Code')
+                if error_code == "PreconditionFailed":
+                    raise S3DownloadFailedError(
+                        f'Contents of stored object "{key}" in bucket '
+                        f'"{bucket}" did not match expected ETag.'
+                    )
+                else:
+                    raise
             except S3_RETRYABLE_DOWNLOAD_ERRORS as e:
                 logger.debug(
                     "Retrying exception caught (%s), "

awscli/s3transfer/exceptions.py

@@ -23,6 +23,10 @@ class S3UploadFailedError(Exception):
     pass
 
 
+class S3DownloadFailedError(Exception):
+    pass
+
+
 class InvalidSubscriberMethodError(Exception):
     pass
 

awscli/s3transfer/futures.py

@@ -127,6 +127,7 @@ def __init__(self, call_args=None, transfer_id=None):
         self._transfer_id = transfer_id
         self._size = None
         self._user_context = {}
+        self._etag = None
 
     @property
     def call_args(self):
@@ -148,6 +149,11 @@ def user_context(self):
         """A dictionary that requesters can store data in"""
         return self._user_context
 
+    @property
+    def etag(self):
+        """The etag of the stored object for validating multipart downloads"""
+        return self._etag
+
     def provide_transfer_size(self, size):
         """A method to provide the size of a transfer request
 
@@ -157,6 +163,15 @@ def provide_transfer_size(self, size):
         """
         self._size = size
 
+    def provide_object_etag(self, etag):
+        """A method to provide the etag of a transfer request
+
+        By providing this value, the TransferManager will validate
+        multipart downloads by supplying an IfMatch parameter with
+        the etag as the value to GetObject requests.
+        """
+        self._etag = etag
+
 
 class TransferCoordinator:
     """A helper class for managing TransferFuture"""

tests/__init__.py

@@ -92,6 +92,7 @@
     StreamWithError,
     RecordingSubscriber,
     FileSizeProvider,
+    ETagProvider,
     RecordingOSUtils,
     RecordingExecutor,
     TransferCoordinatorWithInterrupt,