feat: extend fuzzing, fix one found issue

fuzz/fuzz_targets/protocol_reader.rs

@@ -18,10 +18,13 @@ use rskafka::{
         api_version::{ApiVersion, ApiVersionRange},
         frame::AsyncMessageWrite,
         messages::{
-            ApiVersionsRequest, CreateTopicsRequest, FetchRequest, ListOffsetsRequest,
-            MetadataRequest, ProduceRequest, ReadVersionedType, RequestBody, WriteVersionedType,
+            ApiVersionsRequest, CreateTopicsRequest, DeleteRecordsRequest, DeleteTopicsRequest,
+            FetchRequest, ListOffsetsRequest, MetadataRequest, ProduceRequest, ReadVersionedType,
+            RequestBody, SaslAuthenticateRequest, SaslHandshakeRequest, WriteVersionedType,
+        },
+        primitives::{
+            Array, Bytes, CompactString, Int16, Int32, NullableString, String_, TaggedFields,
         },
-        primitives::{CompactString, Int16, Int32, NullableString, TaggedFields},
         traits::ReadType,
     },
 };
@@ -61,6 +64,26 @@ fn driver(data: &[u8]) -> Result<(), Error> {
             api_key,
             api_version,
         ),
+        ApiKey::DeleteRecords => send_recv(
+            DeleteRecordsRequest {
+                topics: vec![],
+                timeout_ms: Int32(0),
+                tagged_fields: None,
+            },
+            cursor,
+            api_key,
+            api_version,
+        ),
+        ApiKey::DeleteTopics => send_recv(
+            DeleteTopicsRequest {
+                topic_names: Array(None),
+                timeout_ms: Int32(0),
+                tagged_fields: None,
+            },
+            cursor,
+            api_key,
+            api_version,
+        ),
         ApiKey::Fetch => send_recv(
             FetchRequest {
                 replica_id: Int32(0),
@@ -104,6 +127,23 @@ fn driver(data: &[u8]) -> Result<(), Error> {
             api_key,
             api_version,
         ),
+        ApiKey::SaslAuthenticate => send_recv(
+            SaslAuthenticateRequest {
+                auth_bytes: Bytes(vec![]),
+                tagged_fields: None,
+            },
+            cursor,
+            api_key,
+            api_version,
+        ),
+        ApiKey::SaslHandshake => send_recv(
+            SaslHandshakeRequest {
+                mechanism: String_(String::new()),
+            },
+            cursor,
+            api_key,
+            api_version,
+        ),
         _ => Err(format!("Fuzzing not implemented for: {api_key:?}").into()),
     }
 }

src/protocol/primitives.rs

@@ -515,14 +515,17 @@ where
 #[derive(Debug, Eq, PartialEq)]
 #[cfg_attr(test, derive(proptest_derive::Arbitrary))]
 pub struct CompactBytes(pub Vec<u8>);
+
 impl<R> ReadType<R> for CompactBytes
 where
     R: Read,
 {
     fn read(reader: &mut R) -> Result<Self, ReadError> {
         let len = UnsignedVarint::read(reader)?;
         let len = usize::try_from(len.0)?;
-        let len = len - 1;
+        let len = len.checked_sub(1).ok_or_else(|| {
+            ReadError::Malformed(format!("Must be able to substract 1 from len ({len})").into())
+        })?;
         let mut buf = VecBuilder::new(len);
         buf = buf.read_exact(reader)?;
         Ok(Self(buf.into()))
@@ -1028,6 +1031,20 @@ mod tests {
 
     test_roundtrip!(CompactBytes, test_compact_bytes_roundtrip);
 
+    #[test]
+    fn test_compact_read_negative_length() {
+        let mut buf = Cursor::new(Vec::<u8>::new());
+        UnsignedVarint(0).write(&mut buf).unwrap();
+        buf.set_position(0);
+
+        let err = CompactBytes::read(&mut buf).unwrap_err();
+        assert_matches!(err, ReadError::Malformed(_));
+        assert_eq!(
+            err.to_string(),
+            "Malformed data: Must be able to substract 1 from len (0)",
+        );
+    }
+
     test_roundtrip!(NullableBytes, test_nullable_bytes_roundtrip);
 
     #[test]