Add fuzz test to handle length math issues with Messages and Headers

.evergreen/config.yml

@@ -66,6 +66,13 @@ buildvariants:
     tasks:
       - name: .lint
 
+  - name: fuzzer
+    display_name: "Fuzzer"
+    run_on:
+      - ubuntu2204-large
+    tasks:
+      "run-fuzzer"
+
   - name: rhel-8
     display_name: "RHEL 8"
     run_on:
@@ -777,6 +784,10 @@ tasks:
     commands:
       - func: "check rustdoc"
 
+  - name: "run-fuzzer"
+    commands:
+      - func: "run fuzzer"
+
   # Driver test suite runs for the full set of versions and topologies are in
   # suite-tasks.yml, generated by .evergreen/generate_tasks.
 
@@ -1264,6 +1275,17 @@ functions:
           - AWS_SECRET_ACCESS_KEY
           - AWS_SESSION_TOKEN
 
+  "run fuzzer":
+    - command: shell.exec
+      type: test
+      params:
+        shell: bash
+        working_dir: src
+        script: |
+          ${PREPARE_SHELL}
+          .evergreen/install-fuzzer.sh
+          .evergreen/run-fuzzer.sh
+
   "run aws auth test with regular aws credentials":
     - command: subprocess.exec
       type: test

.evergreen/install-fuzzer.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+set -o errexit
+
+source ./.evergreen/env.sh
+
+cargo install cargo-fuzz

.evergreen/run-fuzzer.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+
+set -o errexit
+
+source ./.evergreen/env.sh
+
+mkdir -p artifacts
+
+# Function to run fuzzer and collect crashes
+run_fuzzer() {
+    target=$1
+    echo "Running fuzzer for $target"
+    # Run fuzzer and redirect crashes to artifacts directory
+    RUST_BACKTRACE=1 cargo +nightly fuzz run $target -- \
+        -rss_limit_mb=4096 \
+        -max_total_time=360 \
+        -artifact_prefix=artifacts/ \
+        -print_final_stats=1
+}
+
+run_fuzzer header_length

Cargo.toml

@@ -67,11 +67,14 @@ in-use-encryption-unstable = ["in-use-encryption"]
 # The tracing API is unstable and may have backwards-incompatible changes in minor version updates.
 # TODO: pending https://github.com/tokio-rs/tracing/issues/2036 stop depending directly on log.
 tracing-unstable = ["dep:tracing", "dep:log"]
+fuzzing = ["dep:arbitrary", "arbitrary/derive", "dep:byteorder"]
 
 [dependencies]
 async-trait = "0.1.42"
 base64 = "0.13.0"
