From 5f5456d4334e0f327a8dbac75c20903573fc6f58 Mon Sep 17 00:00:00 2001 From: masumi ryugo <280057467+masumi-ryugo@users.noreply.github.com> Date: Mon, 4 May 2026 00:34:10 +0900 Subject: [PATCH 1/2] fix(arrow-csv): bound RecordDecoder::flush offset accumulation Closes #9885. `RecordDecoder::flush` walks the per-row offsets emitted by `csv_core::Reader` and accumulates them so each end offset is absolute over `self.data` after the loop. The accumulator was a plain `usize` and the loop body did `*x += offset`, which on malformed input that drives `csv_core` to emit row-relative offsets large enough to wrap a `usize` panics with `attempt to add with overflow` in debug builds and silently wraps in release builds. The cargo-fuzz `csv_reader` harness being prototyped for #5332 reproduces this in single-digit minutes from an empty corpus with a 72-byte input. After this patch the same input returns `ArrowError::CsvError("CSV record offsets overflowed usize while flushing")` instead of panicking. Includes a regression test in `reader::records::tests` driving the 72-byte fuzz repro through `RecordDecoder::decode` + `flush`. Co-Authored-By: Claude Opus 4.7 (1M context) --- arrow-csv/src/reader/records.rs | 71 +++++++++++++++++++++++++++------ 1 file changed, 59 insertions(+), 12 deletions(-) diff --git a/arrow-csv/src/reader/records.rs b/arrow-csv/src/reader/records.rs index 33927c93360a..8b2dd3da88d6 100644 --- a/arrow-csv/src/reader/records.rs +++ b/arrow-csv/src/reader/records.rs @@ -196,18 +196,27 @@ impl RecordDecoder { )); } - // csv_core::Reader writes end offsets relative to the start of the row - // Therefore scan through and offset these based on the cumulative row offsets - let mut row_offset = 0; - self.offsets[1..self.offsets_len] - .chunks_exact_mut(self.num_columns) - .for_each(|row| { - let offset = row_offset; - row.iter_mut().for_each(|x| { - *x += offset; - row_offset = *x; - }); - }); + // csv_core::Reader writes end offsets relative to the start of the row. + // Scan through and offset these by the cumulative row offsets so that + // each end offset is absolute over `self.data` after this loop. + // + // Use `checked_add` on the accumulator: malformed input that drives + // `csv_core` to emit row-relative offsets large enough to wrap a + // `usize` would otherwise hit `attempt to add with overflow` in debug + // builds and silently wrap to a wildly out-of-bounds index in release + // builds. Surface the overflow as a `CsvError` instead. + let mut row_offset: usize = 0; + for row in self.offsets[1..self.offsets_len].chunks_exact_mut(self.num_columns) { + let offset = row_offset; + for x in row.iter_mut() { + *x = x.checked_add(offset).ok_or_else(|| { + ArrowError::CsvError( + "CSV record offsets overflowed usize while flushing".to_string(), + ) + })?; + row_offset = *x; + } + } // Need to truncate data t1o the actual amount of data read let data = std::str::from_utf8(&self.data[..self.data_len]).map_err(|e| { @@ -399,4 +408,42 @@ mod tests { assert_eq!(read, 5); assert_eq!(bytes, csv.len()); } + + /// Regression test for a `cargo-fuzz` repro from the `arrow-csv` harness + /// being prototyped for #5332 / apache/arrow-rs#9885: 72 bytes that drove + /// `*x += offset` in `RecordDecoder::flush` past `usize::MAX`. After this + /// patch the overflow is surfaced as `ArrowError::CsvError` instead of + /// panicking. + #[test] + fn test_flush_offset_overflow_does_not_panic() { + let bytes: [u8; 72] = [ + 0x2e, 0x22, 0x3f, 0x0a, 0x31, 0x0a, 0x3f, 0x3f, 0x0a, 0x3c, 0x50, 0x50, 0x0a, 0x3f, + 0x0a, 0x31, 0x0a, 0x3f, 0x38, 0x0a, 0x3c, 0x0a, 0x3f, 0x0a, 0x3c, 0x50, 0x50, 0x0a, + 0x3f, 0x0a, 0x31, 0x0a, 0x3f, 0x38, 0x0a, 0x0a, 0x2e, 0x22, 0x3f, 0x0a, 0x31, 0x0a, + 0x3f, 0x3f, 0x0a, 0xce, 0xce, 0xce, 0xb1, 0xce, 0xce, 0xce, 0xce, 0xce, 0xce, 0xce, + 0xce, 0x31, 0x0a, 0x3f, 0x38, 0x0a, 0x3c, 0x0a, 0x3f, 0x0a, 0x3c, 0x0a, 0x3f, 0x0a, + 0x3f, 0x69, + ]; + // The fuzz harness uses 1 column / no header, which is what + // `Format::default().with_header(false)` infers from this input. + let mut decoder = RecordDecoder::new(Reader::new(), 1, false); + let mut input = &bytes[..]; + loop { + let (_read, consumed) = match decoder.decode(input, 1024) { + Ok(r) => r, + Err(_) => break, // Either parse error or our new overflow guard. + }; + if consumed == 0 { + break; + } + input = &input[consumed..]; + // The buggy version panics inside `flush()`; the patched version + // either returns rows or surfaces a clean `CsvError`. + if decoder.flush().is_err() { + break; + } + decoder.clear(); + } + // Reaching this assertion at all means we did not panic. + } } From 3f7d92a9f70fbb47f32fb6aacdd8858033af66bd Mon Sep 17 00:00:00 2001 From: masumi ryugo <280057467+masumi-ryugo@users.noreply.github.com> Date: Wed, 6 May 2026 16:43:02 +0900 Subject: [PATCH 2/2] Address review feedback from alamb on PR #9886 - Restore the original two-line "what" comment on the offset rebase loop and drop the verbose "add with overflow / debug vs release" exposition; the code is self-explanatory. - Switch back to the existing `try_for_each` chain pattern instead of nested `for` loops, so the structure matches the rest of the file. - Replace the regression test's loop-and-discard structure with a white-box construction that stages the overflow directly. The test now uses the standard `flush().unwrap_err()` + `assert_eq` pattern (matching `test_invalid_fields`) and asserts the specific `Csv error: CSV record offsets overflowed usize while flushing` message rather than just "did not panic". --- arrow-csv/src/reader/records.rs | 85 ++++++++++++--------------------- 1 file changed, 31 insertions(+), 54 deletions(-) diff --git a/arrow-csv/src/reader/records.rs b/arrow-csv/src/reader/records.rs index 8b2dd3da88d6..a6d54f867c1d 100644 --- a/arrow-csv/src/reader/records.rs +++ b/arrow-csv/src/reader/records.rs @@ -196,27 +196,23 @@ impl RecordDecoder { )); } - // csv_core::Reader writes end offsets relative to the start of the row. - // Scan through and offset these by the cumulative row offsets so that - // each end offset is absolute over `self.data` after this loop. - // - // Use `checked_add` on the accumulator: malformed input that drives - // `csv_core` to emit row-relative offsets large enough to wrap a - // `usize` would otherwise hit `attempt to add with overflow` in debug - // builds and silently wrap to a wildly out-of-bounds index in release - // builds. Surface the overflow as a `CsvError` instead. + // csv_core::Reader writes end offsets relative to the start of the row + // Therefore scan through and offset these based on the cumulative row offsets let mut row_offset: usize = 0; - for row in self.offsets[1..self.offsets_len].chunks_exact_mut(self.num_columns) { - let offset = row_offset; - for x in row.iter_mut() { - *x = x.checked_add(offset).ok_or_else(|| { - ArrowError::CsvError( - "CSV record offsets overflowed usize while flushing".to_string(), - ) - })?; - row_offset = *x; - } - } + self.offsets[1..self.offsets_len] + .chunks_exact_mut(self.num_columns) + .try_for_each(|row| -> Result<(), ArrowError> { + let offset = row_offset; + row.iter_mut().try_for_each(|x| -> Result<(), ArrowError> { + *x = x.checked_add(offset).ok_or_else(|| { + ArrowError::CsvError( + "CSV record offsets overflowed usize while flushing".to_string(), + ) + })?; + row_offset = *x; + Ok(()) + }) + })?; // Need to truncate data t1o the actual amount of data read let data = std::str::from_utf8(&self.data[..self.data_len]).map_err(|e| { @@ -409,41 +405,22 @@ mod tests { assert_eq!(bytes, csv.len()); } - /// Regression test for a `cargo-fuzz` repro from the `arrow-csv` harness - /// being prototyped for #5332 / apache/arrow-rs#9885: 72 bytes that drove - /// `*x += offset` in `RecordDecoder::flush` past `usize::MAX`. After this - /// patch the overflow is surfaced as `ArrowError::CsvError` instead of - /// panicking. + /// Regression test for an overflow path found by the `arrow-csv` + /// cargo-fuzz harness being prototyped for #5332. Stages the + /// `RecordDecoder` state directly so that rebasing the second row's + /// end offset overflows `usize`. With the previous `*x += offset` this + /// panicked with `attempt to add with overflow`; the patched code + /// surfaces the condition as `ArrowError::CsvError`. #[test] - fn test_flush_offset_overflow_does_not_panic() { - let bytes: [u8; 72] = [ - 0x2e, 0x22, 0x3f, 0x0a, 0x31, 0x0a, 0x3f, 0x3f, 0x0a, 0x3c, 0x50, 0x50, 0x0a, 0x3f, - 0x0a, 0x31, 0x0a, 0x3f, 0x38, 0x0a, 0x3c, 0x0a, 0x3f, 0x0a, 0x3c, 0x50, 0x50, 0x0a, - 0x3f, 0x0a, 0x31, 0x0a, 0x3f, 0x38, 0x0a, 0x0a, 0x2e, 0x22, 0x3f, 0x0a, 0x31, 0x0a, - 0x3f, 0x3f, 0x0a, 0xce, 0xce, 0xce, 0xb1, 0xce, 0xce, 0xce, 0xce, 0xce, 0xce, 0xce, - 0xce, 0x31, 0x0a, 0x3f, 0x38, 0x0a, 0x3c, 0x0a, 0x3f, 0x0a, 0x3c, 0x0a, 0x3f, 0x0a, - 0x3f, 0x69, - ]; - // The fuzz harness uses 1 column / no header, which is what - // `Format::default().with_header(false)` infers from this input. + fn test_flush_offset_overflow_returns_csv_error() { let mut decoder = RecordDecoder::new(Reader::new(), 1, false); - let mut input = &bytes[..]; - loop { - let (_read, consumed) = match decoder.decode(input, 1024) { - Ok(r) => r, - Err(_) => break, // Either parse error or our new overflow guard. - }; - if consumed == 0 { - break; - } - input = &input[consumed..]; - // The buggy version panics inside `flush()`; the patched version - // either returns rows or surfaces a clean `CsvError`. - if decoder.flush().is_err() { - break; - } - decoder.clear(); - } - // Reaching this assertion at all means we did not panic. + decoder.offsets = vec![0, usize::MAX, 1]; + decoder.offsets_len = 3; + decoder.num_rows = 2; + let err = decoder.flush().unwrap_err(); + assert_eq!( + err.to_string(), + "Csv error: CSV record offsets overflowed usize while flushing" + ); } }