feat(s2n-quic): unstable dc provider

quic/s2n-quic-core/src/dc/disabled.rs

@@ -0,0 +1,37 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{
+    crypto::tls::TlsSession,
+    dc::{ConnectionInfo, Endpoint, Path},
+    stateless_reset::Token,
+};
+
+#[derive(Debug, Default)]
+pub struct Disabled(());
+
+impl Endpoint for Disabled {
+    const ENABLED: bool = false;
+
+    type Path = DisabledPath;
+
+    fn new_path(&mut self, _connection_info: &ConnectionInfo) -> Self::Path {
+        DisabledPath(())
+    }
+}
+
+pub struct DisabledPath(());
+
+impl Path for DisabledPath {
+    fn on_path_secrets_ready(&mut self, _session: &impl TlsSession) {}
+
+    fn on_peer_stateless_reset_tokens<'a>(
+        &mut self,
+        _stateless_reset_tokens: impl Iterator<Item = &'a Token>,
+    ) {
+    }
+
+    fn stateless_reset_tokens(&mut self) -> &[Token] {
+        &[]
+    }
+}

quic/s2n-quic-core/src/dc/mod.rs

@@ -0,0 +1,146 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::{
+    connection::Limits,
+    crypto::tls::TlsSession,
+    event::{api::SocketAddress, IntoEvent as _},
+    inet,
+    path::MaxMtu,
+    stateless_reset,
+    transport::parameters::InitialFlowControlLimits,
+    varint::VarInt,
+};
+use core::time::Duration;
+
+pub use disabled::*;
+
+mod disabled;
+
+// dc versions supported by this code, in order of preference (SUPPORTED_VERSIONS[0] is most preferred)
+const SUPPORTED_VERSIONS: &[u32] = &[0x0];
+
+/// Called on the server to select the dc version to use (if any)
+///
+/// The server's version preference takes precedence
+pub fn select_version(client_supported_versions: &[u32]) -> Option<u32> {
+    SUPPORTED_VERSIONS
+        .iter()
+        .find(|&supported_version| client_supported_versions.contains(supported_version))
+        .copied()
+}
+
+/// The `dc::Endpoint` trait provides a way to support dc functionality
+pub trait Endpoint: 'static + Send {
+    /// If enabled, a dc version will attempt to be negotiated and dc-specific frames
+    /// will be processed. Otherwise, no dc version will be negotiated and dc-specific
+    /// frames received will result in a connection error.
+    const ENABLED: bool = true;
+
+    type Path: Path;
+
+    /// Called when a dc version has been negotiated for the given `ConnectionInfo`
+    fn new_path(&mut self, connection_info: &ConnectionInfo) -> Self::Path;
+}
+
+/// A dc path
+pub trait Path: 'static + Send {
+    /// Called when path secrets are ready to be derived from the given `TlsSession`
+    fn on_path_secrets_ready(&mut self, session: &impl TlsSession);
+
+    /// Called when a `DC_STATELESS_RESET_TOKENS` frame has been received from the peer
+    fn on_peer_stateless_reset_tokens<'a>(
+        &mut self,
+        stateless_reset_tokens: impl Iterator<Item = &'a stateless_reset::Token>,
+    );
+
+    /// Returns the stateless reset tokens to include in a `DC_STATELESS_RESET_TOKENS`
+    /// frame sent to the peer.
+    fn stateless_reset_tokens(&mut self) -> &[stateless_reset::Token];
+}
+
+impl<P: Path> Path for Option<P> {
+    #[inline]
+    fn on_path_secrets_ready(&mut self, session: &impl TlsSession) {
+        if let Some(path) = self {
+            path.on_path_secrets_ready(session)
+        }
+    }
+
+    #[inline]
+    fn on_peer_stateless_reset_tokens<'a>(
+        &mut self,
+        stateless_reset_tokens: impl Iterator<Item = &'a stateless_reset::Token>,
+    ) {
+        if let Some(path) = self {
+            path.on_peer_stateless_reset_tokens(stateless_reset_tokens)
+        }
+    }
+
+    #[inline]
+    fn stateless_reset_tokens(&mut self) -> &[stateless_reset::Token] {
+        if let Some(path) = self {
+            path.stateless_reset_tokens()
+        } else {
+            &[]
+        }
+    }
+}
+
+/// Information about the connection that may be used
+/// when create a new dc path
+#[derive(Clone, Debug)]
+#[non_exhaustive]
+pub struct ConnectionInfo<'a> {
+    /// The address (IP + Port) of the remote peer
+    pub remote_address: SocketAddress<'a>,
+    /// The dc version that has been negotiated
+    pub dc_version: u32,
+    /// Various settings relevant to the dc path
+    pub application_params: ApplicationParams,
+}
+
+impl<'a> ConnectionInfo<'a> {
+    #[inline]
+    #[doc(hidden)]
+    pub fn new(
+        remote_address: &'a inet::SocketAddress,
+        dc_version: u32,
+        application_params: ApplicationParams,
+    ) -> Self {
+        Self {
+            remote_address: remote_address.into_event(),
+            dc_version,
+            application_params,
+        }
+    }
+}
+
+/// Various settings relevant to the dc path
+#[derive(Clone, Debug)]
+#[non_exhaustive]
+pub struct ApplicationParams {
+    pub max_mtu: MaxMtu,
+    pub remote_max_data: VarInt,
+    pub local_send_max_data: VarInt,
+    pub local_recv_max_data: VarInt,
+    pub max_idle_timeout: Option<Duration>,
+    pub max_ack_delay: Duration,
+}
+
+impl ApplicationParams {
+    pub fn new(
+        max_mtu: MaxMtu,
+        peer_flow_control_limits: &InitialFlowControlLimits,
+        limits: &Limits,
+    ) -> Self {
+        Self {
+            max_mtu,
+            remote_max_data: peer_flow_control_limits.max_data,
+            local_send_max_data: limits.initial_stream_limits().max_data_bidi_local,
+            local_recv_max_data: limits.initial_stream_limits().max_data_bidi_remote,
+            max_idle_timeout: limits.max_idle_timeout(),
+            max_ack_delay: limits.max_ack_delay.into(),
+        }
+    }
+}

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

