From dc02db5d734037d3bf361d06fbf00f6a402d4e8b Mon Sep 17 00:00:00 2001
From: Alessandra Romero <24320222+alexgromero@users.noreply.github.com>
Date: Wed, 23 Oct 2024 13:36:47 -0400
Subject: [PATCH] Fix S3 200 Error handling (#3288)

---
 .changes/next-release/bugfix-s3-1405.json |  5 +++++
 botocore/handlers.py                      | 22 +++++++++++++++++-----
 2 files changed, 22 insertions(+), 5 deletions(-)
 create mode 100644 .changes/next-release/bugfix-s3-1405.json

diff --git a/.changes/next-release/bugfix-s3-1405.json b/.changes/next-release/bugfix-s3-1405.json
new file mode 100644
index 0000000000..1cc3d8575a
--- /dev/null
+++ b/.changes/next-release/bugfix-s3-1405.json
@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "``s3``",
+  "description": "Fix S3 200 Error handling to address bugs introduced in payload operations."
+}
diff --git a/botocore/handlers.py b/botocore/handlers.py
index 6c68473d4e..d44cb07469 100644
--- a/botocore/handlers.py
+++ b/botocore/handlers.py
@@ -1245,11 +1245,7 @@ def document_expires_shape(section, event_name, **kwargs):
 def _handle_200_error(operation_model, response_dict, **kwargs):
     # S3 can return a 200 response with an error embedded in the body.
     # Convert the 200 to a 500 for retry resolution in ``_update_status_code``.
-    if (
-        not response_dict
-        or operation_model.has_streaming_output
-        or operation_model.has_event_stream_output
-    ):
+    if not _should_handle_200_error(operation_model, response_dict):
         # Operations with streaming response blobs are excluded as they
         # can't be reliably distinguished from an S3 error.
         return
@@ -1262,6 +1258,22 @@ def _handle_200_error(operation_model, response_dict, **kwargs):
         )
 
 
+def _should_handle_200_error(operation_model, response_dict):
+    output_shape = operation_model.output_shape
+    if (
+        not response_dict
+        or operation_model.has_event_stream_output
+        or not output_shape
+    ):
+        return False
+    payload = output_shape.serialization.get('payload')
+    if payload is not None:
+        payload_shape = output_shape.members[payload]
+        if payload_shape.type_name in ('blob', 'string'):
+            return False
+    return True
+
+
 def _update_status_code(response, **kwargs):
     # Update the http_response status code when the parsed response has been
     # modified in a handler. This enables retries for cases like ``_handle_200_error``.