cloud_storage_clients: check Content-Type header before parsing REST error responses in s3_client

src/v/cloud_io/tests/s3_imposter.cc

@@ -15,6 +15,7 @@
 #include "bytes/iobuf_parser.h"
 #include "cloud_storage_clients/client.h"
 #include "cloud_storage_clients/client_probe.h"
+#include "http/tests/utils.h"
 #include "test_utils/async.h"
 #include "test_utils/test_macros.h"
 
@@ -501,11 +502,11 @@ void s3_imposter_fixture::set_routes(
     using reply = ss::http::reply;
     _content_handler = ss::make_shared<content_handler>(
       expectations, *this, std::move(headers_to_store));
-    _handler = std::make_unique<function_handler>(
-      [this](const_req req, reply& repl) {
+    _handler = std::make_unique<http::test_utils::flexible_function_handler>(
+      [this](const_req req, reply& repl, [[maybe_unused]] ss::sstring& type) {
           return _content_handler->handle(req, repl);
       },
-      "txt");
+      "xml");
     r.add_default_handler(_handler.get());
 }
 

src/v/cloud_storage/tests/anomalies_detector_test.cc

@@ -424,7 +424,10 @@ class bucket_view_fixture : public http_imposter_fixture {
           .request(full_path)
           .with_method(ss::httpd::operation_type::GET)
           .then_reply_with(
-            not_found_response, ss::http::reply::status_type::not_found);
+            not_found_response,
+            std::vector<std::pair<ss::sstring, ss::sstring>>{
+              {"Content-Type", "application/xml"}},
+            ss::http::reply::status_type::not_found);
     }
 
     void parse_manifests(

src/v/cloud_storage_clients/abs_client.cc

@@ -50,7 +50,6 @@ constexpr boost::beast::string_view delete_snapshot_name
 constexpr boost::beast::string_view is_hns_enabled_name = "x-ms-is-hns-enabled";
 constexpr boost::beast::string_view delete_snapshot_value = "include";
 constexpr boost::beast::string_view error_code_name = "x-ms-error-code";
-constexpr boost::beast::string_view content_type_name = "Content-Type";
 constexpr boost::beast::string_view expiry_option_name = "x-ms-expiry-option";
 constexpr boost::beast::string_view expiry_option_value = "RelativeToNow";
 constexpr boost::beast::string_view expiry_time_name = "x-ms-expiry-time";
@@ -74,23 +73,6 @@ bool is_error_retryable(
 
 namespace cloud_storage_clients {
 
-enum class response_content_type : int8_t { unknown, xml, json };
-
-static response_content_type
-get_response_content_type(const http::client::response_header& headers) {
-    if (auto iter = headers.find(content_type_name); iter != headers.end()) {
-        if (iter->value().find("json") != std::string_view::npos) {
-            return response_content_type::json;
-        }
-
-        if (iter->value().find("xml") != std::string_view::npos) {
-            return response_content_type::xml;
-        }
-    }
-
-    return response_content_type::unknown;
-}
-
 static abs_rest_error_response
 parse_xml_rest_error_response(boost::beast::http::status result, iobuf buf) {
     using namespace cloud_storage_clients;
@@ -628,7 +610,7 @@ ss::future<http::client::response_stream_ref> abs_client::do_get_object(
               response_stream->get_headers());
         }
 
-        const auto content_type = get_response_content_type(
+        const auto content_type = util::get_response_content_type(
           response_stream->get_headers());
         auto buf = co_await util::drain_response_stream(
           std::move(response_stream));
@@ -686,7 +668,7 @@ ss::future<> abs_client::do_put_object(
     if (const auto is_no_content_and_accepted = accept_no_content
                                                 && status == no_content;
         status != created && !is_no_content_and_accepted) {
-        const auto content_type = get_response_content_type(
+        const auto content_type = util::get_response_content_type(
           response_stream->get_headers());
         auto buf = co_await util::drain_response_stream(
           std::move(response_stream));
@@ -803,7 +785,7 @@ ss::future<> abs_client::do_delete_object(
 
     const auto status = response_stream->get_headers().result();
     if (status != boost::beast::http::status::accepted) {
-        const auto content_type = get_response_content_type(
+        const auto content_type = util::get_response_content_type(
           response_stream->get_headers());
         auto buf = co_await util::drain_response_stream(
           std::move(response_stream));
@@ -884,7 +866,7 @@ ss::future<abs_client::list_bucket_result> abs_client::do_list_objects(
     const auto status = response_stream->get_headers().result();
 
     if (status != boost::beast::http::status::ok) {
-        const auto content_type = get_response_content_type(
+        const auto content_type = util::get_response_content_type(
           response_stream->get_headers());
         iobuf buf = co_await util::drain_response_stream(response_stream);
         throw parse_rest_error_response(content_type, status, std::move(buf));
@@ -1047,7 +1029,7 @@ ss::future<> abs_client::do_delete_file(
     if (
       status != boost::beast::http::status::accepted
       && status != boost::beast::http::status::ok) {
-        const auto content_type = get_response_content_type(
+        const auto content_type = util::get_response_content_type(
           response_stream->get_headers());
         auto buf = co_await util::drain_response_stream(
           std::move(response_stream));

src/v/cloud_storage_clients/s3_client.cc

@@ -412,36 +412,47 @@ status_to_error_code(boost::beast::http::status s) {
     return cloud_storage_clients::s3_error_code::_unknown;
 }
 
-template<class ResultT = void>
-ss::future<ResultT>
-parse_rest_error_response(boost::beast::http::status result, iobuf&& buf) {
-    if (buf.empty()) {
-        // AWS errors occasionally come with an empty body
-        // (See https://github.com/redpanda-data/redpanda/issues/6061)
-        // Without a proper code, we treat it as a hint to gracefully retry
-        // (synthesize the slow_down code).
-        rest_error_response err(
-          fmt::format("{}", status_to_error_code(result)),
-          fmt::format("Empty error response, status code {}", result),
-          "",
-          "");
-        return ss::make_exception_future<ResultT>(err);
-    } else {
-        try {
-            auto resp = util::iobuf_to_ptree(std::move(buf), s3_log);
-            constexpr const char* empty = "";
-            auto code = resp.get<ss::sstring>("Error.Code", empty);
-            auto msg = resp.get<ss::sstring>("Error.Message", empty);
-            auto rid = resp.get<ss::sstring>("Error.RequestId", empty);
-            auto res = resp.get<ss::sstring>("Error.Resource", empty);
-            rest_error_response err(code, msg, rid, res);
-            return ss::make_exception_future<ResultT>(err);
-        } catch (...) {
-            vlog(
-              s3_log.error, "!!error parse error {}", std::current_exception());
-            throw;
+rest_error_response parse_xml_rest_error_response(iobuf&& buf) {
+    try {
+        auto resp = util::iobuf_to_ptree(std::move(buf), s3_log);
+        constexpr const char* empty = "";
+        auto code = resp.get<ss::sstring>("Error.Code", empty);
+        auto msg = resp.get<ss::sstring>("Error.Message", empty);
+        auto rid = resp.get<ss::sstring>("Error.RequestId", empty);
+        auto res = resp.get<ss::sstring>("Error.Resource", empty);
+        rest_error_response err(code, msg, rid, res);
+        return err;
+    } catch (...) {
+        vlog(s3_log.error, "!!error parse error {}", std::current_exception());
+        throw;
+    }
+}
+
+template<typename ResultT = void>
+ss::future<ResultT> parse_rest_error_response(
+  response_content_type type, boost::beast::http::status result, iobuf&& buf) {
+    // AWS errors occasionally come with an empty body
+    // (See https://github.com/redpanda-data/redpanda/issues/6061)
+    // Without a proper code, we treat it as a hint to gracefully retry
+    // (synthesize the slow_down code).
+    if (!buf.empty()) {
+        if (type == response_content_type::xml) {
+            // Error responses from S3 _should_ have the Content-Type header set
+            // with `application/xml`- however, certain responses (such as `503
+            // Service Unavailable`) may not be of this form.
+            // https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html
+            return ss::make_exception_future<ResultT>(
+              parse_xml_rest_error_response(std::move(buf)));
         }
     }
+
+    auto result_prefix = buf.empty() ? "Empty error response, " : "";
+    rest_error_response err(
+      fmt::format("{}", status_to_error_code(result)),
+      fmt::format("{}status code {}", result_prefix, result),
+      "",
+      "");
+    return ss::make_exception_future<ResultT>(err);
 }
 
 /// Head response doesn't give us an XML encoded error object in
@@ -708,11 +719,13 @@ ss::future<http::client::response_stream_ref> s3_client::do_get_object(
                         ref->get_headers().result(),
                         ref->get_headers());
                   }
+                  const auto content_type = util::get_response_content_type(
+                    ref->get_headers());
                   return util::drain_response_stream(std::move(ref))
-                    .then([result](iobuf&& res) {
+                    .then([content_type, result](iobuf&& res) {
                         return parse_rest_error_response<
                           http::client::response_stream_ref>(
-                          result, std::move(res));
+                          content_type, result, std::move(res));
                     });
               }
               return ss::make_ready_future<http::client::response_stream_ref>(
@@ -839,8 +852,11 @@ ss::future<> s3_client::do_put_object(
                             id,
                             status,
                             ref->get_headers());
+                          const auto content_type
+                            = util::get_response_content_type(
+                              ref->get_headers());
                           return parse_rest_error_response<>(
-                            status, std::move(res));
+                            content_type, status, std::move(res));
                       }
                       return ss::now();
                   });
@@ -924,12 +940,14 @@ ss::future<s3_client::list_bucket_result> s3_client::do_list_objects_v2(
                       header.result(),
                       header);
 
+                    const auto content_type = util::get_response_content_type(
+                      header);
                     // In the error path we drain the response stream fully, the
                     // error response should not be very large.
                     return util::drain_chunked_response_stream(resp).then(
-                      [result = header.result()](iobuf buf) {
+                      [result = header.result(), content_type](iobuf buf) {
                           return parse_rest_error_response<list_bucket_result>(
-                            result, std::move(buf));
+                            content_type, result, std::move(buf));
                       });
                 }
 
@@ -1016,7 +1034,10 @@ ss::future<> s3_client::do_delete_object(
                     key,
                     status,
                     ref->get_headers());
-                  return parse_rest_error_response<>(status, std::move(res));
+                  const auto content_type = util::get_response_content_type(
+                    ref->get_headers());
+                  return parse_rest_error_response<>(
+                    content_type, status, std::move(res));
               }
               return ss::now();
           });
@@ -1106,8 +1127,10 @@ auto s3_client::do_delete_objects(
             [response](iobuf&& res) {
                 auto status = response->get_headers().result();
                 if (status != boost::beast::http::status::ok) {
+                    const auto content_type = util::get_response_content_type(
+                      response->get_headers());
                     return parse_rest_error_response<delete_objects_result>(
-                      status, std::move(res));
+                      content_type, status, std::move(res));
                 }
                 auto parse_result = iobuf_to_delete_objects_result(
                   std::move(res));

src/v/cloud_storage_clients/tests/BUILD

@@ -43,6 +43,7 @@ redpanda_cc_btest(
         "//src/v/bytes:iostream",
         "//src/v/cloud_storage_clients",
         "//src/v/hashing:secure",
+        "//src/v/http/tests:utils",
         "//src/v/net",
         "//src/v/test_utils:seastar_boost",
         "//src/v/utils:base64",

src/v/cloud_storage_clients/tests/CMakeLists.txt

@@ -10,7 +10,7 @@ rp_test(
     exception_test.cc
     util_test.cc
   DEFINITIONS BOOST_TEST_DYN_LINK
-  LIBRARIES v::seastar_testing_main Boost::unit_test_framework v::http v::cloud_storage_clients v::cloud_roles
+  LIBRARIES v::seastar_testing_main Boost::unit_test_framework v::http v::cloud_storage_clients v::cloud_roles v::http_test_utils
   ARGS "-- -c 1"
   LABELS s3
 )