@@ -22,6 +22,7 @@ pub mod counter;
 pub mod crypto;
 pub mod ct;
 pub mod datagram;
+pub mod dc;
 pub mod endpoint;
 pub mod event;
 pub mod frame;

quic/s2n-quic-transport/src/connection/connection_container/tests.rs

@@ -133,6 +133,7 @@ impl connection::Trait for TestConnection {
         _timestamp: Timestamp,
         _subscriber: &mut <Self::Config as endpoint::Config>::EventSubscriber,
         _datagram: &mut <Self::Config as endpoint::Config>::DatagramEndpoint,
+        _dc_endpoint: &mut <Self::Config as endpoint::Config>::DcEndpoint,
     ) -> Result<(), connection::Error> {
         Ok(())
     }
@@ -146,6 +147,7 @@ impl connection::Trait for TestConnection {
         _subscriber: &mut <Self::Config as endpoint::Config>::EventSubscriber,
         _packet_interceptor: &mut <Self::Config as endpoint::Config>::PacketInterceptor,
         _datagram_endpoint: &mut <Self::Config as endpoint::Config>::DatagramEndpoint,
+        _dc_endpoint: &mut <Self::Config as endpoint::Config>::DcEndpoint,
     ) -> Result<(), ProcessingError> {
         Ok(())
     }
@@ -160,6 +162,7 @@ impl connection::Trait for TestConnection {
         _subscriber: &mut <Self::Config as endpoint::Config>::EventSubscriber,
         _packet_interceptor: &mut <Self::Config as endpoint::Config>::PacketInterceptor,
         _datagram_endpoint: &mut <Self::Config as endpoint::Config>::DatagramEndpoint,
+        _dc_endpoint: &mut <Self::Config as endpoint::Config>::DcEndpoint,
     ) -> Result<(), ProcessingError> {
         Ok(())
     }
@@ -174,6 +177,7 @@ impl connection::Trait for TestConnection {
         _subscriber: &mut <Self::Config as endpoint::Config>::EventSubscriber,
         _packet_interceptor: &mut <Self::Config as endpoint::Config>::PacketInterceptor,
         _datagram_endpoint: &mut <Self::Config as endpoint::Config>::DatagramEndpoint,
+        _dc_endpoint: &mut <Self::Config as endpoint::Config>::DcEndpoint,
     ) -> Result<(), ProcessingError> {
         Ok(())
     }
@@ -188,6 +192,7 @@ impl connection::Trait for TestConnection {
         _subscriber: &mut <Self::Config as endpoint::Config>::EventSubscriber,
         _packet_interceptor: &mut <Self::Config as endpoint::Config>::PacketInterceptor,
         _datagram_endpoint: &mut <Self::Config as endpoint::Config>::DatagramEndpoint,
+        _dc_endpoint: &mut <Self::Config as endpoint::Config>::DcEndpoint,
     ) -> Result<(), ProcessingError> {
         Ok(())
     }

quic/s2n-quic-transport/src/connection/connection_impl.rs

