fix: Remove unchecked &buf[offset..] etc. in btree read path #4736

Replace panicking unchecked buffer indexing with bounds-checked access
throughout the btree read path. Corrupt cell pointers or varint data
now return LimboError::Corrupt instead of causing index-out-of-bounds
panics.

Changes:
- pager.rs: Add bounds checks in cell_table_interior_read_rowid,
  cell_table_leaf_read_rowid, cell_index_read_payload_ptr, and
  _cell_get_raw_region_faster (now returns Result)
- sqlite3_ondisk.rs: Add bounds checks in read_btree_cell for all
  page types; make read_payload return Result with cell_len < 4 check
- btree.rs: Update callers of _cell_get_raw_region_faster for Result
- integrity_check.rs: Add tests for corrupt cell pointers on leaf and
  interior pages, and for SELECT on corrupt pages; update existing
  cell overflow test to assert no panic

Closes #4736

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: 2 people

core/error.rs

@@ -180,6 +180,35 @@ macro_rules! bail_corrupt_error {
     };
 }
 
+/// Bounds-checked buffer slicing that returns `LimboError::Corrupt` on out-of-bounds.
+///
+/// Accepts any range expression: `buf, pos..`, `buf, start..end`, etc.
+#[macro_export]
+macro_rules! slice_in_bounds_or_corrupt {
+    ($buf:expr, $range:expr) => {
+        $buf.get($range).ok_or_else(|| {
+            $crate::error::cold_return($crate::error::LimboError::Corrupt(format!(
+                "range {:?} out of bounds for buffer size {}",
+                $range,
+                $buf.len()
+            )))
+        })?
+    };
+}
+
+/// Asserts a condition or bails with `LimboError::Corrupt`.
+///
+/// Usage:
+///   `assert_or_bail_corrupt!(condition, "message {}", arg)`
+#[macro_export]
+macro_rules! assert_or_bail_corrupt {
+    ($cond:expr, $($arg:tt)*) => {
+        if !($cond) {
+            $crate::bail_corrupt_error!($($arg)*);
+        }
+    };
+}
+
 #[macro_export]
 macro_rules! bail_constraint_error {
     ($($arg:tt)*) => {

core/storage/btree.rs

@@ -2957,7 +2957,7 @@ impl BTreeCursor {
                                     parent_max_local,
                                     parent_min_local,
                                     parent_page_type,
-                                );
+                                )?;
                             let buf = parent_contents.as_ptr();
                             &buf[cell_start..cell_start + cell_len]
                         };
@@ -3046,7 +3046,7 @@ impl BTreeCursor {
                                     max_local,
                                     min_local,
                                     page_type,
-                                );
+                                )?;
                             let buf = old_page_contents.as_ptr();
                             let cell_buf = &mut buf[cell_start..cell_start + cell_len];
                             // TODO(pere): make this reference and not copy
