krishvishal commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3827918135
########## gateways/kafka/src/protocol/bounds_guard.rs: ########## @@ -0,0 +1,721 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Pre-decode bounds guard against unbounded allocation in `kafka_protocol`'s decode path. +//! +//! `kafka_protocol` 0.17 validates every wire-declared array/string/bytes length against zero +//! but never against what remains in the frame (`kafka-protocol-0.17.0/src/protocol/types.rs:988` +//! `Vec::with_capacity(n as usize)` from a wire `Int32`, `:1096` the same from an unsigned +//! varint). A tiny frame that declares a huge count reaches `handle_alloc_error`, which calls +//! `abort()` - not a panic, not catchable, and it takes down every connection on the process, not +//! just the one that sent it. Reproduced: a 20-byte `CreateTopics` v5 frame claiming a huge +//! topics count requests ~481 GB; a 4-byte Metadata v0 frame requests ~143 GiB. No SASL/TLS gates +//! any of `SUPPORTED_RANGES`, so this is reachable by anyone who can `connect()`. +//! +//! This module walks the same field shape `kafka_protocol`'s real decode walks for each of the +//! six accepted message types, but only to validate every length-prefixed field (array count, +//! string length, bytes length, tagged-field size) against what could still fit in the bytes +//! remaining in the frame - it never materializes a value or allocates a collection. Call the +//! matching `validate_*_shape` function before handing the body to `kafka_protocol`. +//! +//! Kept independent of `kafka_protocol`'s own type decoders on purpose: if this walker's field +//! order ever drifts from the real schema, the worst outcome is a false-positive rejection of a +//! legitimate request (caught immediately by the wire-fixture round-trip tests), never a false +//! negative that lets an oversized count back through to the crate. + +use bytes::{Buf, Bytes}; + +use crate::error::{KafkaProtocolError, Result}; + +/// Same bound the pre-migration hand-rolled codec used: large enough for any real request, small +/// enough that even `MAX_REQUEST_ELEMENTS` identically-sized nested arrays cannot approach a +/// meaningful fraction of memory. +const MAX_COLLECTION_LEN: usize = 65_536; + +/// Cumulative cap on the total array elements this guard will walk across one request. A single +/// array's count is capped by [`MAX_COLLECTION_LEN`], but not the product across nested arrays +/// (`topics` x `partitions`) - this budget bounds the sum across the whole frame. +const MAX_REQUEST_ELEMENTS: usize = MAX_COLLECTION_LEN; + +struct ShapeCursor { + bytes: Bytes, + element_budget: usize, +} + +impl ShapeCursor { + const fn new(body: Bytes) -> Self { + Self { + bytes: body, + element_budget: MAX_REQUEST_ELEMENTS, + } + } + + fn remaining(&self) -> usize { + self.bytes.remaining() + } + + fn ensure(&self, needed: usize) -> Result<()> { + let remaining = self.bytes.remaining(); + if remaining < needed { + return Err(KafkaProtocolError::BufferUnderflow { needed, remaining }); + } + Ok(()) + } + + fn read_i8(&mut self) -> Result<i8> { + self.ensure(1)?; + Ok(self.bytes.get_i8()) + } + + fn read_bool(&mut self) -> Result<bool> { + Ok(self.read_i8()? != 0) + } + + fn read_i16(&mut self) -> Result<i16> { + self.ensure(2)?; + Ok(self.bytes.get_i16()) + } + + fn read_i32(&mut self) -> Result<i32> { + self.ensure(4)?; + Ok(self.bytes.get_i32()) + } + + fn read_i64(&mut self) -> Result<i64> { + self.ensure(8)?; + Ok(self.bytes.get_i64()) + } + + fn read_varint(&mut self) -> Result<u64> { Review Comment: **C1. Guard unsound: 18-byte frame crashes the process (SIGABRT)** `read_varint` here reads up to 10 bytes (`shift >= 64`). Crate `UnsignedVarInt::decode` reads 5 max, no error on over-long (`types.rs:120-131`). 5th continuation bit desyncs cursors: guard validates different bytes than the decoder allocates from. Guard accepts, `Vec::with_capacity(attacker_n)` aborts via `handle_alloc_error`, kills all connections. Module claim `:33-38` "false-positive only, never false-negative" is wrong. ``` Produce v9 18B guard=ACCEPT decoder count=4294967295 384.0 GiB CreateTopics v5 26B guard=ACCEPT decoder count=4294967295 224.0 GiB POC produce: "memory allocation of 412316860224 bytes failed" exit=134 (SIGABRT) ``` Frame: `81 80 80 80 80 80 80 80 80 00 00 FF FF FF FF 0F 01 00`. Exploitable: Produce, CreateTopics, Fetch v12, ListOffsets v6. Reach: anyone on TCP 9093, no auth, single packet, default config. `tests/common/codec.rs` copies the same 10-byte bug, so tests are blind. Fix: `read_varint` = crate copy (`u32`, `for i in 0..5`, no error on trailing continuation). Verified: exploits rejected, guard tests + DoS repros pass. Breaks: drops untested `"varint overflows 64 bits"` arm. ########## .github/actions/rust/pre-merge/action.yml: ########## @@ -279,6 +279,32 @@ runs: compile_duration=$((compile_end - compile_start)) echo "::notice::Tests compiled in ${compile_duration}s ($(date -ud @${compile_duration} +'%M:%S'))" + # api_handler_tests, version_firewall_tests, and server_e2e_tests need gitignored + # wire fixtures. Generate when iggy-gateway-kafka is in the DAG test scope, or (on + # a full-workspace run) when gateways/** changed vs origin/master — avoid building + # kafka-message-gen / kafka-protocol when the gateway was not touched. + NEEDS_KAFKA_FIXTURES=false + if grep -q 'package(iggy-gateway-kafka)' <<< "$NEXTEST_FILTER"; then + NEEDS_KAFKA_FIXTURES=true + elif [[ -z "$NEXTEST_FILTER" ]]; then + if git diff --name-only origin/master...HEAD 2>/dev/null | grep -qE '^gateways/'; then Review Comment: **C3. Wire-correctness tests never run in CI (from run logs)** `git diff origin/master...HEAD 2>/dev/null` failure is silenced, so on a `gateways/`-only PR under full-workspace mode fixtures are skipped and `KAFKA_FIXTURES_REQUIRED` is never set. Run 32256598833, all 3 partitions: ``` ##[notice]Full workspace build (no DAG filter available) ##[notice]Skipping Kafka wire fixtures (full suite, gateways/** unchanged) ``` `fetch_stub_response_returns_retriable_not_leader` PASSED in 0.005 s, all 9 fixture iterations `continue` (`fixtures.rs:72`, `version_firewall_tests.rs:570-581`). Produce v2 / Fetch v3 / ListOffsets v0 impossible to emit (`API_REGISTRY` floors 3/4/1), so correct wiring turns CI red. Master-push lane (`coverage-baseline.yml`) inlines its own nextest, no fixtures, omits `gateways` from `rust)` pathspec. Fix: gate `request_body_for_scoped_api` on `fixture_exists`; fix this detection; add `gateways` to coverage pathspec. ########## gateways/kafka/src/protocol/bounds_guard.rs: ########## @@ -0,0 +1,721 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Pre-decode bounds guard against unbounded allocation in `kafka_protocol`'s decode path. +//! +//! `kafka_protocol` 0.17 validates every wire-declared array/string/bytes length against zero +//! but never against what remains in the frame (`kafka-protocol-0.17.0/src/protocol/types.rs:988` +//! `Vec::with_capacity(n as usize)` from a wire `Int32`, `:1096` the same from an unsigned +//! varint). A tiny frame that declares a huge count reaches `handle_alloc_error`, which calls +//! `abort()` - not a panic, not catchable, and it takes down every connection on the process, not +//! just the one that sent it. Reproduced: a 20-byte `CreateTopics` v5 frame claiming a huge +//! topics count requests ~481 GB; a 4-byte Metadata v0 frame requests ~143 GiB. No SASL/TLS gates +//! any of `SUPPORTED_RANGES`, so this is reachable by anyone who can `connect()`. +//! +//! This module walks the same field shape `kafka_protocol`'s real decode walks for each of the +//! six accepted message types, but only to validate every length-prefixed field (array count, +//! string length, bytes length, tagged-field size) against what could still fit in the bytes +//! remaining in the frame - it never materializes a value or allocates a collection. Call the +//! matching `validate_*_shape` function before handing the body to `kafka_protocol`. +//! +//! Kept independent of `kafka_protocol`'s own type decoders on purpose: if this walker's field +//! order ever drifts from the real schema, the worst outcome is a false-positive rejection of a +//! legitimate request (caught immediately by the wire-fixture round-trip tests), never a false +//! negative that lets an oversized count back through to the crate. + +use bytes::{Buf, Bytes}; + +use crate::error::{KafkaProtocolError, Result}; + +/// Same bound the pre-migration hand-rolled codec used: large enough for any real request, small +/// enough that even `MAX_REQUEST_ELEMENTS` identically-sized nested arrays cannot approach a +/// meaningful fraction of memory. +const MAX_COLLECTION_LEN: usize = 65_536; + +/// Cumulative cap on the total array elements this guard will walk across one request. A single +/// array's count is capped by [`MAX_COLLECTION_LEN`], but not the product across nested arrays +/// (`topics` x `partitions`) - this budget bounds the sum across the whole frame. +const MAX_REQUEST_ELEMENTS: usize = MAX_COLLECTION_LEN; Review Comment: **C2. Element-budget amplification: 16 connections saturate the runtime (measured)** `MAX_REQUEST_ELEMENTS = 65_536` is 16x too loose, and one constant serves per-array cap and cumulative budget both. Consumed by `encode_produce_response` (`responses.rs:63-81`). Produce v9, 1 topic x 65,535 partitions: ``` req=393,240 B resp=2,162,676 B (5.5x wire) heap=17.0 MiB (45x) 13.6 ms/req, zero .await ``` `handle_request` synchronous `fn`, no yield: 13.6 ms un-preemptible CPU/req. Default `worker_threads = num_cpus`, ~16 connections peg every worker. `max_connections = 1024`, slow reader pins ~2.1 GB. Reach: default config, no auth, only live path. Fix: size cumulative budget separately (e.g. 4096), and/or reject when projected response > `max_frame_size`. Breaks: legit over-budget client gets `INVALID_REQUEST`. Wire contract at merge, pick the number now. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
