Validate untrusted footer metadata on open instead of panicking

vortex-array/src/serde.rs

@@ -613,26 +613,47 @@ impl SerializedArray {
         // SAFETY: fb_buffer was already validated by validate_array_tree above.
         let fb_array = unsafe { fba::root_as_array_unchecked(fb_buffer.as_ref()) };
 
-        let mut offset = 0;
+        let mut offset = 0usize;
         let buffers = fb_array
             .buffers()
             .unwrap_or_default()
             .iter()
             .enumerate()
             .map(|(idx, fb_buf)| {
-                offset += fb_buf.padding() as usize;
-                let buffer_len = fb_buf.length() as usize;
-                let alignment = Alignment::from_exponent(fb_buf.alignment_exponent());
-
                 let idx = u32::try_from(idx).vortex_expect("buffer count must fit in u32");
+
+                // The padding, length, and resulting offsets all come from the flatbuffer, which
+                // may be corrupt. Use checked arithmetic so malformed metadata returns a
+                // `VortexError` rather than panicking (see issue #8819).
+                let buffer_len = fb_buf.length() as usize;
+                let start = offset
+                    .checked_add(fb_buf.padding() as usize)
+                    .ok_or_else(|| {
+                        vortex_err!("Buffer {idx} offset overflows when adding its padding")
+                    })?;
+                let end = start.checked_add(buffer_len).ok_or_else(|| {
+                    vortex_err!("Buffer {idx} offset overflows when adding its length")
+                })?;
+
+                // The alignment exponent comes from the flatbuffer and may be corrupt, so validate
+                // it rather than panicking on a too-large shift (see issue #8819).
+                let alignment = Alignment::try_from_exponent(fb_buf.alignment_exponent())?;
                 let handle = if let Some(host_data) = buffer_overrides.get(&idx) {
                     BufferHandle::new_host(host_data.clone()).ensure_aligned(alignment)?
                 } else {
-                    let buffer = segment.slice(offset..(offset + buffer_len));
-                    buffer.ensure_aligned(alignment)?
+                    // Bounds-check against the segment so an out-of-range buffer returns a
+                    // `VortexError` rather than panicking when slicing (see issue #8819).
+                    if end > segment.len() {
+                        vortex_bail!(
+                            "Buffer {idx} at offset {start} with length {buffer_len} is out of \
+                             bounds of the {}-byte segment",
+                            segment.len(),
+                        );
+                    }
+                    segment.slice(start..end).ensure_aligned(alignment)?
                 };
 
-                offset += buffer_len;
+                offset = end;
                 Ok(handle)
             })
             .collect::<VortexResult<Arc<[_]>>>()?;
@@ -695,3 +716,57 @@ impl TryFrom<BufferHandle> for SerializedArray {
         Self::try_from(value.try_to_host_sync()?)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use vortex_buffer::ByteBufferMut;
+
+    use super::*;
+    use crate::IntoArray;
+    use crate::array_session;
+    use crate::arrays::PrimitiveArray;
+
+    /// A corrupt array tree can declare a buffer that extends past the backing segment. Slicing
+    /// such a buffer must return a [`VortexError`] rather than panicking (see issue #8819).
+    #[test]
+    fn from_flatbuffer_and_segment_rejects_out_of_bounds_buffer() -> VortexResult<()> {
+        let session = array_session();
+        let array_ctx = ArrayContext::empty();
+
+        // Serialize a simple array so we have a valid array tree flatbuffer whose declared buffer
+        // lengths describe the trailing data segment.
+        let serialized = PrimitiveArray::from_iter([1i32, 2, 3, 4])
+            .into_array()
+            .serialize(&array_ctx, &session, &SerializeOptions::default())?;
+
+        let mut concat = ByteBufferMut::empty();
+        for buf in serialized {
+            concat.extend_from_slice(buf.as_ref());
+        }
+        let value = concat.freeze().aligned(Alignment::none());
+
+        // Split the blob into the trailing flatbuffer and the leading data segment, mirroring
+        // `SerializedArray::try_from`.
+        let fb_length = u32::try_from_le_bytes(&value.as_slice()[value.len() - 4..])? as usize;
+        let fb_offset = value.len() - 4 - fb_length;
+        assert!(
+            fb_offset > 0,
+            "the array must have at least one data buffer"
+        );
+        let array_tree = value.slice(fb_offset..fb_offset + fb_length);
+
+        // Truncate the data segment by one byte so the declared buffer no longer fits.
+        let truncated = BufferHandle::new_host(value.slice(0..fb_offset - 1));
+
+        let Some(err) = SerializedArray::from_flatbuffer_and_segment(array_tree, truncated).err()
+        else {
+            vortex_bail!("out-of-bounds buffer must be rejected");
+        };
+        assert!(
+            err.to_string().contains("out of bounds"),
+            "unexpected error: {err}"
+        );
+
+        Ok(())
+    }
+}

vortex-buffer/src/alignment.rs

@@ -6,6 +6,8 @@ use std::ops::Deref;
 
 use vortex_error::VortexError;
 use vortex_error::VortexExpect;
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
 
 /// The alignment of a buffer.
@@ -120,6 +122,22 @@ impl Alignment {
     pub const fn from_exponent(exponent: u8) -> Self {
         Self::new(1 << exponent)
     }
+
+    /// Create from the log2 exponent of the alignment, returning an error rather than panicking if
+    /// `1 << exponent` would overflow `usize`.
+    ///
+    /// Prefer this over [`from_exponent`](Self::from_exponent) when the exponent originates from
+    /// untrusted input such as a serialized file, where a too-large value must not panic.
+    #[inline]
+    pub fn try_from_exponent(exponent: u8) -> VortexResult<Self> {
+        if u32::from(exponent) >= usize::BITS {
+            vortex_bail!(
+                "Alignment exponent {exponent} is too large for a {}-bit usize",
+                usize::BITS
+            );
+        }
+        Ok(Self::new(1 << exponent))
+    }
 }
 
 impl Display for Alignment {
@@ -236,6 +254,18 @@ mod test {
         assert!(Alignment::try_from(3u32).is_err());
     }
 
+    #[test]
+    fn try_from_exponent() {
+        match Alignment::try_from_exponent(10) {
+            Ok(alignment) => assert_eq!(alignment, Alignment::new(1024)),
+            Err(err) => panic!("valid exponent should succeed: {err}"),
+        }
+        // Exponents whose `1 << exponent` would overflow a usize must error rather than panic.
+        // 64 is `>= usize::BITS` on both 32- and 64-bit targets.
+        assert!(Alignment::try_from_exponent(64).is_err());
+        assert!(Alignment::try_from_exponent(u8::MAX).is_err());
+    }
+
     #[test]
     fn into_u32() {
         let alignment = Alignment::new(64);

vortex-file/src/footer/deserializer.rs

@@ -40,7 +40,8 @@ pub struct FooterDeserializer {
 
     // Internal state that we accumulate
 
-    // The file size, possibly provided externally.
+    // The size of the file containing the serialized footer. For a standalone footer, this is the
+    // size of the footer blob rather than the data file described by the footer.
     file_size: Option<u64>,
     // The postscript, once we've parsed it.
     postscript: Option<Postscript>,
@@ -71,13 +72,17 @@ impl FooterDeserializer {
         self
     }
 
-    /// Provide the total file size.
+    /// Provide the size of the file containing this serialized footer.
+    ///
+    /// For a footer read from the end of a Vortex file, this is the file size. For a standalone
+    /// blob created by [`crate::footer::FooterSerializer`] with its default offset, this is the
+    /// size of that blob, not the size of the data file described by the footer.
     pub fn with_size(mut self, file_size: u64) -> Self {
         self.file_size = Some(file_size);
         self
     }
 
-    /// Provide or clear the total file size.
+    /// Provide or clear the size of the file containing this serialized footer.
     pub fn with_some_size(mut self, file_size: Option<u64>) -> Self {
         self.file_size = file_size;
         self
@@ -120,11 +125,18 @@ impl FooterDeserializer {
         // The other postscript segments are required, so now we figure out our the offset that
         // contains all the required segments.
 
-        // The initial offset is the file size - the size of our initial read.
+        // The initial offset is the file size minus the size of our initial read.
         let Some(file_size) = self.file_size else {
             return Ok(DeserializeStep::NeedFileSize);
         };
-        let initial_offset = file_size - (self.buffer.len() as u64);
+        let initial_offset = file_size
+            .checked_sub(self.buffer.len() as u64)
+            .ok_or_else(|| {
+                vortex_err!(
+                    "Footer buffer length {} exceeds declared file size {file_size}",
+                    self.buffer.len()
+                )
+            })?;
 
         let mut read_more_offset = initial_offset;
         if let Some(dtype_segment) = &dtype_segment {
@@ -373,6 +385,25 @@ mod tests {
         assert!(err.to_string().contains("out of bounds"), "{err}");
         Ok(())
     }
+
+    #[test]
+    fn deserialize_rejects_buffer_larger_than_declared_size() -> VortexResult<()> {
+        let postscript = Postscript {
+            dtype: None,
+            layout: segment(0, 1),
+            statistics: None,
+            footer: segment(1, 1),
+        };
+        let buffer = eof_buffer(&postscript)?;
+        let declared_size = buffer.len() as u64 - 1;
+
+        let mut deserializer = FooterDeserializer::new(buffer, array_session())
+            .with_dtype(DType::Primitive(PType::I32, Nullability::NonNullable))
+            .with_size(declared_size);
+        let err = deserializer.deserialize().unwrap_err();
+        assert!(err.to_string().contains("exceeds declared"), "{err}");
+        Ok(())
+    }
 }
 
 #[derive(Debug)]

vortex-file/src/footer/mod.rs

@@ -172,6 +172,11 @@ impl Footer {
         self.root_layout.row_count()
     }
 
+    /// Validate that every segment declared in the footer lies within a file of `file_size` bytes.
+    pub(crate) fn validate_file_size(&self, file_size: u64) -> VortexResult<()> {
+        validate_segments_within_file(&self.segments, file_size)
+    }
+
     /// Returns a serializer for this footer.
     pub fn into_serializer(self) -> FooterSerializer {
         FooterSerializer::new(self)
@@ -182,3 +187,60 @@ impl Footer {
         FooterDeserializer::new(eof_buffer, session)
     }
 }
+
+/// Validate that every segment declared in the footer lies within a file of `file_size` bytes.
+///
+/// A corrupt or malicious file can declare a segment whose offset or length extends past the end
+/// of the file. Rejecting such files up front ensures that later slicing of the backing buffer
+/// returns a [`VortexError`](vortex_error::VortexError) rather than panicking (see issue #8819).
+fn validate_segments_within_file(segments: &[SegmentSpec], file_size: u64) -> VortexResult<()> {
+    for segment in segments {
+        let within_file = segment
+            .offset
+            .checked_add(segment.length as u64)
+            .is_some_and(|end| end <= file_size);
+        if !within_file {
+            vortex_bail!(
+                "Segment at offset {} with length {} extends past the end of the \
+                 {file_size}-byte file",
+                segment.offset,
+                segment.length,
+            );
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use vortex_buffer::Alignment;
+
+    use super::*;
+
+    fn segment(offset: u64, length: u32) -> SegmentSpec {
+        SegmentSpec {
+            offset,
+            length,
+            alignment: Alignment::none(),
+        }
+    }
+
+    #[test]
+    fn accepts_segments_within_file() -> VortexResult<()> {
+        validate_segments_within_file(&[segment(0, 100), segment(100, 50)], 150)?;
+        Ok(())
+    }
+
+    #[test]
+    fn rejects_segment_extending_past_end_of_file() {
+        let err =
+            validate_segments_within_file(&[segment(0, 100), segment(100, 51)], 150).unwrap_err();
+        assert!(err.to_string().contains("past the end"), "{err}");
+    }
+
+    #[test]
+    fn rejects_segment_offset_length_overflow() {
+        let err = validate_segments_within_file(&[segment(u64::MAX, 1)], u64::MAX).unwrap_err();
+        assert!(err.to_string().contains("past the end"), "{err}");
+    }
+}

vortex-file/src/footer/postscript.rs

@@ -121,7 +121,9 @@ impl ReadFlatBuffer for PostscriptSegment {
         Ok(PostscriptSegment {
             offset: fb.offset(),
             length: fb.length(),
-            alignment: Alignment::from_exponent(fb.alignment_exponent()),
+            // The alignment exponent comes from the file and may be corrupt, so validate it rather
+            // than panicking on a too-large shift (see issue #8819).
+            alignment: Alignment::try_from_exponent(fb.alignment_exponent())?,
         })
     }
 }

vortex-file/src/footer/segment.rs

@@ -43,7 +43,23 @@ impl TryFrom<&fb::SegmentSpec> for SegmentSpec {
         Ok(Self {
             offset: value.offset(),
             length: value.length(),
-            alignment: Alignment::from_exponent(value.alignment_exponent()),
+            // The alignment exponent comes from the file and may be corrupt, so validate it rather
+            // than panicking on a too-large shift (see issue #8819).
+            alignment: Alignment::try_from_exponent(value.alignment_exponent())?,
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn rejects_out_of_range_alignment_exponent() {
+        // A fuzzed segment spec can declare an alignment exponent that would overflow a usize
+        // shift. Parsing it must return an error rather than panicking (see issue #8819).
+        let fb_spec = fb::SegmentSpec::new(0, 0, u8::MAX, 0, 0);
+        let err = SegmentSpec::try_from(&fb_spec).unwrap_err();
+        assert!(err.to_string().contains("too large"), "{err}");
+    }
+}