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