validate region from S3 redirect response in S3RegionRedirector

.changes/next-release/bugfix-s3-42137.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "s3",
+  "description": "Validate the bucket region sourced from S3 redirect responses in the deprecated ``S3RegionRedirector``, matching ``S3RegionRedirectorv2``."
+}

botocore/utils.py

@@ -2077,21 +2077,20 @@ 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']
+        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)
+            region = headers.get('x-amz-bucket-region', None)
+        validate_region_name(region)
         return region
 
     def set_request_url(self, params, context, **kwargs):

tests/unit/test_utils.py

@@ -17,6 +17,7 @@
 import os
 import shutil
 import tempfile
+import warnings
 from contextlib import contextmanager
 from sys import getrefcount
 
@@ -66,6 +67,7 @@
     JSONFileCache,
     S3ArnParamHandler,
     S3EndpointSetter,
+    S3RegionRedirector,
     S3RegionRedirectorv2,
     SSOTokenLoader,
     _get_bearer_env_var_name,
@@ -2208,6 +2210,61 @@ def test_get_region_validates_region_from_head_bucket(self):
             self.redirector.get_bucket_region('foo', response)
 
 
+class TestDeprecatedS3RegionRedirector(unittest.TestCase):
+    def setUp(self):
+        self.client = mock.Mock()
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore', FutureWarning)
+            self.redirector = S3RegionRedirector(mock.Mock(), self.client)
+
+    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.client.head_bucket.side_effect = ClientError(
+            {
+                'Error': {'Code': '', 'Message': ''},
+                'ResponseMetadata': {
+                    'HTTPHeaders': {'x-amz-bucket-region': 'invalid region!'}
+                },
+            },
+            'HeadBucket',
+        )
+        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):
         self.parser = ArnParser()