Merge pull request #30196 from vbotbuildovich/backport-pr-30191-v26.1.x-979

Author: rockwotj

src/v/kafka/client/transport.cc

@@ -107,7 +107,8 @@ ss::future<> transport::read_loop() {
         _pending_requests.erase(it);
         std::optional<tagged_fields> reply_tags;
         if (entry->is_flexible) {
-            auto [tags, bytes_read] = co_await parse_tags(in());
+            auto [tags, bytes_read] = co_await parse_tags(
+              in(), bytes_remaining);
             reply_tags = std::move(tags);
             bytes_remaining -= bytes_read;
         }

src/v/kafka/protocol/BUILD

@@ -73,6 +73,7 @@ redpanda_cc_library(
     deps = [
         ":messages",
         ":protocol",
+        "//src/v/base",
         "//src/v/net:types",
         "//src/v/utils:vint",
         "@seastar",

src/v/kafka/protocol/flex_versions.cc

@@ -9,6 +9,7 @@
 
 #include "kafka/protocol/flex_versions.h"
 
+#include "base/units.h"
 #include "kafka/protocol/messages.h"
 #include "kafka/protocol/types.h"
 #include "kafka/protocol/wire.h"
@@ -17,8 +18,6 @@
 
 #include <seastar/core/iostream.hh>
 
-#include <stdexcept>
-
 namespace kafka {
 
 namespace {
@@ -49,6 +48,11 @@ get_flexible_request_min_versions_list(type_list<RequestTypes...> r) {
 constexpr auto g_flex_mapping = get_flexible_request_min_versions_list(
   request_types());
 
+struct protocol_parse_exception : public net::parsing_exception {
+    explicit protocol_parse_exception(const std::string& m)
+      : net::parsing_exception(m) {}
+};
+
 } // namespace
 
 bool flex_versions::is_flexible_request(api_key key, api_version version) {
@@ -67,9 +71,20 @@ bool flex_versions::is_api_in_schema(api_key key) noexcept {
     return first_flex_version != invalid_api;
 }
 
+namespace {
+// TODO(C++26): replace with std::sub_sat
+size_t sub_sat(size_t a, size_t b) {
+    size_t c = 0;
+    if (!__builtin_sub_overflow(a, b, &c)) {
+        return c;
+    }
+    return 0;
+}
+} // namespace
+
 ss::future<std::pair<std::optional<tagged_fields>, size_t>>
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
-parse_tags(ss::input_stream<char>& src) {
+parse_tags(ss::input_stream<char>& src, size_t max_bytes) {
     size_t total_bytes_read = 0;
     auto read_unsigned_vint =
       // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
@@ -94,26 +109,44 @@ parse_tags(ss::input_stream<char>& src) {
     while (num_tags-- > 0) {
         auto id = co_await read_unsigned_vint(total_bytes_read, src);
         auto next_len = co_await read_unsigned_vint(total_bytes_read, src);
+        if (next_len > sub_sat(max_bytes, total_bytes_read)) {
+            throw protocol_parse_exception(
+              fmt::format(
+                "tagged field {} length {} exceeds remaining message budget {}",
+                id,
+                next_len,
+                max_bytes - total_bytes_read));
+        }
         if (next_len > 128_KiB) {
-            throw std::invalid_argument(
-              fmt::format("Too large of a tagged field: {}", next_len));
+            throw protocol_parse_exception(
+              fmt::format(
+                "tagged field {} length {} exceeds 128 KiB limit",
+                id,
+                next_len));
         }
         auto buf = co_await src.read_exactly(next_len);
+        if (buf.size() != next_len) {
+            throw protocol_parse_exception(
+              fmt::format(
+                "short read for tagged field {} length {} but got {}",
+                id,
+                next_len,
+                buf.size()));
+        }
         bytes data(bytes::initialized_later{}, buf.size());
         std::copy_n(buf.begin(), buf.size(), data.begin());
         total_bytes_read += next_len;
         auto [_, succeded] = tags.emplace(tag_id(id), std::move(data));
         if (!succeded) {
-            throw std::logic_error(
-              fmt::format("Protocol error, duplicate tag id detected, {}", id));
+            throw protocol_parse_exception(
+              fmt::format("duplicate tag id detected: {}", id));
         }
     }
     co_return std::make_pair(std::move(tags), total_bytes_read);
 }
 
 namespace {
 struct invalid_buffer_size_exception : public net::parsing_exception {
-public:
     explicit invalid_buffer_size_exception(const std::string& m)
       : net::parsing_exception(m) {}
 };

src/v/kafka/protocol/flex_versions.h

@@ -42,6 +42,6 @@ class flex_versions {
 };
 
 ss::future<std::pair<std::optional<tagged_fields>, size_t>>
-parse_tags(ss::input_stream<char>&);
+parse_tags(ss::input_stream<char>&, size_t max_bytes);
 
 } // namespace kafka

src/v/kafka/server/connection_context.cc

@@ -480,7 +480,7 @@ ss::future<> connection_context::process_one_request() {
         }
     }
 
-    auto h = co_await parse_header(conn->input());
+    auto h = co_await parse_header(conn->input(), sz.value());
     _server.probe().add_bytes_received(sz.value());
     if (!h) {
         vlog(klog.debug, "could not parse header from client: {}", conn->addr);

src/v/kafka/server/protocol_utils.cc

@@ -98,7 +98,7 @@ parse_v1_header(ss::input_stream<char>& src) {
 }
 
 ss::future<std::optional<request_header>>
-parse_header(ss::input_stream<char>& src) {
+parse_header(ss::input_stream<char>& src, size_t request_size) {
     auto header = co_await parse_v1_header(src);
     if (header) {
         /// Conditionally handle v1 (flex) header
@@ -108,7 +108,7 @@ parse_header(ss::input_stream<char>& src) {
             /// reaches the request router
         } else if (
           flex_versions::is_flexible_request(header->key, header->version)) {
-            auto [tags, bytes_read] = co_await parse_tags(src);
+            auto [tags, bytes_read] = co_await parse_tags(src, request_size);
             header->tags = std::move(tags);
             header->tags_size_bytes = bytes_read;
         }

src/v/kafka/server/protocol_utils.h

@@ -23,7 +23,8 @@
 namespace kafka {
 
 // TODO: move to iobuf_parser
-ss::future<std::optional<request_header>> parse_header(ss::input_stream<char>&);
+ss::future<std::optional<request_header>>
+parse_header(ss::input_stream<char>&, size_t request_size);
 
 ss::scattered_message<char> response_as_scattered(response_ptr response);