feat(s2n-quic-dc): latest import (#2257)

* feat(s2n-quic-dc): latest import

* add timeout to thread_shutdown test

* fix merge conflicts

.github/config/cargo-deny.toml

@@ -1,7 +1,4 @@
 [advisories]
-vulnerability = "deny"
-unmaintained = "deny"
-notice = "deny"
 yanked = "deny"
 ignore = [
     "RUSTSEC-2021-0139", # criterion, structopt, and tracing-subscriber (test dependencies) use ansi_term
@@ -15,7 +12,6 @@ skip-tree = [
     { name = "cuckoofilter" },
 
     # all of these are going to be just test dependencies
-    { name = "aes-gcm" },
     { name = "bach" },
     { name = "bolero" },
     { name = "criterion" },
@@ -30,23 +26,18 @@ skip-tree = [
 ]
 
 [sources]
-allow-git = [
-    "https://github.com/camshaft/aya", # TODO: Remove once aya supports XdpMaps - https://github.com/aya-rs/aya/pull/527
-]
 unknown-registry = "deny"
 unknown-git = "deny"
 
 [licenses]
-unlicensed = "deny"
-allow-osi-fsf-free = "neither"
-copyleft = "deny"
 confidence-threshold = 0.9
 # ignore licenses for private crates
 private = { ignore = true }
 allow = [
     "Apache-2.0",
     "BSD-2-Clause",
     "BSD-3-Clause",
+    "CC0-1.0",
     "ISC",
     "MIT",
     "OpenSSL",

dc/s2n-quic-dc/Cargo.toml

@@ -21,22 +21,25 @@ bolero-generator = { version = "0.11", optional = true }
 bytes = "1"
 crossbeam-channel = "0.5"
 crossbeam-queue = { version = "0.3" }
+flurry = "0.5"
 libc = "0.2"
 num-rational = { version = "0.4", default-features = false }
 once_cell = "1"
+rand = { version = "0.8", features = ["small_rng"] }
 s2n-codec = { version = "=0.40.0", path = "../../common/s2n-codec", default-features = false }
 s2n-quic-core = { version = "=0.40.0", path = "../../quic/s2n-quic-core", default-features = false }
 s2n-quic-platform = { version = "=0.40.0", path = "../../quic/s2n-quic-platform" }
 slotmap = "1"
 thiserror = "1"
-tokio = { version = "1", features = ["io-util"], optional = true }
+tokio = { version = "1", features = ["sync"] }
 tracing = "0.1"
 zerocopy = { version = "0.7", features = ["derive"] }
+zeroize = "1"
 
 [dev-dependencies]
 bolero = "0.11"
 bolero-generator = "0.11"
 insta = "1"
 s2n-codec = { path = "../../common/s2n-codec", features = ["testing"] }
 s2n-quic-core = { path = "../../quic/s2n-quic-core", features = ["testing"] }
-tokio = { version = "1", features = ["io-util"] }
+tokio = { version = "1", features = ["sync"] }

dc/s2n-quic-dc/src/datagram.rs

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

dc/s2n-quic-dc/src/datagram/tunneled.rs

@@ -0,0 +1,8 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+pub mod recv;
+pub mod send;
+
+pub use recv::Receiver;
+pub use send::Sender;

dc/s2n-quic-dc/src/datagram/tunneled/recv.rs

@@ -0,0 +1,80 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{
+    crypto::{decrypt, UninitSlice},
+    packet::datagram::{decoder, Tag},
+};
+use s2n_codec::{decoder_invariant, DecoderBufferMut, DecoderError};
+use s2n_quic_core::packet::number::{PacketNumberSpace, SlidingWindow, SlidingWindowError};
+
+pub use crate::crypto::decrypt::Error;
+pub use decoder::Packet;
+
+#[derive(Default)]
+pub struct Endpoint {}
+
+impl Endpoint {
+    pub fn parse<'a>(&self, payload: &'a mut [u8]) -> Option<(Packet<'a>, &'a mut [u8])> {
+        let buffer = DecoderBufferMut::new(payload);
+        let (packet, buffer) = Packet::decode(buffer, TagValidator, 16).ok()?;
+        let buffer = buffer.into_less_safe_slice();
+        Some((packet, buffer))
+    }
+}
+
+struct TagValidator;
+
+impl decoder::Validator for TagValidator {
+    #[inline]
+    fn validate_tag(&mut self, tag: Tag) -> Result<(), DecoderError> {
+        decoder_invariant!(!tag.ack_eliciting(), "expected tunnelled datagram");
+        decoder_invariant!(
+            !tag.has_application_header(),
+            "application headers currently unsupported"
+        );
+        Ok(())
+    }
+}
+
+pub struct Receiver<K: decrypt::Key> {
+    key: K,
+}
+
+impl<K: decrypt::Key> Receiver<K> {
+    pub fn new(key: K) -> Self {
+        Self { key }
+    }
+
+    pub fn recv_into(
+        &mut self,
+        packet: &Packet,
+        payload_out: &mut UninitSlice,
+    ) -> Result<(), Error> {
+        debug_assert_eq!(packet.payload().len(), payload_out.len());
+
+        self.key.decrypt(
+            packet.crypto_nonce(),
+            packet.header(),
+            packet.payload(),
+            packet.auth_tag(),
+            payload_out,
+        )?;
+
+        Ok(())
+    }
+}
+
+#[derive(Default)]
+pub struct SeenFilter {
+    window: SlidingWindow,
+}
+
+impl SeenFilter {
+    #[inline]
+    pub fn on_packet(&mut self, packet: &Packet) -> Result<(), SlidingWindowError> {
+        let packet_number =
+            PacketNumberSpace::ApplicationData.new_packet_number(packet.packet_number());
+        self.window.insert(packet_number)
+    }
+}

dc/s2n-quic-dc/src/datagram/tunneled/send.rs

@@ -0,0 +1,97 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{
+    control,
+    crypto::encrypt,
+    packet::{self, datagram::encoder},
+};
+use core::sync::atomic::{AtomicU64, Ordering};
+use s2n_codec::EncoderBuffer;
+use s2n_quic_core::{ensure, varint::VarInt};
+
+#[derive(Clone, Copy, Debug)]
+pub enum Error {
+    PayloadTooLarge,
+    PacketBufferTooSmall,
+    PacketNumberExhaustion,
+}
+
+pub struct Sender<E> {
+    encrypt_key: E,
+    packet_number: AtomicU64,
+}
+
+impl<E> Sender<E>
+where
+    E: encrypt::Key,
+{
+    #[inline]
+    pub fn new(encrypt_key: E) -> Self {
+        Self {
+            encrypt_key,
+            packet_number: AtomicU64::new(0),
+        }
+    }
+
+    #[inline]
+    pub fn estimated_send_size(&self, cleartext_payload_len: usize) -> Option<usize> {
+        let payload_len = packet::PayloadLen::try_from(cleartext_payload_len).ok()?;
+        Some(encoder::estimate_len(
+            VarInt::ZERO,
+            None,
+            VarInt::ZERO,
+            payload_len,
+            E::tag_len(&self.encrypt_key),
+        ))
+    }
+
+    #[inline]
+    pub fn send_into<C>(
+        &self,
+        control_port: &C,
+        mut cleartext_payload: &[u8],
+        encrypted_packet: &mut [u8],
+    ) -> Result<usize, Error>
+    where
+        C: control::Controller,
+    {
+        let packet_number = self.packet_number.fetch_add(1, Ordering::Relaxed);
+        let packet_number =
+            VarInt::new(packet_number).map_err(|_| Error::PacketNumberExhaustion)?;
+
+        let payload_len = packet::PayloadLen::try_from(cleartext_payload.len())
+            .map_err(|_| Error::PayloadTooLarge)?;
+
+        let estimated_packet_len = self
+            .estimated_send_size(cleartext_payload.len())
+            .ok_or(Error::PayloadTooLarge)?;
+
+        // ensure the descriptor has enough capacity after MTU/allocation
+        ensure!(
+            encrypted_packet.len() >= estimated_packet_len,
+            Err(Error::PacketBufferTooSmall)
+        );
+
+        let actual_packet_len = {
+            let source_control_port = control_port.source_port();
+
+            let encoder = EncoderBuffer::new(encrypted_packet);
+
+            encoder::encode(
+                encoder,
+                source_control_port,
+                Some(packet_number),
+                None,
+                VarInt::ZERO,
+                &mut &[][..],
+                &(),
+                payload_len,
+                &mut cleartext_payload,
+                &self.encrypt_key,
+            )
+        };
+
+        Ok(actual_packet_len)
+    }
+}

dc/s2n-quic-dc/src/lib.rs

@@ -6,10 +6,13 @@ pub mod congestion;
 pub mod control;
 pub mod credentials;
 pub mod crypto;
+pub mod datagram;
 pub mod msg;
 pub mod packet;
 pub mod path;
 pub mod pool;
 pub mod recovery;
 pub mod socket;
 pub mod stream;
+
+pub use s2n_quic_core::dc::{Version, SUPPORTED_VERSIONS};