S3 200 errors implementation

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

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "``s3``",
+  "description": "Adds logic to gracefully handle HTTP 200 OK responses that contain error information in the response body."
+}

botocore/handlers.py

@@ -132,6 +132,7 @@ def escape_xml_payload(params, **kwargs):
 
 
 def check_for_200_error(response, **kwargs):
+    """This function has been deprecated, but is kept for backwards compatibility."""
     # From: http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectCOPY.html
     # There are two opportunities for a copy request to return an error. One
     # can occur when Amazon S3 receives the copy request and the other can
@@ -162,13 +163,13 @@ def check_for_200_error(response, **kwargs):
         http_response.status_code = 500
 
 
-def _looks_like_special_case_error(http_response):
-    if http_response.status_code == 200:
+def _looks_like_special_case_error(status_code, body):
+    if status_code == 200 and body:
         try:
             parser = ETree.XMLParser(
                 target=ETree.TreeBuilder(), encoding='utf-8'
             )
-            parser.feed(http_response.content)
+            parser.feed(body)
             root = parser.close()
         except XMLParseError:
             # In cases of network disruptions, we may end up with a partial
@@ -1239,6 +1240,44 @@ def document_expires_shape(section, event_name, **kwargs):
         )
 
 
+def _handle_200_error(operation_model, response_dict, **kwargs):
+    # S3 can return a 200 OK response with an error embedded in the body.
+    # Conceptually, this should be handled like a 500 response in terms of
+    # raising exceptions and retries which we handle in _retry_200_error.
+    # This handler converts the 200 response to a 500 response to ensure
+    # correct error handling.
+    if not response_dict or operation_model.has_streaming_output:
+        # Operations with streaming response blobs should be excluded as they
+        # may contain customer content which mimics the form of an S3 error.
+        return
+    if _looks_like_special_case_error(
+        response_dict['status_code'], response_dict['body']
+    ):
+        response_dict['status_code'] = 500
+        logger.debug(
+            f"Error found for response with 200 status code: {response_dict['body']}. "
+            f"Changing the http_response status code to 500 will be propagated in "
+            f"the _retry_200_error handler."
+        )
+
+
+def _retry_200_error(response, **kwargs):
+    # Adjusts the HTTP status code for responses that may contain errors
+    # embedded in a 200 OK response body. The _handle_200_error function
+    # modifies the parsed response status code to 500 if it detects an error.
+    # This function checks if the HTTP status code differs from the parsed
+    # status code and updates the HTTP response accordingly, ensuring
+    # correct handling for retries.
+    if response is None:
+        return
+    http_response, parsed = response
+    parsed_status_code = parsed.get('ResponseMetadata', {}).get(
+        'HTTPStatusCode'
+    )
+    if http_response.status_code != parsed_status_code:
+        http_response.status_code = parsed_status_code
+
+
 # This is a list of (event_name, handler).
 # When a Session is created, everything in this list will be
 # automatically registered with that Session.
@@ -1269,6 +1308,7 @@ def document_expires_shape(section, event_name, **kwargs):
     ('after-call.cloudformation.GetTemplate', json_decode_template_body),
     ('after-call.s3.GetBucketLocation', parse_get_bucket_location),
     ('before-parse.s3.*', handle_expires_header),
+    ('before-parse.s3.*', _handle_200_error, REGISTER_FIRST),
     ('before-parameter-build', generate_idempotent_uuid),
     ('before-parameter-build.s3', validate_bucket_name),
     ('before-parameter-build.s3', remove_bucket_from_url_paths_from_model),
@@ -1312,13 +1352,7 @@ def document_expires_shape(section, event_name, **kwargs):
     ('before-call.ec2.CopySnapshot', inject_presigned_url_ec2),
     ('request-created', add_retry_headers),
     ('request-created.machinelearning.Predict', switch_host_machinelearning),
