fix: fix potential OOM involving snappy data, found by fuzzer

Author: crepererum

src/protocol/record.rs

@@ -630,7 +630,8 @@ where
                 // There are "normal" compression libs, and there is Java
                 // See https://github.com/edenhill/librdkafka/blob/2b76b65212e5efda213961d5f84e565038036270/src/rdkafka_msgset_reader.c#L307-L318
                 let output = if input.starts_with(JAVA_MAGIC) {
-                    let mut cursor = Cursor::new(&input[JAVA_MAGIC.len()..]);
+                    let cursor_content = &input[JAVA_MAGIC.len()..];
+                    let mut cursor = Cursor::new(cursor_content);
 
                     let mut buf_version = [0u8; 4];
                     cursor.read_exact(&mut buf_version)?;
@@ -653,6 +654,11 @@ where
                         let mut buf_chunk_length = [0u8; 4];
                         cursor.read_exact(&mut buf_chunk_length)?;
                         let chunk_length = u32::from_be_bytes(buf_chunk_length) as usize;
+                        let bytes_left = cursor_content.len() - (cursor.position() as usize);
+                        if chunk_length > bytes_left {
+                            // do NOT try to allocate massive buffer for `chunk_data` but instead fail early
+                            return Err(ReadError::Malformed(format!("Java-specific Snappy-compressed data has illegal chunk length, got {chunk_length} bytes but only {bytes_left} bytes are left.").into()));
+                        }
 
                         let mut chunk_data = vec![0u8; chunk_length];
                         cursor.read_exact(&mut chunk_data)?;