Validate region name from S3 redirect (boto#3644)

Author: hssyoo

.changes/next-release/enhancement-S3redirect-45693.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "S3 redirect",
+  "description": "Validate new region name when redirecting."
+}

botocore/utils.py

@@ -1887,21 +1887,19 @@ def get_bucket_region(self, bucket, response):
         service_response = response[1]
         response_headers = service_response['ResponseMetadata']['HTTPHeaders']
         if 'x-amz-bucket-region' in response_headers:
-            return response_headers['x-amz-bucket-region']
-
+            region = response_headers['x-amz-bucket-region']
         # Next, check the error body
-        region = service_response.get('Error', {}).get('Region', None)
-        if region is not None:
-            return region
-
-        # Finally, HEAD the bucket. No other choice sadly.
-        try:
-            response = self._client.head_bucket(Bucket=bucket)
-            headers = response['ResponseMetadata']['HTTPHeaders']
-        except ClientError as e:
-            headers = e.response['ResponseMetadata']['HTTPHeaders']
-
-        region = headers.get('x-amz-bucket-region', None)
+        elif r := service_response.get('Error', {}).get('Region', None):
+            region = r
+        else:
+            # Finally, HEAD the bucket. No other choice sadly.
+            try:
+                response = self._client.head_bucket(Bucket=bucket)
+                headers = response['ResponseMetadata']['HTTPHeaders']
+            except ClientError as e:
+                headers = e.response['ResponseMetadata']['HTTPHeaders']
+            region = headers.get('x-amz-bucket-region', None)
+        validate_region_name(region)
         return region
 
     def set_request_url(self, old_url, new_endpoint, **kwargs):

tests/unit/test_utils.py

@@ -38,6 +38,7 @@
     InvalidExpressionError,
     InvalidIMDSEndpointError,
     InvalidIMDSEndpointModeError,
+    InvalidRegionError,
     MetadataRetrievalError,
     ReadTimeoutError,
     SSOTokenLoadError,
@@ -2113,6 +2114,47 @@ def test_no_redirect_from_error_for_mrap_accesspoint(self):
         )
         self.assertEqual(redirect_response, None)
 
+    def test_get_region_validates_region_from_header(self):
+        response = (
+            None,
+            {
+                'Error': {'Code': 'PermanentRedirect'},
+                'ResponseMetadata': {
+                    'HTTPHeaders': {'x-amz-bucket-region': 'invalid region!'}
+                },
+            },
+        )
+        with self.assertRaises(InvalidRegionError):
+            self.redirector.get_bucket_region('foo', response)
+
+    def test_get_region_validates_region_from_error_body(self):
+        response = (
+            None,
+            {
+                'Error': {
+                    'Code': 'PermanentRedirect',
+                    'Region': 'invalid region!',
+                },
+                'ResponseMetadata': {'HTTPHeaders': {}},
+            },
+        )
+        with self.assertRaises(InvalidRegionError):
+            self.redirector.get_bucket_region('foo', response)
+
+    def test_get_region_validates_region_from_head_bucket(self):
+        self.set_client_response_headers(
+            {'x-amz-bucket-region': 'invalid region!'}
+        )
+        response = (
+            None,
+            {
+                'Error': {'Code': 'PermanentRedirect'},
+                'ResponseMetadata': {'HTTPHeaders': {}},
+            },
+        )
+        with self.assertRaises(InvalidRegionError):
+            self.redirector.get_bucket_region('foo', response)
+
 
 class TestArnParser(unittest.TestCase):
     def setUp(self):