-    ('needs-retry.s3.UploadPartCopy', check_for_200_error, REGISTER_FIRST),
-    ('needs-retry.s3.CopyObject', check_for_200_error, REGISTER_FIRST),
-    (
-        'needs-retry.s3.CompleteMultipartUpload',
-        check_for_200_error,
-        REGISTER_FIRST,
-    ),
+    ('needs-retry.s3.*', _retry_200_error, REGISTER_FIRST),
     ('choose-signer.cognito-identity.GetId', disable_signing),
     ('choose-signer.cognito-identity.GetOpenIdToken', disable_signing),
     ('choose-signer.cognito-identity.UnlinkIdentity', disable_signing),

tests/functional/test_s3.py

@@ -28,7 +28,6 @@
     UnsupportedS3AccesspointConfigurationError,
     UnsupportedS3ConfigurationError,
 )
-from botocore.parsers import ResponseParserError
 from tests import (
     BaseSessionTest,
     ClientHTTPStubber,
@@ -435,21 +434,20 @@ def create_stubbed_s3_client(self, **kwargs):
         http_stubber.start()
         return client, http_stubber
 
-    def test_s3_copy_object_with_empty_response(self):
+    def test_s3_copy_object_with_incomplete_response(self):
         self.client, self.http_stubber = self.create_stubbed_s3_client(
             region_name="us-east-1"
         )
 
-        empty_body = b""
+        incomplete_body = b'<?xml version="1.0" encoding="UTF-8"?>\n\n\n'
         complete_body = (
             b'<?xml version="1.0" encoding="UTF-8"?>\n\n'
             b"<CopyObjectResult "
             b'xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
             b"<LastModified>2020-04-21T21:03:31.000Z</LastModified>"
             b"<ETag>&quot;s0mEcH3cK5uM&quot;</ETag></CopyObjectResult>"
         )
-
-        self.http_stubber.add_response(status=200, body=empty_body)
+        self.http_stubber.add_response(status=200, body=incomplete_body)
         self.http_stubber.add_response(status=200, body=complete_body)
         response = self.client.copy_object(
             Bucket="bucket",
@@ -462,19 +460,33 @@ def test_s3_copy_object_with_empty_response(self):
         self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)
         self.assertTrue("CopyObjectResult" in response)
 
-    def test_s3_copy_object_with_incomplete_response(self):
+    def test_s3_copy_object_with_200_error_response(self):
         self.client, self.http_stubber = self.create_stubbed_s3_client(
             region_name="us-east-1"
         )
-
-        incomplete_body = b'<?xml version="1.0" encoding="UTF-8"?>\n\n\n'
-        self.http_stubber.add_response(status=200, body=incomplete_body)
-        with self.assertRaises(ResponseParserError):
+        error_body = (
+            b"<Error>"
+            b"<Code>SlowDown</Code>"
+            b"<Message>Please reduce your request rate.</Message>"
+            b"</Error>"
+        )
+        # Populate 5 retries for SlowDown
+        for i in range(5):
+            self.http_stubber.add_response(status=200, body=error_body)
+        with self.assertRaises(botocore.exceptions.ClientError) as context:
             self.client.copy_object(
                 Bucket="bucket",
                 CopySource="other-bucket/test.txt",
                 Key="test.txt",
             )
+        self.assertEqual(len(self.http_stubber.requests), 5)
+        self.assertEqual(
+            context.exception.response["ResponseMetadata"]["HTTPStatusCode"],
+            500,
+        )
+        self.assertEqual(
+            context.exception.response["Error"]["Code"], "SlowDown"
+        )
 
 
 class TestAccesspointArn(BaseS3ClientConfigurationTest):

tests/unit/test_handlers.py

@@ -501,33 +501,6 @@ def test_presigned_url_casing_changed_for_rds(self):
         )
         self.assertIn('X-Amz-Signature', params['PreSignedUrl'])
 
