Support following S3 bucket region redirects when the CRT client is enabled

.changes/next-release/enhancement-s3-42614.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "``s3``",
+  "description": "Follow bucket region redirects for the ``s3`` commands when CRT is enabled."
+}

awscli/botocore/utils.py

@@ -1576,7 +1576,7 @@ class S3RegionRedirectorv2:
     """
 
     def __init__(self, endpoint_bridge, client, cache=None):
-        self._cache = cache or {}
+        self._cache = {} if cache is None else cache
         self._client = weakref.proxy(client)
 
     def register(self, event_emitter=None):
@@ -1616,49 +1616,9 @@ def redirect_from_error(self, request_dict, response, operation, **kwargs):
             )
             return
 
-        error = response[1].get('Error', {})
-        error_code = error.get('Code')
-        response_metadata = response[1].get('ResponseMetadata', {})
-
-        # We have to account for 400 responses because
-        # if we sign a Head* request with the wrong region,
-        # we'll get a 400 Bad Request but we won't get a
-        # body saying it's an "AuthorizationHeaderMalformed".
-        is_special_head_object = (
-            error_code in ('301', '400') and operation.name == 'HeadObject'
-        )
-        is_special_head_bucket = (
-            error_code in ('301', '400')
-            and operation.name == 'HeadBucket'
-            and 'x-amz-bucket-region'
-            in response_metadata.get('HTTPHeaders', {})
-        )
-        is_wrong_signing_region = (
-            error_code == 'AuthorizationHeaderMalformed' and 'Region' in error
-        )
-        is_redirect_status = response[0] is not None and response[
-            0
-        ].status_code in (301, 302, 307)
-        is_permanent_redirect = error_code == 'PermanentRedirect'
-        is_opt_in_region_redirect = (
-            error_code == 'IllegalLocationConstraintException'
-            and operation.name != 'CreateBucket'
-        )
-        if not any(
-            [
-                is_special_head_object,
-                is_wrong_signing_region,
-                is_permanent_redirect,
-                is_special_head_bucket,
-                is_redirect_status,
-                is_opt_in_region_redirect,
-            ]
-        ):
-            return
-
-        bucket = request_dict['context']['s3_redirect']['bucket']
-        client_region = request_dict['context'].get('client_region')
-        new_region = self.get_bucket_region(bucket, response)
+        bucket = redirect_ctx.get('bucket')
+        client_region = request_dict.get('context', {}).get('client_region')
+        new_region = self.get_redirect_region(bucket, response, operation)
 
         if new_region is None:
             logger.debug(
@@ -1702,6 +1662,52 @@ def redirect_from_error(self, request_dict, response, operation, **kwargs):
         # Return 0 so it doesn't wait to retry
         return 0
 
+    def get_redirect_region(self, bucket, response, operation):
+        if bucket is None:
+            return None
+        if ArnParser.is_arn(bucket):
+            return None
+        error = response[1].get('Error', {})
+        error_code = error.get('Code')
+        response_metadata = response[1].get('ResponseMetadata', {})
+
+        # We have to account for 400 responses because
+        # if we sign a Head* request with the wrong region,
+        # we'll get a 400 Bad Request but we won't get a
+        # body saying it's an "AuthorizationHeaderMalformed".
+        is_special_head_object = (
+            error_code in ('301', '400') and operation.name == 'HeadObject'
+        )
+        is_special_head_bucket = (
+            error_code in ('301', '400')
+            and operation.name == 'HeadBucket'
+            and 'x-amz-bucket-region'
+            in response_metadata.get('HTTPHeaders', {})
+        )
+        is_wrong_signing_region = (
+            error_code == 'AuthorizationHeaderMalformed' and 'Region' in error
+        )
+        is_redirect_status = response[0] is not None and response[
+            0
+        ].status_code in (301, 302, 307)
+        is_permanent_redirect = error_code == 'PermanentRedirect'
+        is_opt_in_region_redirect = (
+            error_code == 'IllegalLocationConstraintException'
+            and operation.name != 'CreateBucket'
+        )
+        if not any(
+            [
+                is_special_head_object,
+                is_wrong_signing_region,
+                is_permanent_redirect,
+                is_special_head_bucket,
+                is_redirect_status,
+                is_opt_in_region_redirect,
+            ]
+        ):
+            return None
+        return self.get_bucket_region(bucket, response)
+
     def get_bucket_region(self, bucket, response):
         """
         There are multiple potential sources for the new region to redirect to,

awscli/customizations/s3/factory.py

@@ -20,6 +20,7 @@
     BotocoreCRTRequestSerializer,
     CRTTransferManager,
     acquire_crt_s3_process_lock,
+    create_crt_client_bootstrap,
     create_s3_crt_client,
 )
 from s3transfer.manager import TransferManager
@@ -114,16 +115,34 @@ def _acquire_crt_s3_process_lock(self):
 
     def _create_crt_transfer_manager(self, params, runtime_config):
         self._acquire_crt_s3_process_lock()
-        return CRTTransferManager(
-            self._create_crt_client(params, runtime_config),
-            self._create_crt_request_serializer(params),
+        region = self._resolve_region(params)
+        bootstrap = create_crt_client_bootstrap()
+        transfer_manager = CRTTransferManager(
+            crt_client_factory=lambda client_region=None: (
+                self._create_crt_client(
+                    params,
+                    runtime_config,
+                    region=client_region or region,
+                    bootstrap=bootstrap,
+                )
+            ),
+            crt_request_serializer=self._create_crt_request_serializer(params),
         )
-
-    def _create_crt_client(self, params, runtime_config):
+        # Clients for redirected regions are created on demand, but create the
+        # one for the configured region now. Otherwise invalid client
+        # configuration is not reported until a transfer is submitted, which
+        # reports it once per object instead of once for the command.
+        transfer_manager.get_crt_client()
+        return transfer_manager
+
+    def _create_crt_client(
+        self, params, runtime_config, region=None, bootstrap=None
+    ):
         config_file_params = self._session.get_scoped_config().get('s3', {})
         create_crt_client_kwargs = {
-            'region': self._resolve_region(params),
+            'region': region or self._resolve_region(params),
             'verify': self._resolve_verify(params),
+            'bootstrap': bootstrap,
         }
         endpoint_url = params.get('endpoint_url')
         if endpoint_url and urlparse.urlparse(endpoint_url).scheme == 'http':
@@ -161,6 +180,9 @@ def _create_crt_request_serializer(self, params):
                 'region_name': self._resolve_region(params),
                 'endpoint_url': params.get('endpoint_url'),
             },
+            region_redirect_client_factory=lambda: (
+                self._botocore_client_factory.create_client(params)
+            ),
         )
 
     def _create_classic_transfer_manager(