This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-11881-35b20e63ddf211d5d011ff28bea44ae400fbf5c2 in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
commit fd2b5d156d80c986ff9ba5ace2db84500edf50ec Author: Wleeaf <[email protected]> AuthorDate: Tue Oct 27 17:38:27 2026 +0000 fix(arrow-ipc): return an error instead of panicking on out-of-bounds IPC buffers (#11881) # Which issue does this PR close? - Closes #11856. - Closes #11857. # Rationale for this change The IPC reader slices buffers using offsets and lengths that come from the message, without checking them against the data they point into. `Buffer::slice_with_length` panics when the range is out of bounds, so corrupt input panics instead of returning an error. # What changes are included in this PR? - A small `slice_buffer` helper in `reader.rs` that checks the range (the same condition `slice_with_length` asserts) and returns an `ArrowError::IpcError` instead of panicking. - `read_buffer` uses it for the buffer offset and length from the message metadata (#11857). - The union reader uses it for the type ids buffer and the dense offsets buffer, whose expected length comes from the field node (#11856). # Are these changes tested? Yes. The reproducers from both issues are added as `test_read_record_batch_buffer_out_of_bounds` and `test_union_short_buffers`. The union test also covers a dense union with a short offsets buffer. Both tests panic without the fix and pass with it. # Are there any significant user-facing changes? No. AI usage: this change was prepared with Claude Code (an AI coding agent). The fix and tests were generated by it and checked against the failing tests from the issues. --------- Co-authored-by: Wleeaf <[email protected]> --- arrow-buffer/src/buffer/immutable.rs | 51 +++++++++++++++--- arrow-buffer/src/error.rs | 38 +++++++++++++ arrow-buffer/src/lib.rs | 2 +- arrow-ipc/src/reader.rs | 101 +++++++++++++++++++++++++++++++++-- 4 files changed, 178 insertions(+), 14 deletions(-) diff --git a/arrow-buffer/src/buffer/immutable.rs b/arrow-buffer/src/buffer/immutable.rs index bd6837aa6a..658d045f06 100644 --- a/arrow-buffer/src/buffer/immutable.rs +++ b/arrow-buffer/src/buffer/immutable.rs @@ -23,7 +23,7 @@ use std::sync::Arc; use crate::BufferBuilder; use crate::alloc::{Allocation, Deallocation}; use crate::util::bit_chunk_iterator::{BitChunks, UnalignedBitChunk}; -use crate::{bit_util, bytes::Bytes, native::ArrowNativeType}; +use crate::{SliceOutOfBoundsError, bit_util, bytes::Bytes, native::ArrowNativeType}; #[cfg(feature = "pool")] use crate::pool::MemoryPool; @@ -296,19 +296,31 @@ impl Buffer { /// # Panics /// Panics iff `(offset + length)` is larger than the existing length. pub fn slice_with_length(&self, offset: usize, length: usize) -> Self { - assert!( - offset.saturating_add(length) <= self.length, - "the offset of the new Buffer cannot exceed the existing length: slice offset={offset} length={length} selflen={}", - self.length - ); + self.try_slice_with_length(offset, length) + .unwrap_or_else(|e| panic!("{e}")) + } + + /// Returns a new [Buffer] that is a slice of this buffer starting at `offset`, + /// with `length` bytes, or an error if `(offset + length)` is larger than the + /// existing length. + /// + /// This is the non-panicking version of [`Self::slice_with_length`]. + pub fn try_slice_with_length( + &self, + offset: usize, + length: usize, + ) -> Result<Self, SliceOutOfBoundsError> { + if offset.saturating_add(length) > self.length { + return Err(SliceOutOfBoundsError::new(offset, length, self.length)); + } // Safety: // offset + length <= self.length let ptr = unsafe { self.ptr.add(offset) }; - Self { + Ok(Self { data: self.data.clone(), ptr, length, - } + }) } /// Returns a pointer to the start of this buffer. @@ -707,6 +719,29 @@ mod tests { assert_eq!([0, 1, 2, 3, 4], buf2.as_slice()); } + #[test] + fn test_try_slice_with_length() { + let buf = Buffer::from(&[2, 4, 6, 8, 10]); + assert_eq!( + [4, 6, 8], + buf.try_slice_with_length(1, 3).unwrap().as_slice() + ); + assert!(buf.try_slice_with_length(5, 0).unwrap().is_empty()); + assert!(buf.try_slice_with_length(6, 0).is_err()); + assert!(buf.try_slice_with_length(usize::MAX, 1).is_err()); + assert_eq!( + buf.try_slice_with_length(3, 3).unwrap_err().to_string(), + "the offset of the new Buffer cannot exceed the existing length: slice offset=3 length=3 selflen=5" + ); + + let buf2 = buf.slice(2); + assert_eq!( + [8, 10], + buf2.try_slice_with_length(1, 2).unwrap().as_slice() + ); + assert!(buf2.try_slice_with_length(1, 3).is_err()); + } + #[test] fn test_slice() { let buf = Buffer::from(&[2, 4, 6, 8, 10]); diff --git a/arrow-buffer/src/error.rs b/arrow-buffer/src/error.rs index ad769848f5..c814617e55 100644 --- a/arrow-buffer/src/error.rs +++ b/arrow-buffer/src/error.rs @@ -87,3 +87,41 @@ impl std::fmt::Display for OverflowError { } impl std::error::Error for OverflowError {} + +/// A slice that does not fit in the buffer it is taken from. +/// +/// Returned by [`Buffer::try_slice_with_length`](crate::Buffer::try_slice_with_length), +/// the alternative to [`Buffer::slice_with_length`](crate::Buffer::slice_with_length) +/// that does not panic. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SliceOutOfBoundsError { + offset: usize, + length: usize, + buffer_len: usize, +} + +impl SliceOutOfBoundsError { + pub(crate) const fn new(offset: usize, length: usize, buffer_len: usize) -> Self { + Self { + offset, + length, + buffer_len, + } + } +} + +impl std::fmt::Display for SliceOutOfBoundsError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let Self { + offset, + length, + buffer_len, + } = self; + write!( + f, + "the offset of the new Buffer cannot exceed the existing length: slice offset={offset} length={length} selflen={buffer_len}" + ) + } +} + +impl std::error::Error for SliceOutOfBoundsError {} diff --git a/arrow-buffer/src/lib.rs b/arrow-buffer/src/lib.rs index 345c8f17e6..c5a85457fe 100644 --- a/arrow-buffer/src/lib.rs +++ b/arrow-buffer/src/lib.rs @@ -54,7 +54,7 @@ pub mod builder; pub use builder::*; mod error; -pub use error::OverflowError; +pub use error::{OverflowError, SliceOutOfBoundsError}; mod bigint; pub use bigint::i256; diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs index 35ca545ac3..9a2ce3d720 100644 --- a/arrow-ipc/src/reader.rs +++ b/arrow-ipc/src/reader.rs @@ -63,7 +63,9 @@ fn read_buffer( decompression_context: &mut DecompressionContext, ) -> Result<Buffer, ArrowError> { let start_offset = buf.offset() as usize; - let buf_data = a_data.slice_with_length(start_offset, buf.length() as usize); + let buf_data = a_data + .try_slice_with_length(start_offset, buf.length() as usize) + .map_err(|e| ArrowError::IpcError(e.to_string()))?; // corner case: empty buffer match (buf_data.is_empty(), compression_codec) { (true, _) | (_, None) => Ok(buf_data), @@ -72,6 +74,7 @@ fn read_buffer( } } } + impl RecordBatchDecoder<'_> { /// Coordinates reading arrays based on data types. /// @@ -201,13 +204,19 @@ impl RecordBatchDecoder<'_> { self.next_buffer()?; } - let type_ids: ScalarBuffer<i8> = - self.next_buffer()?.slice_with_length(0, len).into(); + let type_ids: ScalarBuffer<i8> = self + .next_buffer()? + .try_slice_with_length(0, len) + .map_err(|e| ArrowError::IpcError(e.to_string()))? + .into(); let value_offsets = match mode { UnionMode::Dense => { - let offsets: ScalarBuffer<i32> = - self.next_buffer()?.slice_with_length(0, len * 4).into(); + let offsets: ScalarBuffer<i32> = self + .next_buffer()? + .try_slice_with_length(0, len * 4) + .map_err(|e| ArrowError::IpcError(e.to_string()))? + .into(); Some(offsets) } UnionMode::Sparse => None, @@ -2314,6 +2323,88 @@ mod tests { } } + #[test] + fn test_read_record_batch_buffer_out_of_bounds() { + use crate::r#gen::Message::*; + use flatbuffers::FlatBufferBuilder; + + let schema = Arc::new(Schema::new(vec![Field::new("col", DataType::Int32, false)])); + let body = Buffer::from(vec![0u8; 8]); + let mut fbb = FlatBufferBuilder::new(); + let nodes = fbb.create_vector(&[FieldNode::new(2, 0)]); + // Empty validity buffer followed by two Int32 values whose range exceeds the body. + let buffers = fbb.create_vector(&[crate::Buffer::new(0, 0), crate::Buffer::new(4, 8)]); + let batch = RecordBatch::create( + &mut fbb, + &RecordBatchArgs { + length: 2, + nodes: Some(nodes), + buffers: Some(buffers), + ..Default::default() + }, + ); + fbb.finish_minimal(batch); + let batch = flatbuffers::root::<RecordBatch>(fbb.finished_data()).unwrap(); + + read_record_batch( + &body, + batch, + schema, + &HashMap::new(), + None, + &MetadataVersion::V5, + ) + .expect_err("invalid IPC buffer range must return an error"); + } + + #[test] + fn test_union_short_buffers() { + use crate::r#gen::Message::*; + + let fields = UnionFields::from_fields([Field::new("child", DataType::Null, true)]); + // All buffer ranges fit in the body. Only the union length is inconsistent: + // sparse has one type id for two rows, dense has one offset for two rows + let cases = [ + (UnionMode::Sparse, vec![crate::Buffer::new(0, 1)]), + ( + UnionMode::Dense, + vec![crate::Buffer::new(0, 2), crate::Buffer::new(8, 4)], + ), + ]; + for (mode, buffers) in cases { + let schema = Arc::new(Schema::new(vec![Field::new( + "union", + DataType::Union(fields.clone(), mode), + false, + )])); + + let mut fbb = flatbuffers::FlatBufferBuilder::new(); + let nodes = fbb.create_vector(&[FieldNode::new(2, 0), FieldNode::new(2, 2)]); + let buffers = fbb.create_vector(&buffers); + let batch = RecordBatch::create( + &mut fbb, + &RecordBatchArgs { + length: 2, + nodes: Some(nodes), + buffers: Some(buffers), + ..Default::default() + }, + ); + fbb.finish_minimal(batch); + let batch = flatbuffers::root::<RecordBatch>(fbb.finished_data()).unwrap(); + + read_record_batch( + &Buffer::from(vec![0u8; 16]), + batch, + schema, + &HashMap::new(), + None, + &MetadataVersion::V5, + ) + .expect_err("a short union buffer should return an error"); + } + } + /// A `Utf8View` batch whose variadic buffer count is `count`, with two buffers /// in the message. Returns the error from reading it with the given projection. fn read_batch_with_variadic_count(count: i64, projection: Option<&[usize]>) -> ArrowError {