-    def test_500_status_code_set_for_200_response(self):
-        http_response = mock.Mock()
-        http_response.status_code = 200
-        http_response.content = """
-            <Error>
-              <Code>AccessDenied</Code>
-              <Message>Access Denied</Message>
-              <RequestId>id</RequestId>
-              <HostId>hostid</HostId>
-            </Error>
-        """
-        handlers.check_for_200_error((http_response, {}))
-        self.assertEqual(http_response.status_code, 500)
-
-    def test_200_response_with_no_error_left_untouched(self):
-        http_response = mock.Mock()
-        http_response.status_code = 200
-        http_response.content = "<NotAnError></NotAnError>"
-        handlers.check_for_200_error((http_response, {}))
-        # We don't touch the status code since there are no errors present.
-        self.assertEqual(http_response.status_code, 200)
-
-    def test_500_response_can_be_none(self):
-        # A 500 response can raise an exception, which means the response
-        # object is None.  We need to handle this case.
-        handlers.check_for_200_error(None)
-
     def test_route53_resource_id(self):
         event = 'before-parameter-build.route53.GetHostedZone'
         params = {
@@ -1202,14 +1175,14 @@ def test_s3_special_case_is_before_other_retry(self):
             caught_exception=None,
         )
         # This is implementation specific, but we're trying to verify that
-        # the check_for_200_error is before any of the retry logic in
+        # the _retry_200_error is before any of the retry logic in
         # botocore.retryhandlers.
         # Technically, as long as the relative order is preserved, we don't
         # care about the absolute order.
         names = self.get_handler_names(responses)
-        self.assertIn('check_for_200_error', names)
+        self.assertIn('_retry_200_error', names)
         self.assertIn('RetryHandler', names)
-        s3_200_handler = names.index('check_for_200_error')
+        s3_200_handler = names.index('_retry_200_error')
         general_retry_handler = names.index('RetryHandler')
         self.assertTrue(
             s3_200_handler < general_retry_handler,
@@ -1944,3 +1917,80 @@ def test_document_response_params_without_expires(document_expires_mocks):
     mocks['section'].get_section.assert_not_called()
     mocks['param_section'].add_new_section.assert_not_called()
     mocks['doc_section'].write.assert_not_called()
+
+
+@pytest.fixture()
+def operation_model_for_200_error():
+    operation_model = mock.Mock()
+    operation_model.has_streaming_output = False
+    return operation_model
+
+
+@pytest.fixture()
+def response_dict_for_200_error():
+    return {
+        'status_code': 200,
+        'body': (
+            b"<Error>"
+            b"<Code>SlowDown</Code>"
+            b"<Message>Please reduce your request rate.</Message>"
+            b"</Error>"
+        ),
+    }
+
+
+def test_500_status_code_set_for_200_response(
+    operation_model_for_200_error, response_dict_for_200_error
+):
+    handlers._handle_200_error(
+        operation_model_for_200_error, response_dict_for_200_error
+    )
+    assert response_dict_for_200_error['status_code'] == 500
+
+
+def test_200_response_with_no_error_left_untouched(
+    operation_model_for_200_error, response_dict_for_200_error
+):
+    response_dict_for_200_error['body'] = b"<NotAnError/>"
+    handlers._handle_200_error(
+        operation_model_for_200_error, response_dict_for_200_error
+    )
+    # We don't touch the status code since there are no errors present.
+    assert response_dict_for_200_error['status_code'] == 200
+
+
+def test_200_response_with_streaming_output_left_untouched(
+    operation_model_for_200_error,
+    response_dict_for_200_error,
+):
+    operation_model_for_200_error.has_streaming_output = True
+    handlers._handle_200_error(
+        operation_model_for_200_error, response_dict_for_200_error
+    )
+    # We don't touch the status code on streaming operations.
+    assert response_dict_for_200_error['status_code'] == 200
+
+
+def test_200_response_with_no_body_left_untouched(
+    operation_model_for_200_error, response_dict_for_200_error
+):
+    response_dict_for_200_error['body'] = b""
+    handlers._handle_200_error(
+        operation_model_for_200_error, response_dict_for_200_error
+    )
+    assert response_dict_for_200_error['status_code'] == 200
+
+
+def test_http_status_code_updated_to_retry_200_response():
+    http_response = mock.Mock()
+    http_response.status_code = 200
+    parsed = {}
+    parsed.setdefault('ResponseMetadata', {})['HTTPStatusCode'] = 500
+    handlers._retry_200_error((http_response, parsed))
+    assert http_response.status_code == 500
+
+
+def test_500_response_can_be_none():
+    # A 500 response can raise an exception, which means the response
+    # object is None.  We need to handle this case.
+    handlers._retry_200_error(None)