From 311c3b51105edaffe58862004c05a2d634c9746d Mon Sep 17 00:00:00 2001 From: Ran Reichman Date: Thu, 30 Jul 2026 14:47:49 -0400 Subject: [PATCH] fix(arrow-avro): fix split sync marker assembly and validate per-block sync markers --- arrow-avro/src/reader/async_reader/mod.rs | 23 +++++++++++++++++++++ arrow-avro/src/reader/block.rs | 25 +++++++++++++++++++++-- arrow-avro/src/reader/header.rs | 5 +++-- arrow-avro/src/reader/mod.rs | 22 ++++++++++++++++++++ 4 files changed, 71 insertions(+), 4 deletions(-) diff --git a/arrow-avro/src/reader/async_reader/mod.rs b/arrow-avro/src/reader/async_reader/mod.rs index 43ff931a050d..93bdf2b4573b 100644 --- a/arrow-avro/src/reader/async_reader/mod.rs +++ b/arrow-avro/src/reader/async_reader/mod.rs @@ -386,6 +386,11 @@ impl AsyncAvroFileReader { // If we reached the end of the block, flush it, and move to read batches. if let Some(block) = self.block_decoder.flush() { // Successfully decoded a block. + if block.sync != self.sync_marker { + return self.finish_with_error(AvroError::ParseError( + "Avro block sync marker does not match file header".to_string(), + )); + } let block_count = block.count; let block_data = Bytes::from_owner(if let Some(ref codec) = self.codec { match codec.decompress(&block.data) { @@ -1083,6 +1088,24 @@ mod tests { assert!(batch.num_rows() > 0); } + #[tokio::test] + async fn test_block_sync_marker_mismatch_errors() { + use tempfile::tempdir; + let file = arrow_test_data("avro/alltypes_plain.avro"); + let mut bytes = std::fs::read(&file).unwrap(); + // The file ends with the final block's 16-byte sync marker. + let last = bytes.len() - 1; + bytes[last] ^= 0xFF; + let dir = tempdir().unwrap(); + let path = dir.path().join("corrupt_sync.avro"); + std::fs::write(&path, bytes).unwrap(); + let schema = get_alltypes_schema(); + let err = read_async_file(path.to_str().unwrap(), 1024, None, Some(schema), None) + .await + .expect_err("corrupted block sync marker should fail the read"); + assert!(err.to_string().contains("sync marker"), "{err}"); + } + #[tokio::test] async fn test_range_no_sync_marker() { // Small range unlikely to contain sync marker diff --git a/arrow-avro/src/reader/block.rs b/arrow-avro/src/reader/block.rs index 3ec6e507ea51..44161e264692 100644 --- a/arrow-avro/src/reader/block.rs +++ b/arrow-avro/src/reader/block.rs @@ -118,8 +118,10 @@ impl BlockDecoder { } BlockDecoderState::Sync => { let to_decode = buf.len().min(self.bytes_remaining); - let write = &mut self.in_progress.sync[16 - to_decode..]; - write[..to_decode].copy_from_slice(&buf[..to_decode]); + // Fill from the front: the marker may arrive split across decode() calls. + let offset = 16 - self.bytes_remaining; + self.in_progress.sync[offset..offset + to_decode] + .copy_from_slice(&buf[..to_decode]); self.bytes_remaining -= to_decode; buf = &buf[to_decode..]; if self.bytes_remaining == 0 { @@ -222,4 +224,23 @@ mod tests { assert_eq!(block.data, payload); assert_eq!(block.sync, sync); } + + #[test] + fn test_sync_marker_split_across_decode_calls() { + // count=1 (zig-zag 0x02), size=1 (0x02), one data byte, 16-byte sync marker + let sync: [u8; 16] = core::array::from_fn(|i| i as u8); + let mut block_bytes = vec![0x02, 0x02, 0xAA]; + block_bytes.extend_from_slice(&sync); + + for chunk_size in 1..block_bytes.len() { + let mut decoder = BlockDecoder::default(); + for chunk in block_bytes.chunks(chunk_size) { + decoder.decode(chunk).unwrap(); + } + let block = decoder.flush().expect("complete block"); + assert_eq!(block.count, 1); + assert_eq!(block.data, vec![0xAA]); + assert_eq!(block.sync, sync, "chunk_size {chunk_size}"); + } + } } diff --git a/arrow-avro/src/reader/header.rs b/arrow-avro/src/reader/header.rs index 235166bb7658..9c28ef64fef1 100644 --- a/arrow-avro/src/reader/header.rs +++ b/arrow-avro/src/reader/header.rs @@ -314,8 +314,9 @@ impl HeaderDecoder { } HeaderDecoderState::Sync => { let to_decode = buf.len().min(self.bytes_remaining); - let write = &mut self.sync_marker[16 - to_decode..]; - write[..to_decode].copy_from_slice(&buf[..to_decode]); + // Fill from the front: the marker may arrive split across decode() calls. + let offset = 16 - self.bytes_remaining; + self.sync_marker[offset..offset + to_decode].copy_from_slice(&buf[..to_decode]); self.bytes_remaining -= to_decode; buf = &buf[to_decode..]; if self.bytes_remaining == 0 { diff --git a/arrow-avro/src/reader/mod.rs b/arrow-avro/src/reader/mod.rs index 6fb75422a4b2..aee5ce3ac798 100644 --- a/arrow-avro/src/reader/mod.rs +++ b/arrow-avro/src/reader/mod.rs @@ -1370,6 +1370,11 @@ impl Reader { self.reader.consume(consumed); if let Some(block) = self.block_decoder.flush() { // Successfully decoded a block. + if block.sync != self.header.sync() { + return Err(AvroError::ParseError( + "Avro block sync marker does not match file header".to_string(), + )); + } self.block_data = if let Some(ref codec) = self.header.compression()? { let decompressed: Vec = codec.decompress(&block.data)?; decompressed @@ -1491,6 +1496,23 @@ mod test { arrow::compute::concat_batches(&schema, &batches).unwrap() } + #[test] + fn test_block_sync_marker_mismatch_errors() { + let path = arrow_test_data("avro/alltypes_plain.avro"); + let mut bytes = std::fs::read(&path).unwrap(); + // The file ends with the final block's 16-byte sync marker. + let last = bytes.len() - 1; + bytes[last] ^= 0xFF; + let reader = ReaderBuilder::new() + .with_batch_size(1024) + .build(std::io::Cursor::new(bytes)) + .unwrap(); + let err = reader + .collect::, _>>() + .expect_err("corrupted block sync marker should fail the read"); + assert!(err.to_string().contains("sync marker"), "{err}"); + } + fn read_file_strict( path: &str, batch_size: usize,