Implement Event Stream message frame ser/de in smithy-eventstream

rust-runtime/Cargo.toml

@@ -3,6 +3,7 @@
 members = [
     "smithy-http",
     "smithy-client",
+    "smithy-eventstream",
     "smithy-http-tower",
     "smithy-json",
     "smithy-query",

rust-runtime/smithy-eventstream/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "smithy-eventstream"
+version = "0.1.0"
+authors = ["AWS Rust SDK Team <[email protected]>", "John DiSanti <[email protected]>"]
+edition = "2018"
+
+[features]
+derive-arbitrary = ["arbitrary"]
+default = []
+
+[dependencies]
+arbitrary = { version = "1", optional = true, features = ["derive"] }
+bytes = "1"
+crc32fast = "1"
+smithy-types = { path = "../smithy-types" }
+
+[dev-dependencies]
+bytes-utils = "0.1"

rust-runtime/smithy-eventstream/fuzz/.gitignore

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

rust-runtime/smithy-eventstream/fuzz/Cargo.toml

@@ -0,0 +1,54 @@
+[package]
+name = "smithy-eventstream-fuzz"
+version = "0.1.0"
+authors = ["AWS Rust SDK Team <[email protected]>", "John DiSanti <[email protected]>"]
+publish = false
+edition = "2018"
+
+[package.metadata]
+cargo-fuzz = true
+
+[dependencies]
+arbitrary = { version = "1", features = ["derive"] }
+bytes = "1"
+crc32fast = "1"
+libfuzzer-sys = "0.4"
+smithy-types = { path = "../../smithy-types" }
+
+[dependencies.smithy-eventstream]
+features = ["derive-arbitrary"]
+path = ".."
+
+# Prevent this from interfering with workspaces
+[workspace]
+members = ["."]
+
+[[bin]]
+name = "raw_bytes"
+path = "fuzz_targets/raw_bytes.rs"
+test = false
+doc = false
+
+[[bin]]
+name = "round_trip"
+path = "fuzz_targets/round_trip.rs"
+test = false
+doc = false
+
+[[bin]]
+name = "corrected_prelude_crc"
+path = "fuzz_targets/corrected_prelude_crc.rs"
+test = false
+doc = false
+
+[[bin]]
+name = "mutated_headers"
+path = "fuzz_targets/mutated_headers.rs"
+test = false
+doc = false
+
+[[bin]]
+name = "prelude"
+path = "fuzz_targets/prelude.rs"
+test = false
+doc = false

rust-runtime/smithy-eventstream/fuzz/fuzz_targets/corrected_prelude_crc.rs

@@ -0,0 +1,41 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#![no_main]
+
+use arbitrary::Arbitrary;
+use bytes::BufMut;
+use crc32fast::Hasher as Crc;
+use libfuzzer_sys::fuzz_target;
+use smithy_eventstream::frame::Message;
+
+#[derive(Arbitrary, Debug)]
+struct Input {
+    headers: Vec<u8>,
+    payload: Vec<u8>,
+}
+
+// This fuzz test assists the fuzzer by creating a well formed prelude and message CRC
+fuzz_target!(|input: Input| {
+    let total_len = (12 + input.headers.len() + input.payload.len() + 4) as u32;
+    let header_len = input.headers.len() as u32;
+
+    let mut message = Vec::<u8>::new();
+    message.put_u32(total_len);
+    message.put_u32(header_len);
+    message.put_u32(crc(&message));
+    message.put_slice(&input.headers);
+    message.put_slice(&input.payload);
+    message.put_u32(crc(&message));
+
+    let mut data = &mut &message[..];
+    let _ = Message::read_from(&mut data);
+});
+
+fn crc(input: &[u8]) -> u32 {
+    let mut crc = Crc::new();
+    crc.update(input);
+    crc.finalize()
+}

rust-runtime/smithy-eventstream/fuzz/fuzz_targets/mutated_headers.rs

@@ -0,0 +1,82 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#![no_main]
+
+use bytes::{Buf, BufMut};
+use crc32fast::Hasher as Crc;
+use libfuzzer_sys::{fuzz_mutator, fuzz_target};
+use smithy_eventstream::frame::{Header, HeaderValue, Message};
+use smithy_types::Instant;
+
+// This fuzz test uses a custom mutator to manipulate the headers.
+// If it fails to parse a message from the unmutated input, it will create a message
+// with every single possible header type to give the fuzzer a leg up.
+// After the headers are mutated, a new valid prelude and valid message CRC are generated
+// so that the fuzzer can actually explore the header parsing logic.
+fn mutate(data: &mut [u8], size: usize, max_size: usize) -> usize {
+    let input = &mut &data[..size];
+    let message = if let Ok(message) = Message::read_from(input) {
+        message
+    } else {
+        Message::new(&b"some payload"[..])
+            .add_header(Header::new("true", HeaderValue::Bool(true)))
+            .add_header(Header::new("false", HeaderValue::Bool(false)))
+            .add_header(Header::new("byte", HeaderValue::Byte(50)))
+            .add_header(Header::new("short", HeaderValue::Int16(20_000)))
+            .add_header(Header::new("int", HeaderValue::Int32(500_000)))
+            .add_header(Header::new("long", HeaderValue::Int64(50_000_000_000)))
+            .add_header(Header::new(
+                "bytes",
+                HeaderValue::ByteArray((&b"some bytes"[..]).into()),
+            ))
+            .add_header(Header::new("str", HeaderValue::String("some str".into())))
+            .add_header(Header::new(
+                "time",
+                HeaderValue::Timestamp(Instant::from_epoch_seconds(5_000_000_000)),
+            ))
+            .add_header(Header::new(
+                "uuid",
+                HeaderValue::Uuid(0xb79bc914_de21_4e13_b8b2_bc47e85b7f0b),
+            ))
+    };
+
+    let mut bytes = Vec::new();
+    message.write_to(&mut bytes).unwrap();
+
+    let headers_len = (&bytes[4..8]).get_u32();
+    let non_header_len = bytes.len() - headers_len as usize;
+    let max_header_len = max_size - non_header_len;
+    let mut headers = (&bytes[12..(12 + headers_len as usize)]).to_vec();
+    headers.resize(max_header_len, 0);
+    let new_header_len =
+        libfuzzer_sys::fuzzer_mutate(&mut headers, headers_len as usize, max_header_len);
+
+    let mut mutated = Vec::<u8>::new();
+    mutated.put_u32((new_header_len + non_header_len) as u32);
+    mutated.put_u32(new_header_len as u32);
+    mutated.put_u32(crc(&mutated));
+    mutated.put_slice(&headers[..new_header_len]);
+    mutated.put_slice(message.payload());
+    mutated.put_u32(crc(&mutated));
+
+    data[..mutated.len()].copy_from_slice(&mutated);
+    mutated.len()
+}
+
+fuzz_mutator!(
+    |data: &mut [u8], size: usize, max_size: usize, _seed: u32| { mutate(data, size, max_size) }
+);
+
+fuzz_target!(|data: &[u8]| {
+    let mut message = data;
+    let _ = Message::read_from(&mut message);
+});
+
+fn crc(input: &[u8]) -> u32 {
+    let mut crc = Crc::new();
+    crc.update(input);
+    crc.finalize()
+}

rust-runtime/smithy-eventstream/fuzz/fuzz_targets/prelude.rs

@@ -0,0 +1,46 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#![no_main]
+
+use arbitrary::Arbitrary;
+use bytes::{Buf, BufMut};
+use crc32fast::Hasher as Crc;
+use libfuzzer_sys::fuzz_target;
+use smithy_eventstream::frame::{Header, HeaderValue, Message};
+
+#[derive(Arbitrary, Debug)]
+struct Input {
+    total_len: u32,
+    header_len: u32,
+}
+
+// This fuzz test exclusively fuzzes the message prelude while keeping the CRCs valid.
+fuzz_target!(|input: Input| {
+    let message = Message::new(&b"some payload"[..])
+        .add_header(Header::new("str", HeaderValue::String("some str".into())));
+
+    let mut bytes = Vec::new();
+    message.write_to(&mut bytes).unwrap();
+
+    let headers_len = (&bytes[4..8]).get_u32();
+    let headers = &bytes[12..(12 + headers_len as usize)];
+
+    let mut mutated = Vec::<u8>::new();
+    mutated.put_u32(input.total_len);
+    mutated.put_u32(input.header_len);
+    mutated.put_u32(crc(&mutated));
+    mutated.put_slice(headers);
+    mutated.put_slice(message.payload());
+    mutated.put_u32(crc(&mutated));
+
+    let _ = Message::read_from(&mut &mutated[..]);
+});
+
+fn crc(input: &[u8]) -> u32 {
+    let mut crc = Crc::new();
+    crc.update(input);
+    crc.finalize()
+}

rust-runtime/smithy-eventstream/fuzz/fuzz_targets/raw_bytes.rs

@@ -0,0 +1,14 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#![no_main]
+
+use libfuzzer_sys::fuzz_target;
+use smithy_eventstream::frame::Message;
+
+fuzz_target!(|data: &[u8]| {
+    let mut message = data;
+    let _ = Message::read_from(&mut message);
+});

rust-runtime/smithy-eventstream/fuzz/fuzz_targets/round_trip.rs

@@ -0,0 +1,27 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#![no_main]
+use libfuzzer_sys::fuzz_target;
+use smithy_eventstream::error::Error;
+use smithy_eventstream::frame::Message;
+
+fuzz_target!(|message: Message| {
+    let mut buffer = Vec::new();
+    match message.write_to(&mut buffer) {
+        Err(
+            Error::HeadersTooLong
+            | Error::PayloadTooLong
+            | Error::MessageTooLong
+            | Error::InvalidHeaderNameLength,
+        ) => {}
+        Err(err) => panic!("unexpected error on write: {}", err),
+        Ok(_) => {
+            let mut data = &buffer[..];
+            let parsed = Message::read_from(&mut data).unwrap();
+            assert_eq!(message, parsed);
+        }
+    }
+});

rust-runtime/smithy-eventstream/src/buf.rs

@@ -0,0 +1,7 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+pub mod count;
+pub mod crc;

rust-runtime/smithy-eventstream/src/buf/count.rs

@@ -0,0 +1,89 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+//! A [`Buf`] implementation that counts bytes read.
+
+use bytes::Buf;
+
+/// A [`Buf`] implementation that counts bytes read.
+pub struct CountBuf<'a, B>
+where
+    B: Buf,
+{
+    buffer: &'a mut B,
+    count: usize,
+}
+
+impl<'a, B> CountBuf<'a, B>
+where
+    B: Buf,
+{
+    /// Creates a new `CountBuf` by wrapping the given `buffer`.
+    pub fn new(buffer: &'a mut B) -> Self {
+        CountBuf { buffer, count: 0 }
+    }
+
+    /// Consumes the `CountBuf` and returns the number of bytes read.
+    pub fn into_count(self) -> usize {
+        self.count
+    }
+}
+
+impl<'a, B> Buf for CountBuf<'a, B>
+where
+    B: Buf,
+{
+    fn remaining(&self) -> usize {
+        self.buffer.remaining()
+    }
+
+    fn chunk(&self) -> &[u8] {
+        self.buffer.chunk()
+    }
+
+    fn advance(&mut self, cnt: usize) {
+        self.count += cnt;
+        self.buffer.advance(cnt);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::CountBuf;
+    use bytes::Buf;
+
+    #[test]
+    fn count_no_data_read() {
+        let mut data: &[u8] = &[];
+        let buf = CountBuf::new(&mut data);
+        assert_eq!(0, buf.into_count());
+    }
+
+    #[test]
+    fn count_data_read() {
+        let mut data: &[u8] = &[0, 0, 0, 5, 0, 10u8];
+        let mut buf = CountBuf::new(&mut data);
+        assert_eq!(5, buf.get_i32());
+        assert_eq!(4, buf.into_count());
+
+        let mut data: &[u8] = &[0, 0, 0, 5, 0, 10u8];
+        let mut buf = CountBuf::new(&mut data);
+        assert_eq!(5, buf.get_i32());
+        assert_eq!(10, buf.get_i16());
+        assert_eq!(6, buf.into_count());
+    }
+
+    #[test]
+    fn chunk_called_multiple_times_before_advance() {
+        let mut data: &[u8] = &[0, 0, 0, 5, 0, 10u8];
+        let mut buf = CountBuf::new(&mut data);
+        for _ in 0..3 {
+            buf.chunk();
+        }
+        buf.advance(4);
+        assert_eq!(10, buf.get_i16());
+        assert_eq!(6, buf.into_count());
+    }
+}