diff --git a/.changes/next-release/bugfix-s3-42137.json b/.changes/next-release/bugfix-s3-42137.json new file mode 100644 index 0000000000..916d9c35f9 --- /dev/null +++ b/.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``." +} diff --git a/botocore/utils.py b/botocore/utils.py index 9553e0c82a..997c2713d2 100644 --- a/botocore/utils.py +++ b/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): diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 955e04c2d9..fcd9deb6fc 100644 --- a/tests/unit/test_utils.py +++ b/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()