kafka/protocol: bound parse_tags by remaining message bytes - #30191
Conversation
d3753ab to
ec227f5
Compare
There was a problem hiding this comment.
Pull request overview
This PR tightens Kafka flexible tagged-field parsing by adding an explicit byte budget (max_bytes) and ensuring parse failures are surfaced as net::parsing_exception-derived exceptions, so the server classifies them as parse errors.
Changes:
- Change
parse_tagsto acceptmax_bytesand enforce both a remaining-message budget and a per-field 128 KiB cap. - Thread request-size / remaining-bytes information into tag parsing from server (
parse_header) and client (transport) call sites. - Replace
std::invalid_argument/std::logic_errorthrows in tag parsing withnet::parsing_exceptionsubclasses.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| src/v/kafka/server/protocol_utils.h | Updates parse_header signature to accept request size. |
| src/v/kafka/server/protocol_utils.cc | Passes size into tag parsing when handling flexible headers. |
| src/v/kafka/server/connection_context.cc | Supplies request size from parse_size into parse_header. |
| src/v/kafka/protocol/flex_versions.h | Extends parse_tags API to include max_bytes. |
| src/v/kafka/protocol/flex_versions.cc | Implements byte-budget enforcement and parsing_exception-based errors. |
| src/v/kafka/protocol/BUILD | Adds the Bazel dependency needed for 128_KiB units. |
| src/v/kafka/client/transport.cc | Bounds response tag parsing by bytes_remaining. |
| } 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; |
There was a problem hiding this comment.
parse_tags is given request_size (the whole message size), but parse_tags only tracks bytes consumed within the tags section. This means tags can consume up to request_size bytes, exceeding the bytes remaining after parse_v1_header and causing remaining = size - request_header_size - client_id - tags_size_bytes to underflow later (potentially leading to huge reads). Pass the bytes remaining after header+client_id (and validate request_size >= header_bytes) instead of the full request size.
| if (next_len > 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( |
There was a problem hiding this comment.
The new max_bytes enforcement in parse_tags is security-sensitive (bounds checks + exception type changes), but there are no unit tests covering: remaining-budget violations, the 128KiB per-field cap, duplicate tag ids, and short reads. Consider adding focused tests under src/v/kafka/protocol/tests/ to ensure these cases throw net::parsing_exception and report correct bytes_read.
ec227f5 to
e0f0bb3
Compare
parse_tags now takes a max_bytes parameter and checks two limits before allocating each tagged field: 1. A cumulative budget check: the field length cannot exceed the bytes remaining in the message. Both callers already have a validated total message size (connection_context has sz from parse_size; transport has bytes_remaining), so this bound is derived from actual data rather than a hardcoded constant. 2. A per-field cap of 128 KiB, retained as a sanity limit on any single field. All exceptions are now net::parsing_exception subclasses so the server classifies them as parse-error disconnects rather than unexpected errors.
e0f0bb3 to
17b5168
Compare
|
|
||
| namespace { | ||
| // TODO(C++26): replace with std::sub_sat | ||
| size_t sub_sat(size_t a, size_t b) { |
There was a problem hiding this comment.
i wonder how many other places in our parsing logic could use some serious combing through with saturating arithmetic
WillemKauf
left a comment
WillemKauf
left a comment
There was a problem hiding this comment.
seems fine and sufficiently hardened to me
|
/backport v26.1.x |
|
/backport v25.3.x |
|
/backport v25.2.x |
|
Failed to create a backport PR to v25.3.x branch. I tried: |
|
Failed to create a backport PR to v25.2.x branch. I tried: |
parse_tags now takes a max_bytes parameter and checks two limits
before allocating each tagged field:
A cumulative budget check: the field length cannot exceed the
bytes remaining in the message. Both callers already have a
validated total message size (connection_context has sz from
parse_size; transport has bytes_remaining), so this bound is
derived from actual data rather than a hardcoded constant.
A per-field cap of 128 KiB, retained as a sanity limit on any
single field.
All exceptions are now net::parsing_exception subclasses so the
server classifies them as parse-error disconnects rather than
unexpected errors.
Backports Required
Release Notes
Improvements