fix: Validate metadata len in IPC reader

arrow-ipc/src/reader.rs

@@ -1402,7 +1402,9 @@ impl<R: Read> StreamReader<R> {
             i32::from_le_bytes(meta_size)
         };
 
-        let mut meta_buffer = vec![0; meta_len as usize];
+        let meta_len = usize::try_from(meta_len)
+            .map_err(|_| ArrowError::ParseError(format!("Invalid metadata length: {meta_len}")))?;
+        let mut meta_buffer = vec![0; meta_len];
         reader.read_exact(&mut meta_buffer)?;
 
         let message = crate::root_as_message(meta_buffer.as_slice()).map_err(|err| {
@@ -1484,13 +1486,16 @@ impl<R: Read> StreamReader<R> {
             i32::from_le_bytes(meta_size)
         };
 
+        let meta_len = usize::try_from(meta_len)
+            .map_err(|_| ArrowError::ParseError(format!("Invalid metadata length: {meta_len}")))?;
+
         if meta_len == 0 {
             // the stream has ended, mark the reader as finished
             self.finished = true;
             return Ok(None);
         }
 
-        let mut meta_buffer = vec![0; meta_len as usize];
+        let mut meta_buffer = vec![0; meta_len];
         self.reader.read_exact(&mut meta_buffer)?;
 
         let vecs = &meta_buffer.to_vec();
@@ -1594,6 +1599,8 @@ impl<R: Read> RecordBatchReader for StreamReader<R> {
 
 #[cfg(test)]
 mod tests {
+    use std::io::Cursor;
+
     use crate::convert::fb_to_schema;
     use crate::writer::{
         unslice_run_array, write_message, DictionaryTracker, IpcDataGenerator, IpcWriteOptions,
@@ -1740,6 +1747,49 @@ mod tests {
         .unwrap()
     }
 
+    #[test]
+    fn test_negative_meta_len_start_stream() {
+        let bytes = i32::to_le_bytes(-1);
+        let mut buf = vec![];
+        buf.extend(CONTINUATION_MARKER);
+        buf.extend(bytes);
+
+        let reader_err = StreamReader::try_new(Cursor::new(buf), None).err();
+        assert!(reader_err.is_some());
+        assert_eq!(
+            reader_err.unwrap().to_string(),
+            "Parser error: Invalid metadata length: -1"
+        );
+    }
+
+    #[test]
+    fn test_negative_meta_len_mid_stream() {
+        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+        let mut buf = Vec::new();
+        {
+            let mut writer = crate::writer::StreamWriter::try_new(&mut buf, &schema).unwrap();
+            let batch =
+                RecordBatch::try_new(Arc::new(schema), vec![Arc::new(Int32Array::from(vec![1]))])
+                    .unwrap();
+            writer.write(&batch).unwrap();
+        }
+
+        let bytes = i32::to_le_bytes(-1);
+        buf.extend(CONTINUATION_MARKER);
+        buf.extend(bytes);
+
+        let mut reader = StreamReader::try_new(Cursor::new(buf), None).unwrap();
+        // Read the valid value
+        assert!(reader.maybe_next().is_ok());
+        // Read the invalid meta len
+        let batch_err = reader.maybe_next().err();
+        assert!(batch_err.is_some());
+        assert_eq!(
+            batch_err.unwrap().to_string(),
+            "Parser error: Invalid metadata length: -1"
+        );
+    }
+
     #[test]
     fn test_projection_array_values() {
         // define schema