@@ -269,6 +269,7 @@ impl<Config: endpoint::Config> ConnectionImpl<Config> {
         timestamp: Timestamp,
         subscriber: &mut Config::EventSubscriber,
         datagram: &mut Config::DatagramEndpoint,
+        dc: &mut Config::DcEndpoint,
     ) -> Result<(), connection::Error> {
         let mut publisher = self.event_context.publisher(timestamp, subscriber);
         let space_manager = &mut self.space_manager;
@@ -281,6 +282,7 @@ impl<Config: endpoint::Config> ConnectionImpl<Config> {
             &self.waker,
             &mut publisher,
             datagram,
+            dc,
         ) {
             Poll::Ready(res) => res?,
             Poll::Pending => return Ok(()),
@@ -644,6 +646,7 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
                 parameters.timestamp,
                 parameters.event_subscriber,
                 parameters.datagram_endpoint,
+                parameters.dc_endpoint,
             ) {
                 connection.with_event_publisher(
                     parameters.timestamp,
@@ -1109,12 +1112,13 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
         timestamp: Timestamp,
         subscriber: &mut Config::EventSubscriber,
         datagram: &mut Config::DatagramEndpoint,
+        dc: &mut Config::DcEndpoint,
     ) -> Result<(), connection::Error> {
         // reset the queued state first so that new wakeup request are not missed
         self.wakeup_handle.wakeup_handled();
 
         // check if crypto progress can be made
-        self.update_crypto_state(timestamp, subscriber, datagram)?;
+        self.update_crypto_state(timestamp, subscriber, datagram, dc)?;
 
         // return an error if the application set one
         self.error?;
@@ -1200,6 +1204,7 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
         subscriber: &mut Config::EventSubscriber,
         packet_interceptor: &mut Config::PacketInterceptor,
         datagram_endpoint: &mut Config::DatagramEndpoint,
+        dc_endpoint: &mut Config::DcEndpoint,
     ) -> Result<(), ProcessingError> {
         //= https://www.rfc-editor.org/rfc/rfc9000#section-7.2
         //= type=TODO
@@ -1240,6 +1245,7 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
                 subscriber,
                 packet_interceptor,
                 datagram_endpoint,
+                dc_endpoint,
             )?;
         }
 
@@ -1256,6 +1262,7 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
         subscriber: &mut Config::EventSubscriber,
         packet_interceptor: &mut Config::PacketInterceptor,
         datagram_endpoint: &mut Config::DatagramEndpoint,
+        dc_endpoint: &mut Config::DcEndpoint,
     ) -> Result<(), ProcessingError> {
         if let Some((space, handshake_status)) = self.space_manager.initial_mut() {
             let mut publisher = self.event_context.publisher(datagram.timestamp, subscriber);
@@ -1312,7 +1319,12 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
             )?;
 
             // try to move the crypto state machine forward
-            self.update_crypto_state(datagram.timestamp, subscriber, datagram_endpoint)?;
+            self.update_crypto_state(
+                datagram.timestamp,
+                subscriber,
+                datagram_endpoint,
+                dc_endpoint,
+            )?;
 
             // notify the connection a packet was processed
             self.on_processed_packet(&processed_packet, subscriber)?;
@@ -1332,6 +1344,7 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
         subscriber: &mut Config::EventSubscriber,
         packet_interceptor: &mut Config::PacketInterceptor,
         datagram_endpoint: &mut Config::DatagramEndpoint,
+        dc_endpoint: &mut Config::DcEndpoint,
     ) -> Result<(), ProcessingError> {
         let mut publisher = self.event_context.publisher(datagram.timestamp, subscriber);
 
@@ -1412,7 +1425,12 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
             self.path_manager[path_id].on_handshake_packet();
 
             // try to move the crypto state machine forward
-            self.update_crypto_state(datagram.timestamp, subscriber, datagram_endpoint)?;
+            self.update_crypto_state(
+                datagram.timestamp,
+                subscriber,
+                datagram_endpoint,
+                dc_endpoint,
+            )?;
 
             // notify the connection a packet was processed
             self.on_processed_packet(&processed_packet, subscriber)?;
@@ -1431,6 +1449,7 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
         subscriber: &mut Config::EventSubscriber,
         packet_interceptor: &mut Config::PacketInterceptor,
         datagram_endpoint: &mut <Self::Config as endpoint::Config>::DatagramEndpoint,
+        dc_endpoint: &mut Config::DcEndpoint,
     ) -> Result<(), ProcessingError> {
         let mut publisher = self.event_context.publisher(datagram.timestamp, subscriber);
 
@@ -1546,6 +1565,7 @@ impl<Config: endpoint::Config> connection::Trait for ConnectionImpl<Config> {
                     &self.waker,
                     &mut publisher,
                     datagram_endpoint,
+                    dc_endpoint,
                 )?;
             }
             // notify the connection a packet was processed