From 9c7e1411865e2f1a1fcfd7c42939f151b2186f8c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 03:19:28 +0000 Subject: [PATCH] fix(node): bound the emitted Arrow IPC stream length at the producer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Arrow IPC encode guards bounded the *input* length (`data_len` in `uint8_layout`, the body offset in the fast-path builder) against `MAX_IPC_BYTES`, but the emitted stream is ~1.125x larger (validity bitmap, 64-byte alignment padding, schema + record-batch message blocks). Every receiver bounds the *whole stream* against the same `MAX_IPC_BYTES`, so a payload whose `data_len` sits just under the cap could encode and be sent, then be rejected by all receivers — a silent drop on the zenoh path. Guard the resulting stream length instead, failing loudly at the producer: - `uint8_layout` now checks the computed `total` (covers the `send_output_raw` UInt8 construct-in-place path, incl. the Python bindings). - `send_output_array` runs `check_ipc_size` on the fast-path length and on the fallback stream before allocating, so anything the producer emits is decodable by every receiver. `check_ipc_size` is now shared (producer + decode guard); its doc notes the dual role. Fixes #2586 Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01C79MriEbXiurvZQQ3QoCkU --- .../node/src/node/arrow_utils/ipc_encode.rs | 31 ++++++++++++++++--- apis/rust/node/src/node/mod.rs | 7 +++++ 2 files changed, 33 insertions(+), 5 deletions(-) diff --git a/apis/rust/node/src/node/arrow_utils/ipc_encode.rs b/apis/rust/node/src/node/arrow_utils/ipc_encode.rs index 9763509006..9479edab88 100644 --- a/apis/rust/node/src/node/arrow_utils/ipc_encode.rs +++ b/apis/rust/node/src/node/arrow_utils/ipc_encode.rs @@ -853,11 +853,14 @@ impl InputDecoder { } } -/// Reject an IPC payload larger than [`super::MAX_IPC_BYTES`] before decoding. -/// Defense-in-depth against an oversized peer-controlled zenoh payload, mirroring -/// the guard in [`decode_arrow_ipc_zero_copy`](super::decode_arrow_ipc_zero_copy) -/// (the persistent-decoder paths receive the same untrusted bytes). -fn check_ipc_size(len: usize) -> eyre::Result<()> { +/// Reject an IPC stream larger than [`super::MAX_IPC_BYTES`]. +/// +/// Used on both sides so the encode and decode limits agree: as a producer-side +/// guard (so a stream the producer emits is always decodable by every receiver, +/// rather than silently dropped — see #2586) and as defense-in-depth on decode +/// against an oversized peer-controlled zenoh payload, mirroring the guard in +/// [`decode_arrow_ipc_zero_copy`](super::decode_arrow_ipc_zero_copy). +pub(crate) fn check_ipc_size(len: usize) -> eyre::Result<()> { if len > super::MAX_IPC_BYTES { bail!( "Arrow IPC payload too large: {len} bytes (max {})", @@ -1462,6 +1465,24 @@ mod tests { assert_eq!(decoded.len(), 0); } + /// A `UInt8` payload whose *encoded stream* would exceed `MAX_IPC_BYTES` + /// must be rejected at encode time even though `data_len` itself is within + /// the limit: the validity bitmap, alignment padding, and message blocks + /// make the stream larger, and every receiver bounds the whole stream. The + /// producer must fail loudly rather than emit a stream that is silently + /// dropped on the zenoh path. Regression for #2586. + #[test] + fn uint8_ipc_len_rejects_oversized_stream() { + // `data_len == MAX_IPC_BYTES` clears a naive `data_len` check, but the + // ~1.125x stream overhead pushes the emitted stream over the cap. + let err = uint8_ipc_len(super::super::MAX_IPC_BYTES) + .expect_err("stream exceeding MAX_IPC_BYTES must be rejected") + .to_string(); + assert!(err.contains("too large"), "unexpected error: {err}"); + // A comfortably-small payload still encodes to an in-bounds stream. + assert!(uint8_ipc_len(1024).unwrap() <= super::super::MAX_IPC_BYTES); + } + /// The schema-once receive path must decode a 0-row (empty) batch, not drop /// it. A schema-less batch carries no trailing end-of-stream marker, and /// arrow's `StreamDecoder` only emits a zero-length body on the poll *after* diff --git a/apis/rust/node/src/node/mod.rs b/apis/rust/node/src/node/mod.rs index fb10dd8b9f..c35356165d 100644 --- a/apis/rust/node/src/node/mod.rs +++ b/apis/rust/node/src/node/mod.rs @@ -2712,6 +2712,11 @@ impl SampleAllocator { pub fn encode_arrow(&self, array: &ArrayData) -> NodeResult { let sample = match ipc_encode::PreparedIpc::new(array) { Some(prepared) => { + // Reject a stream larger than every receiver will accept before + // emitting it, rather than sending an undecodable payload that + // is silently dropped on the zenoh path (#2586). + ipc_encode::check_ipc_size(prepared.byte_len()) + .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?; // Prepare once: size the sample from the prepared layout, then // encode into it — avoids rebuilding the layout + IPC headers. let mut sample = self.allocate(prepared.byte_len())?; @@ -2723,6 +2728,8 @@ impl SampleAllocator { None => { let bytes = ipc_encode::encode_ipc_to_vec(array) .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?; + ipc_encode::check_ipc_size(bytes.len()) + .map_err(|e| NodeError::Output(format!("Arrow IPC encode: {e}")))?; let mut sample = self.allocate(bytes.len())?; sample.copy_from_slice(&bytes); sample