fix(influxdb3_catalog): guard payload_len against integer overflow

influxdb3_catalog/src/format/reader.rs

@@ -50,13 +50,20 @@ impl CatalogFile {
         let pos = cursor.position() as usize;
         let payload_len = header.payload_len as usize;
         let underlying = cursor.get_ref().as_ref();
-        let payload_end = pos + payload_len;
-        if payload_end > underlying.len() {
-            return Err(FormatError::BufferTooShort {
-                expected: payload_end,
-                actual: underlying.len(),
-            });
-        }
+        // `header.payload_len` is attacker-controlled in a crafted or corrupt
+        // file; a value near usize::MAX overflows `pos + payload_len` and wraps
+        // past the length check, so the payload slice below then panics on an
+        // out-of-range index. Treat an overflowing or oversized length as a
+        // too-short buffer, the same outcome as a plainly truncated file.
+        let payload_end = match pos.checked_add(payload_len) {
+            Some(end) if end <= underlying.len() => end,
+            _ => {
+                return Err(FormatError::BufferTooShort {
+                    expected: pos.saturating_add(payload_len),
+                    actual: underlying.len(),
+                });
+            }
+        };
         if verify_payload_crc {
             let computed_crc = crc32fast::hash(&underlying[pos..payload_end]);
             if computed_crc != header.payload_crc {

influxdb3_catalog/src/format/reader/tests.rs

@@ -299,6 +299,32 @@ fn snapshot_rejects_record_count_exceeding_payload() {
     ));
 }
 
+#[test]
+fn rejects_payload_len_overflowing_offset() {
+    // A crafted or corrupt header can claim a payload_len near u64::MAX. The
+    // bounds check must reject it cleanly: `pos + payload_len` would otherwise
+    // overflow and wrap past the check, panicking when the payload is sliced.
+    let header = Header {
+        format_version: FORMAT_VERSION,
+        flags: 0,
+        catalog_uuid: 1337,
+        sequence_number: 1,
+        record_count: 0,
+        group_count: 0,
+        payload_crc: crc32fast::hash(&[]),
+        payload_len: u64::MAX,
+    };
+    let mut file = Vec::new();
+    file.extend_from_slice(&header.to_bytes());
+    file.extend_from_slice(&[0u8; 8]);
+    let mut cursor = Cursor::new(Bytes::from(file));
+
+    assert!(matches!(
+        CatalogFile::read_from(&mut cursor),
+        Err(FormatError::BufferTooShort { .. }),
+    ));
+}
+
 #[cfg(test)]
 mod skip_crc_tests {
     use super::*;