-bitflags = "1.1.0"
+bitflags = { version = "1.1.0" }
+arbitrary = { version = "1.3", optional = true }
+byteorder = { version = "1.4", optional = true }
 bson = { git = "https://github.com/mongodb/bson-rust", branch = "main", version = "2.11.0" }
 chrono = { version = "0.4.7", default-features = false, features = [
     "clock",

fuzz/.gitignore

@@ -0,0 +1,4 @@
+target
+corpus
+artifacts
+coverage

fuzz/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "mongodb-fuzz"
+version = "0.0.0"
+publish = false
+edition = "2021"
+
+[package.metadata]
+cargo-fuzz = true
+
+[dependencies]
+libfuzzer-sys = "0.4"
+mongodb = { path = "..", features = ["fuzzing"] }
+
+[[bin]]
+name = "header_length"
+path = "fuzz_targets/header_length.rs"
+test = false
+doc = false

fuzz/fuzz_targets/header_length.rs

@@ -0,0 +1,28 @@
+#![no_main]
+use libfuzzer_sys::fuzz_target;
+use mongodb::cmap::conn::wire::{
+    header::{Header, OpCode},
+    message::Message,
+};
+
+fuzz_target!(|data: &[u8]| {
+    if data.len() < Header::LENGTH {
+        return;
+    }
+    if let Ok(mut header) = Header::from_slice(data) {
+        // read_from_slice will adjust the data for the header length, this first check will
+        // almost always have length mismatches, but length mismatches are a possible attack
+        // vector.
+        // This will also often have the wrong opcode, but that is also a possible attack vector.
+        if let Ok(message) = Message::read_from_slice(data, header.clone()) {
+            let _ = message;
+        }
+        // try again with the header.length set to the data length and the header.opcode ==
+        // OpCode::Message to catch other attack vectors.
+        header.length = data.len() as i32 - Header::LENGTH as i32;
+        header.op_code = OpCode::Message;
+        if let Ok(message) = Message::read_from_slice(data, header) {
+            let _ = message;
+        }
+    }
+});

src/bson_util.rs

@@ -158,12 +158,14 @@ fn num_decimal_digits(mut n: usize) -> usize {
 
 /// Read a document's raw BSON bytes from the provided reader.
 pub(crate) fn read_document_bytes<R: Read>(mut reader: R) -> Result<Vec<u8>> {
-    let length = reader.read_i32_sync()?;
+    let length = Checked::new(reader.read_i32_sync()?);
 
-    let mut bytes = Vec::with_capacity(length as usize);
-    bytes.write_all(&length.to_le_bytes())?;
+    let mut bytes = Vec::with_capacity(length.try_into()?);
+    bytes.write_all(&length.try_into::<u32>()?.to_le_bytes())?;
 
-    reader.take(length as u64 - 4).read_to_end(&mut bytes)?;
+    reader
+        .take((length - 4).try_into()?)
+        .read_to_end(&mut bytes)?;
 
     Ok(bytes)
 }

src/client/options/parse.rs

@@ -80,6 +80,7 @@ impl Action for ParseConnectionString {
                                 "srvMaxHosts and loadBalanced=true cannot both be present",
                             ));
                         }
+                        // max is u32, so this is safe.
                         config.hosts = crate::sdam::choose_n(&config.hosts, max as usize)
                             .cloned()
                             .collect();

src/cmap.rs

@@ -1,7 +1,13 @@
 #[cfg(test)]
 pub(crate) mod test;
 
+#[cfg(feature = "fuzzing")]
+#[allow(missing_docs)]
+pub mod conn;
+
+#[cfg(not(feature = "fuzzing"))]
 pub(crate) mod conn;
+
 mod connection_requester;
 pub(crate) mod establish;
 mod manager;

src/cmap/conn.rs

@@ -1,6 +1,12 @@
 mod command;
 pub(crate) mod pooled;
 mod stream_description;
+
+#[cfg(feature = "fuzzing")]
+#[allow(missing_docs)]
+pub mod wire;
+
+#[cfg(not(feature = "fuzzing"))]
 pub(crate) mod wire;
 
 use std::{sync::Arc, time::Instant};

src/cmap/conn/wire.rs

@@ -1,8 +1,26 @@
+#[cfg(feature = "fuzzing")]
+#[allow(missing_docs)]
+pub mod header;
+#[cfg(not(feature = "fuzzing"))]
 mod header;
+#[cfg(feature = "fuzzing")]
+#[allow(missing_docs)]
+pub mod message;
+#[cfg(not(feature = "fuzzing"))]
 pub(crate) mod message;
+#[cfg(feature = "fuzzing")]
+#[allow(missing_docs)]
+pub mod util;
+#[cfg(not(feature = "fuzzing"))]
 mod util;
 
-pub(crate) use self::{
-    message::{Message, MessageFlags},
-    util::next_request_id,
-};
+pub(crate) use self::util::next_request_id;
+
+#[cfg(feature = "fuzzing")]
+pub use self::message::Message;
+
+#[cfg(feature = "fuzzing")]
+pub use crate::fuzz::message_flags::MessageFlags;
+
+#[cfg(not(feature = "fuzzing"))]
+pub use message::{Message, MessageFlags};

src/cmap/conn/wire/header.rs

@@ -1,10 +1,17 @@
+use crate::error::{ErrorKind, Result};
 use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
-use crate::error::{ErrorKind, Result};
+#[cfg(feature = "fuzzing")]
+use arbitrary::Arbitrary;
+#[cfg(feature = "fuzzing")]
+use byteorder::{LittleEndian, ReadBytesExt};
+#[cfg(feature = "fuzzing")]
+use std::io::Cursor;
 
 /// The wire protocol op codes.
 #[derive(Copy, Clone, Debug, Eq, PartialEq)]
-pub(crate) enum OpCode {
+#[cfg_attr(feature = "fuzzing", derive(Arbitrary))]
+pub enum OpCode {
     Reply = 1,
     Query = 2004,
     Message = 2013,
@@ -13,7 +20,7 @@ pub(crate) enum OpCode {
 
 impl OpCode {
     /// Attempt to infer the op code based on the numeric value.
-    fn from_i32(i: i32) -> Result<Self> {
+    pub fn from_i32(i: i32) -> Result<Self> {
         match i {
             1 => Ok(OpCode::Reply),
             2004 => Ok(OpCode::Query),
@@ -28,18 +35,71 @@ impl OpCode {
 }
 
 /// The header for any wire protocol message.
-#[derive(Debug)]
-pub(crate) struct Header {
+#[derive(Debug, Clone)]
+#[cfg_attr(feature = "fuzzing", derive(Arbitrary))]
+pub struct Header {
     pub length: i32,
     pub request_id: i32,
     pub response_to: i32,
     pub op_code: OpCode,
 }
 
 impl Header {
+    #[cfg(feature = "fuzzing")]
+    pub const LENGTH: usize = 4 * std::mem::size_of::<i32>();
+
+    #[cfg(not(feature = "fuzzing"))]
     pub(crate) const LENGTH: usize = 4 * std::mem::size_of::<i32>();
 
-    /// Serializes the Header and writes the bytes to `w`.
+    // generates a Header from a randomly generated slice of bytes, as long as the slice is at least
+    // 16 bytes long this is used for fuzzing
+    #[cfg(feature = "fuzzing")]
+    pub fn from_slice(data: &[u8]) -> Result<Self> {
+        if data.len() < Self::LENGTH {
+            return Err(ErrorKind::InvalidResponse {
+                message: format!(
+                    "Header requires {} bytes but only got {}",
+                    Self::LENGTH,
+                    data.len()
+                ),
+            }
+            .into());
+        }
+        let mut cursor = Cursor::new(data);
+
+        let length = ReadBytesExt::read_i32::<LittleEndian>(&mut cursor).map_err(|e| {
+            ErrorKind::InvalidResponse {
+                message: format!("Failed to read length: {}", e),
+            }
+        })?;
+
+        let request_id = ReadBytesExt::read_i32::<LittleEndian>(&mut cursor).map_err(|e| {
+            ErrorKind::InvalidResponse {
+                message: format!("Failed to read request_id: {}", e),
+            }
+        })?;
+
+        let response_to = ReadBytesExt::read_i32::<LittleEndian>(&mut cursor).map_err(|e| {
+            ErrorKind::InvalidResponse {
+                message: format!("Failed to read response_to: {}", e),
+            }
+        })?;
+
+        let op_code =
+            OpCode::from_i32(ReadBytesExt::read_i32::<LittleEndian>(&mut cursor).map_err(
+                |e| ErrorKind::InvalidResponse {
+                    message: format!("Failed to read op_code: {}", e),
+                },
+            )?)?;
+
+        Ok(Self {
+            length,
+            request_id,
+            response_to,
+            op_code,
+        })
+    }
+
     pub(crate) async fn write_to<W: AsyncWrite + Unpin>(&self, stream: &mut W) -> Result<()> {
         stream.write_all(&self.length.to_le_bytes()).await?;
         stream.write_all(&self.request_id.to_le_bytes()).await?;
@@ -51,7 +111,6 @@ impl Header {
         Ok(())
     }
 
-    /// Reads bytes from `r` and deserializes them into a header.
     pub(crate) async fn read_from<R: tokio::io::AsyncRead + Unpin + Send>(
         reader: &mut R,
     ) -> Result<Self> {