@@ -7691,7 +7691,7 @@ fn defragment_page(page: &PageContent, usable_space: usize, max_frag_bytes: isiz
             max_local,
             min_local,
             page_type,
-        );
+        )?;
 
         if pc > last_offset {
             is_physically_sorted = false;

core/storage/pager.rs

@@ -397,7 +397,10 @@ impl PageInner {
         let cell_pointer = cell_pointer_array_start + (idx * CELL_PTR_SIZE_BYTES);
         let cell_pointer = self.read_u16(cell_pointer) as usize;
         const LEFT_CHILD_PAGE_SIZE_BYTES: usize = 4;
-        let (rowid, _) = read_varint(&buf[cell_pointer + LEFT_CHILD_PAGE_SIZE_BYTES..])?;
+        let (rowid, _) = read_varint(crate::slice_in_bounds_or_corrupt!(
+            buf,
+            cell_pointer + LEFT_CHILD_PAGE_SIZE_BYTES..
+        ))?;
         Ok(rowid as i64)
     }
 
@@ -411,14 +414,12 @@ impl PageInner {
         let cell_pointer_array_start = self.header_size();
         let cell_pointer = cell_pointer_array_start + (idx * CELL_PTR_SIZE_BYTES);
         let cell_pointer = self.read_u16(cell_pointer) as usize;
-        // Validate cell pointer is within page bounds (need 4 bytes for u32)
-        if cell_pointer + 4 > buf.len() {
-            return Err(crate::LimboError::Corrupt(format!(
-                "cell pointer {} out of bounds for page size {}",
-                cell_pointer,
-                buf.len()
-            )));
-        }
+        crate::assert_or_bail_corrupt!(
+            cell_pointer + 4 <= buf.len(),
+            "cell pointer {} out of bounds for page size {}",
+            cell_pointer,
+            buf.len()
+        );
         Ok(u32::from_be_bytes([
             buf[cell_pointer],
             buf[cell_pointer + 1],
@@ -435,9 +436,9 @@ impl PageInner {
         let cell_pointer = cell_pointer_array_start + (idx * CELL_PTR_SIZE_BYTES);
         let cell_pointer = self.read_u16(cell_pointer) as usize;
         let mut pos = cell_pointer;
-        let (_, nr) = read_varint(&buf[pos..])?;
+        let (_, nr) = read_varint(crate::slice_in_bounds_or_corrupt!(buf, pos..))?;
         pos += nr;
-        let (rowid, _) = read_varint(&buf[pos..])?;
+        let (rowid, _) = read_varint(crate::slice_in_bounds_or_corrupt!(buf, pos..))?;
         Ok(rowid as i64)
     }
 
@@ -461,11 +462,13 @@ impl PageInner {
         let page_type = self.page_type()?;
         let (payload_size, varint_len, header_skip) = match page_type {
             PageType::IndexInterior => {
-                let (size, len) = read_varint(&buf[cell_offset + 4..])?;
+                let (size, len) =
+                    read_varint(crate::slice_in_bounds_or_corrupt!(buf, cell_offset + 4..))?;
                 (size, len, 4usize)
             }
             PageType::IndexLeaf => {
-                let (size, len) = read_varint(&buf[cell_offset..])?;
+                let (size, len) =
+                    read_varint(crate::slice_in_bounds_or_corrupt!(buf, cell_offset..))?;
                 (size, len, 0usize)
             }
             _ => unreachable!("cell_index_read_payload_ptr called on non-index page"),
@@ -484,25 +487,43 @@ impl PageInner {
 
         let (payload_slice, first_overflow) = if overflows {
             let overflow_ptr_offset = payload_start + local_size - 4;
+            crate::assert_or_bail_corrupt!(
+                overflow_ptr_offset + 4 <= buf.len(),
+                "overflow pointer offset {} out of bounds for page size {}",
+                overflow_ptr_offset,
+                buf.len()
+            );
             let first_overflow_page = u32::from_be_bytes([
                 buf[overflow_ptr_offset],
                 buf[overflow_ptr_offset + 1],
                 buf[overflow_ptr_offset + 2],
                 buf[overflow_ptr_offset + 3],
             ]);
+            let payload_end = payload_start + local_size - 4;
+            crate::assert_or_bail_corrupt!(
+                payload_start < payload_end && payload_end <= buf.len(),
+                "payload range {}..{} out of bounds for page size {}",
+                payload_start,
+                payload_end,
+                buf.len()
+            );
             // SAFETY: valid as long as page is alive
             let slice = unsafe {
-                std::mem::transmute::<&[u8], &'static [u8]>(
-                    &buf[payload_start..payload_start + local_size - 4],
-                )
+                std::mem::transmute::<&[u8], &'static [u8]>(&buf[payload_start..payload_end])
             };
             (slice, Some(first_overflow_page))
         } else {
+            let payload_end = payload_start + payload_size as usize;
+            crate::assert_or_bail_corrupt!(
+                payload_end <= buf.len(),
+                "payload range {}..{} out of bounds for page size {}",
+                payload_start,
+                payload_end,
+                buf.len()
+            );
             // SAFETY: valid as long as page is alive
             let slice = unsafe {
-                std::mem::transmute::<&[u8], &'static [u8]>(
-                    &buf[payload_start..payload_start + payload_size as usize],
-                )
+                std::mem::transmute::<&[u8], &'static [u8]>(&buf[payload_start..payload_end])
             };
             (slice, None)
         };
@@ -540,14 +561,14 @@ impl PageInner {
         let max_local = payload_overflow_threshold_max(page_type, usable_size);
         let min_local = payload_overflow_threshold_min(page_type, usable_size);
         let cell_count = self.cell_count();
-        Ok(self._cell_get_raw_region_faster(
+        self._cell_get_raw_region_faster(
             idx,
             usable_size,
             cell_count,
             max_local,
             min_local,
             page_type,
-        ))
+        )
     }
 
     #[inline]
@@ -559,13 +580,14 @@ impl PageInner {
         max_local: usize,
         min_local: usize,
         page_type: PageType,
-    ) -> (usize, usize) {
+    ) -> crate::Result<(usize, usize)> {
         let buf = self.as_ptr();
         turso_assert_less_than!(idx, cell_count);
         let start = self.cell_get_raw_start_offset(idx);
         let len = match page_type {
             PageType::IndexInterior => {
-                let (len_payload, n_payload) = read_varint(&buf[start + 4..]).unwrap();
+                let (len_payload, n_payload) =
+                    read_varint(crate::slice_in_bounds_or_corrupt!(buf, start + 4..))?;
                 let (overflows, to_read) = sqlite3_ondisk::payload_overflows(
                     len_payload as usize,
                     max_local,
@@ -579,11 +601,13 @@ impl PageInner {
                 }
             }
             PageType::TableInterior => {
-                let (_, n_rowid) = read_varint(&buf[start + 4..]).unwrap();
+                let (_, n_rowid) =
+                    read_varint(crate::slice_in_bounds_or_corrupt!(buf, start + 4..))?;
                 4 + n_rowid
             }
             PageType::IndexLeaf => {
-                let (len_payload, n_payload) = read_varint(&buf[start..]).unwrap();
+                let (len_payload, n_payload) =
+                    read_varint(crate::slice_in_bounds_or_corrupt!(buf, start..))?;
                 let (overflows, to_read) = sqlite3_ondisk::payload_overflows(
                     len_payload as usize,
                     max_local,
@@ -601,8 +625,10 @@ impl PageInner {
                 }
             }
             PageType::TableLeaf => {
-                let (len_payload, n_payload) = read_varint(&buf[start..]).unwrap();
-                let (_, n_rowid) = read_varint(&buf[start + n_payload..]).unwrap();
+                let (len_payload, n_payload) =
+                    read_varint(crate::slice_in_bounds_or_corrupt!(buf, start..))?;
+                let (_, n_rowid) =
+                    read_varint(crate::slice_in_bounds_or_corrupt!(buf, start + n_payload..))?;
                 let (overflows, to_read) = sqlite3_ondisk::payload_overflows(
                     len_payload as usize,
                     max_local,
@@ -620,7 +646,14 @@ impl PageInner {
                 }
             }
         };
-        (start, len)
+        crate::assert_or_bail_corrupt!(
+            start + len <= buf.len(),
+            "cell region {}..{} out of bounds for page size {}",
+            start,
+            start + len,
+            buf.len()
+        );
+        Ok((start, len))
     }
 
     pub fn is_leaf(&self) -> bool {

core/storage/sqlite3_ondisk.rs

@@ -805,18 +805,31 @@ pub fn read_btree_cell(
     match page_type {
         PageType::IndexInterior => {
             let mut pos = pos;
+            crate::assert_or_bail_corrupt!(
+                pos + 4 <= page.len(),
+                "cell offset {} out of bounds for page size {}",
+                pos,
+                page.len()
+            );
             let left_child_page =
                 u32::from_be_bytes([page[pos], page[pos + 1], page[pos + 2], page[pos + 3]]);
             pos += 4;
-            let (payload_size, nr) = read_varint(&page[pos..])?;
+            let (payload_size, nr) = read_varint(crate::slice_in_bounds_or_corrupt!(page, pos..))?;
             pos += nr;
 
             let (overflows, to_read) =
                 payload_overflows(payload_size as usize, max_local, min_local, usable_size);
             let to_read = if overflows { to_read } else { page.len() - pos };
 
+            crate::assert_or_bail_corrupt!(
+                pos + to_read <= page.len(),
+                "payload range {}..{} out of bounds for page size {}",
+                pos,
+                pos + to_read,
+                page.len()
+            );
             let (payload, first_overflow_page) =
-                read_payload(&page[pos..pos + to_read], payload_size as usize);
+                read_payload(&page[pos..pos + to_read], payload_size as usize)?;
             Ok(BTreeCell::IndexInteriorCell(IndexInteriorCell {
                 left_child_page,
                 payload,
@@ -826,26 +839,39 @@ pub fn read_btree_cell(
         }
         PageType::TableInterior => {
             let mut pos = pos;
+            crate::assert_or_bail_corrupt!(
+                pos + 4 <= page.len(),
+                "cell offset {} out of bounds for page size {}",
+                pos,
+                page.len()
+            );
             let left_child_page =
                 u32::from_be_bytes([page[pos], page[pos + 1], page[pos + 2], page[pos + 3]]);
             pos += 4;
-            let (rowid, _) = read_varint(&page[pos..])?;
+            let (rowid, _) = read_varint(crate::slice_in_bounds_or_corrupt!(page, pos..))?;
             Ok(BTreeCell::TableInteriorCell(TableInteriorCell {
                 left_child_page,
                 rowid: rowid as i64,
             }))
         }
         PageType::IndexLeaf => {
             let mut pos = pos;
-            let (payload_size, nr) = read_varint(&page[pos..])?;
+            let (payload_size, nr) = read_varint(crate::slice_in_bounds_or_corrupt!(page, pos..))?;
             pos += nr;
 
             let (overflows, to_read) =
                 payload_overflows(payload_size as usize, max_local, min_local, usable_size);
             let to_read = if overflows { to_read } else { page.len() - pos };
 
+            crate::assert_or_bail_corrupt!(
+                pos + to_read <= page.len(),
+                "payload range {}..{} out of bounds for page size {}",
+                pos,
+                pos + to_read,
+                page.len()
+            );
             let (payload, first_overflow_page) =
-                read_payload(&page[pos..pos + to_read], payload_size as usize);
+                read_payload(&page[pos..pos + to_read], payload_size as usize)?;
             Ok(BTreeCell::IndexLeafCell(IndexLeafCell {
                 payload,
                 first_overflow_page,
@@ -854,17 +880,24 @@ pub fn read_btree_cell(
         }
         PageType::TableLeaf => {
             let mut pos = pos;
-            let (payload_size, nr) = read_varint(&page[pos..])?;
+            let (payload_size, nr) = read_varint(crate::slice_in_bounds_or_corrupt!(page, pos..))?;
             pos += nr;
-            let (rowid, nr) = read_varint(&page[pos..])?;
+            let (rowid, nr) = read_varint(crate::slice_in_bounds_or_corrupt!(page, pos..))?;
             pos += nr;
 
             let (overflows, to_read) =
                 payload_overflows(payload_size as usize, max_local, min_local, usable_size);
             let to_read = if overflows { to_read } else { page.len() - pos };
 
+            crate::assert_or_bail_corrupt!(
+                pos + to_read <= page.len(),
+                "payload range {}..{} out of bounds for page size {}",
+                pos,
+                pos + to_read,
+                page.len()
+            );
             let (payload, first_overflow_page) =
-                read_payload(&page[pos..pos + to_read], payload_size as usize);
+                read_payload(&page[pos..pos + to_read], payload_size as usize)?;
             Ok(BTreeCell::TableLeafCell(TableLeafCell {
                 rowid: rowid as i64,
                 payload,
@@ -878,21 +911,30 @@ pub fn read_btree_cell(
 /// read_payload takes in the unread bytearray with the payload size
 /// and returns the payload on the page, and optionally the first overflow page number.
 #[allow(clippy::readonly_write_lock)]
-fn read_payload(unread: &'static [u8], payload_size: usize) -> (&'static [u8], Option<u32>) {
+fn read_payload(
+    unread: &'static [u8],
+    payload_size: usize,
+) -> Result<(&'static [u8], Option<u32>)> {
     let cell_len = unread.len();
     // We will let overflow be constructed back if needed or requested.
     if payload_size <= cell_len {
         // fit within 1 page
-        (&unread[..payload_size], None)
+        Ok((&unread[..payload_size], None))
     } else {
         // overflow
+        if cell_len < 4 {
+            bail_corrupt_error!(
+                "overflow cell too small: {} bytes, need at least 4",
+                cell_len
+            );
+        }
         let first_overflow_page = u32::from_be_bytes([
             unread[cell_len - 4],
             unread[cell_len - 3],
             unread[cell_len - 2],
             unread[cell_len - 1],
         ]);
-        (&unread[..cell_len - 4], Some(first_overflow_page))
+        Ok((&unread[..cell_len - 4], Some(first_overflow_page)))
     }
 }