Optionally allow HTTP redirects

botocore/args.py

@@ -234,6 +234,7 @@ def compute_client_args(
                 client_cert=client_config.client_cert,
                 inject_host_prefix=client_config.inject_host_prefix,
                 tcp_keepalive=client_config.tcp_keepalive,
+                allow_http_redirects=client_config.allow_http_redirects,
             )
         self._compute_retry_config(config_kwargs)
         self._compute_connect_timeout(config_kwargs)

botocore/config.py

@@ -188,6 +188,18 @@ class Config:
     :param tcp_keepalive: Enables the TCP Keep-Alive socket option used when
         creating new connections if set to True.
 
+        Defaults to False.
+
+    :type allow_http_redirects: bool
+    :param allow_http_redirects: Whether to allow HTTP redirects using the
+        'location' header.
+
+        Setting this True will cause botocore to redirect requests to
+        the URL specified by the HTTP 'location' header, if any, on receiving
+        a HTTP 301, 302 or 307 from the upstream service.
+
+        HTTP redirections are limited to a maximum of three per request.
+
         Defaults to False.
     """
 
@@ -212,6 +224,7 @@ class Config:
             ('use_fips_endpoint', None),
             ('defaults_mode', None),
             ('tcp_keepalive', None),
+            ('allow_http_redirects', False),
         ]
     )
 

botocore/utils.py

@@ -1549,12 +1549,6 @@ def redirect_from_error(self, request_dict, response, operation, **kwargs):
             )
             return
 
-        if redirect_ctx.get('redirected'):
-            logger.debug(
-                'S3 request was previously redirected, not redirecting.'
-            )
-            return
-
         error = response[1].get('Error', {})
         error_code = error.get('Code')
         response_metadata = response[1].get('ResponseMetadata', {})
@@ -1590,6 +1584,41 @@ def redirect_from_error(self, request_dict, response, operation, **kwargs):
         ):
             return
 
+        # If the user has enabled the "allow_http_redirects" config option,
+        # set our URL to that provided by the "location" response header
+        # on receiving a HTTP 301, 302 or 307.
+        #
+        # This helps compatibility with some third-party storage
+        # systems.
+        #
+        # We dont treat the codes as semantically different, and we don't
+        # implement any sort of caching (ie, we don't respect the
+        # "Cache-Control" or "Expires" response headers).
+        #
+        # We also won't redirect more than three times.
+        location = response_metadata.get('HTTPHeaders', {}).get('location')
+        is_http_redirect = (
+            self._client._client_config.allow_http_redirects
+            and is_redirect_status
+            and location
+        )
+        if is_http_redirect:
+            redirect_count = redirect_ctx.get('redirect_count', 0)
+            if redirect_count >= 3:
+                logger.debug(
+                    'S3 request was previously redirected too many times, not redirecting.'
+                )
+                return
+            request_dict['url'] = location
+            redirect_ctx['redirect_count'] = redirect_count + 1
+            return 0
+
+        if redirect_ctx.get('redirected'):
+            logger.debug(
+                'S3 request was previously redirected, not redirecting.'
+            )
+            return
+
         bucket = request_dict['context']['s3_redirect']['bucket']
         client_region = request_dict['context'].get('client_region')
         new_region = self.get_bucket_region(bucket, response)
@@ -1697,6 +1726,7 @@ def annotate_request_context(self, params, context, **kwargs):
         bucket = params.get('Bucket')
         context['s3_redirect'] = {
             'redirected': False,
+            'redirect_count': 0,
             'bucket': bucket,
             'params': params,
         }

tests/functional/test_s3.py

@@ -1709,6 +1709,162 @@ def test_no_region_redirect_for_accesspoint(self):
                 self.fail("PermanentRedirect error should have been raised")
 
 
+class TestHTTPRedirect(BaseS3OperationTest):
+    def setUp(self):
+        super().setUp()
+
+        self.default_client = self.session.create_client(
+            "s3",
+            "us-west-2",
+            config=Config(
+                signature_version="s3v4",
+                s3={"addressing_style": "path"},
+            ),
+        )
+        self.default_http_stubber = ClientHTTPStubber(self.default_client)
+
+        self.client = self.session.create_client(
+            "s3",
+            "us-west-2",
+            config=Config(
+                allow_http_redirects="True",
+                signature_version="s3v4",
+                s3={"addressing_style": "path"},
+            ),
+        )
+        self.http_stubber = ClientHTTPStubber(self.client)
+
+        self.redirect_response = {
+            "status": 307,
+            "headers": {
+                "location": "https://ham.s3.eu-central-1.amazonaws.com:1234/foo?bar=spam"
+            },
+            "body": (
+                b'<?xml version="1.0" encoding="UTF-8"?>\n'
+                b"<Error>"
+                b"    <Code>TemporaryRedirect</Code>"
+                b"    <Message>An arbitrary temporary redirect message."
+                b"    </Message>"
+                b"    <Bucket>foo</Bucket>"
+                b"    <Endpoint>foo.s3.eu-central-1.amazonaws.com</Endpoint>"
+                b"</Error>"
+            ),
+        }
+        self.success_response = {
+            "status": 200,
+            "headers": {},
+            "body": (
+                b'<?xml version="1.0" encoding="UTF-8"?>\n'
+                b"<ListBucketResult>"
+                b"    <Name>foo</Name>"
+                b"    <Prefix></Prefix>"
+                b"    <Marker></Marker>"
+                b"    <MaxKeys>1000</MaxKeys>"
+                b"    <EncodingType>url</EncodingType>"
+                b"    <IsTruncated>false</IsTruncated>"
+                b"</ListBucketResult>"
+            ),
+        }
+
+    def test_http_redirects_disabled_by_default(self):
+        self.default_http_stubber.add_response(**self.redirect_response)
+        self.default_http_stubber.add_response(**self.success_response)
+        with self.default_http_stubber:
+            try:
+                self.default_client.list_objects(Bucket="foo")
+            except self.client.exceptions.ClientError as e:
+                self.assertEqual(
+                    e.response["Error"]["Code"], "TemporaryRedirect"
+                )
+            else:
+                self.fail("TemporaryRedirect error should have been raised")
+
+    def test_http_redirect(self):
+        self.http_stubber.add_response(**self.redirect_response)
+        self.http_stubber.add_response(**self.success_response)
+        with self.http_stubber:
+            response = self.client.list_objects(Bucket="foo")
+        self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)
+        self.assertEqual(len(self.http_stubber.requests), 2)
+
+        initial_url = (
+            "https://s3.us-west-2.amazonaws.com/foo" "?encoding-type=url"
+        )
+        self.assertEqual(self.http_stubber.requests[0].url, initial_url)
+
+        fixed_url = (
+            "https://ham.s3.eu-central-1.amazonaws.com:1234/foo" "?bar=spam"
+        )
+        self.assertEqual(self.http_stubber.requests[1].url, fixed_url)
+
+    def test_http_redirect_bypasses_cache(self):
+        self.http_stubber.add_response(**self.redirect_response)
+        self.http_stubber.add_response(**self.success_response)
+        self.http_stubber.add_response(**self.redirect_response)
+        self.http_stubber.add_response(**self.success_response)
+
+        with self.http_stubber:
+            first_response = self.client.list_objects(Bucket="foo")
+            second_response = self.client.list_objects(Bucket="foo")
+
+        self.assertEqual(
+            first_response["ResponseMetadata"]["HTTPStatusCode"], 200
+        )
+        self.assertEqual(
+            second_response["ResponseMetadata"]["HTTPStatusCode"], 200
+        )
+
+        self.assertEqual(len(self.http_stubber.requests), 4)
+        initial_url = (
+            "https://s3.us-west-2.amazonaws.com/foo" "?encoding-type=url"
+        )
+        fixed_url = (
+            "https://ham.s3.eu-central-1.amazonaws.com:1234/foo" "?bar=spam"
+        )
+
+        self.assertEqual(self.http_stubber.requests[0].url, initial_url)
+        self.assertEqual(self.http_stubber.requests[1].url, fixed_url)
+        self.assertEqual(self.http_stubber.requests[2].url, initial_url)
+        self.assertEqual(self.http_stubber.requests[3].url, fixed_url)
+
+    def test_http_multiple_redirections(self):
+        self.http_stubber.add_response(**self.redirect_response)
+        self.http_stubber.add_response(**self.redirect_response)
+        self.http_stubber.add_response(**self.success_response)
+
+        with self.http_stubber:
+            response = self.client.list_objects(Bucket="foo")
+
+        self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)
+
+        self.assertEqual(len(self.http_stubber.requests), 3)
+        initial_url = (
+            "https://s3.us-west-2.amazonaws.com/foo" "?encoding-type=url"
+        )
+        fixed_url = (
+            "https://ham.s3.eu-central-1.amazonaws.com:1234/foo" "?bar=spam"
+        )
+
+        self.assertEqual(self.http_stubber.requests[0].url, initial_url)
+        self.assertEqual(self.http_stubber.requests[1].url, fixed_url)
+        self.assertEqual(self.http_stubber.requests[2].url, fixed_url)
+
+    def test_http_redirection_limit(self):
+        for _ in range(4):
+            self.http_stubber.add_response(**self.redirect_response)
+        self.http_stubber.add_response(**self.success_response)
+
+        with self.http_stubber:
+            try:
+                self.client.list_objects(Bucket="foo")
+            except self.client.exceptions.ClientError as e:
+                self.assertEqual(
+                    e.response["Error"]["Code"], "TemporaryRedirect"
+                )
+            else:
+                self.fail("TemporaryRedirect error should have been raised")
+
+
 class TestFipsRegionRedirect(BaseS3OperationTest):
     def setUp(self):
         super().setUp()

tests/unit/test_utils.py

@@ -1714,6 +1714,89 @@ def test_redirects_301(self):
         )
         self.assertIsNone(redirect_response)
 
+    def test_allow_http_redirects(self):
+        self.client._client_config.allow_http_redirects = True
+
+        request_dict = {
+            'url': 'https://us-west-2.amazonaws.com/foo',
+            'context': {
+                's3_redirect': {
+                    'bucket': 'foo',
+                    'redirected': False,
+                    'params': {'Bucket': 'foo'},
+                },
+                'signing': {},
+            },
+        }
+
+        raw_response = mock.Mock()
+        raw_response.status_code = 307
+        raw_response.content = 'Temporary Redirect'
+        raw_response.headers = {'location': 'https://foo.spam:1234/ham?eggs'}
+        response = (
+            raw_response,
+            {
+                'Error': {'Code': '307', 'Message': 'Temporary Redirect'},
+                'ResponseMetadata': {
+                    'HTTPHeaders': {
+                        'location': 'https://foo.spam:1234/ham?eggs',
+                    }
+                },
+            },
+        )
+
+        self.operation.name = 'HeadObject'
+        redirect_response = self.redirector.redirect_from_error(
+            request_dict, response, self.operation
+        )
+        self.assertEqual(redirect_response, 0)
+        self.assertEqual(request_dict['url'], 'https://foo.spam:1234/ham?eggs')
+
+        self.operation.name = 'ListObjects'
+        redirect_response = self.redirector.redirect_from_error(
+            request_dict, response, self.operation
+        )
+        self.assertEqual(redirect_response, 0)
+        self.assertEqual(request_dict['url'], 'https://foo.spam:1234/ham?eggs')
+
+    def test_allow_http_redirects_limit(self):
+        self.client._client_config.allow_http_redirects = True
+
+        request_dict = {
+            'context': {
+                'signing': {'bucket': 'foo', 'region': 'us-west-2'},
+                's3_redirected': False,
+                's3_redirect': {
+                    'bucket': 'foo',
+                    'redirected': False,
+                    'params': {'Bucket': 'foo'},
+                    'redirect_count': 4,
+                },
+            },
+            'url': 'https://us-west-2.amazonaws.com/foo',
+        }
+
+        raw_response = mock.Mock()
+        raw_response.status_code = 307
+        raw_response.content = 'Temporary Redirect'
+        raw_response.headers = {'location': 'https://foo.spam:1234/ham?eggs'}
+        response = (
+            raw_response,
+            {
+                'Error': {'Code': '307', 'Message': 'Temporary Redirect'},
+                'ResponseMetadata': {
+                    'HTTPHeaders': {
+                        'location': 'https://foo.spam:1234/ham?eggs',
+                    }
+                },
+            },
+        )
+
+        redirect_response = self.redirector.redirect_from_error(
+            request_dict, response, self.operation
+        )
+        self.assertIsNone(redirect_response)
+
     def test_redirects_400_head_bucket(self):
         request_dict = {
             'url': 'https://us-west-2.amazonaws.com/foo',