feat[tokio-quiche]: implement server-side support for 0-rtt

Author: toidiu

h3i/src/client/sync_client.rs

@@ -105,6 +105,10 @@ fn create_config(args: &Config, should_log_keys: bool) -> quiche::Config {
     config.set_max_stream_window(args.max_stream_window);
     config.grease(false);
 
+    if args.enable_early_data {
+        config.enable_early_data();
+    }
+
     if args.enable_dgram {
         config.enable_dgram(
             true,
@@ -132,6 +136,16 @@ fn create_config(args: &Config, should_log_keys: bool) -> quiche::Config {
 pub fn connect(
     args: Config, actions: Vec<Action>,
     close_trigger_frames: Option<CloseTriggerFrames>,
+) -> std::result::Result<ConnectionSummary, ClientError> {
+    connect_with_early_data(args, None, actions, close_trigger_frames)
+}
+
+/// Connect to a server and execute provided early_action and actions.
+///
+/// See `connect` for additional documentation.
+pub fn connect_with_early_data(
+    args: Config, early_actions: Option<Vec<Action>>, actions: Vec<Action>,
+    close_trigger_frames: Option<CloseTriggerFrames>,
 ) -> std::result::Result<ConnectionSummary, ClientError> {
     let mut buf = [0; 65535];
     let mut out = [0; MAX_DATAGRAM_SIZE];
@@ -178,11 +192,16 @@ pub fn connect(
         return Err(ClientError::Other("invalid socket".to_string()));
     };
 
-    // Create a QUIC connection and initiate handshake.
+    // Create a new client-side QUIC connection.
     let mut conn =
         quiche::connect(connect_url, &scid, local_addr, peer_addr, &mut config)
             .map_err(|e| ClientError::Other(e.to_string()))?;
 
+    if let Some(session) = &args.session {
+        conn.set_session(session)
+            .map_err(|error| ClientError::Other(error.to_string()))?;
+    }
+
     if let Some(keylog) = &mut keylog {
         if let Ok(keylog) = keylog.try_clone() {
             conn.set_keylog(Box::new(keylog));
@@ -195,8 +214,30 @@ pub fn connect(
 
     let mut app_proto_selected = false;
 
+    // Send ClientHello and initiate the handshake.
     let (write, send_info) = conn.send(&mut out).expect("initial send failed");
 
+    let mut client = SyncClient::new(close_trigger_frames);
+    // Send early data if connection is_in_early_data (resumption with 0-RTT was
+    // successful) and if we have early_actions.
+    if conn.is_in_early_data() {
+        if let Some(early_actions) = early_actions {
+            let mut early_action_iter = early_actions.iter();
+            let mut wait_duration = None;
+            let mut wait_instant = None;
+            let mut waiting_for = WaitingFor::default();
+
+            check_duration_and_do_actions(
+                &mut wait_duration,
+                &mut wait_instant,
+                &mut early_action_iter,
+                &mut conn,
+                &mut waiting_for,
+                client.stream_parsers_mut(),
+            );
+        }
+    }
+
     while let Err(e) = socket.send_to(&out[..write], send_info.to) {
         if e.kind() == std::io::ErrorKind::WouldBlock {
             log::debug!(
@@ -216,7 +257,6 @@ pub fn connect(
     let mut wait_duration = None;
     let mut wait_instant = None;
 
-    let mut client = SyncClient::new(close_trigger_frames);
     let mut waiting_for = WaitingFor::default();
 
     loop {

h3i/src/config.rs

@@ -63,6 +63,8 @@ pub struct Config {
     pub max_stream_window: u64,
     /// Set the session to attempt resumption.
     pub session: Option<Vec<u8>>,
+    /// Enables sending or receiving early data.
+    pub enable_early_data: bool,
     /// Whether to enable datagram sending.
     pub enable_dgram: bool,
     /// Datagram receive queue length.
@@ -194,6 +196,7 @@ impl Config {
             max_window: self.max_window,
             max_stream_window: self.max_stream_window,
             session: None,
+            enable_early_data: self.enable_early_data,
             enable_dgram: self.enable_dgram,
             dgram_recv_queue_len: self.dgram_recv_queue_len,
             dgram_send_queue_len: self.dgram_send_queue_len,
@@ -220,6 +223,7 @@ impl Default for Config {
             max_window: 25165824,
             max_stream_window: 16777216,
             session: None,
+            enable_early_data: false,
             enable_dgram: true,
             dgram_recv_queue_len: 65536,
             dgram_send_queue_len: 65536,

h3i/src/main.rs

@@ -326,6 +326,7 @@ fn config_from_clap() -> std::result::Result<Config, String> {
         max_window,
         max_stream_window,
         session: None,
+        enable_early_data: false,
         enable_dgram,
         dgram_recv_queue_len,
         dgram_send_queue_len,

quiche/src/lib.rs

@@ -7153,6 +7153,17 @@ impl<F: BufFactory> Connection<F> {
         self.handshake.is_in_early_data()
     }
 
+    /// Returns the early data reason for the connection.
+    ///
+    /// This status can be useful for logging and debugging. See [BoringSSL]
+    /// documentation for a definition of the reasons.
+    ///
+    /// [BoringSSL]: https://commondatastorage.googleapis.com/chromium-boringssl-docs/ssl.h.html#ssl_early_data_reason_t
+    #[inline]
+    pub fn early_data_reason(&self) -> u32 {
+        self.handshake.early_data_reason()
+    }
+
     /// Returns whether there is stream or DATAGRAM data available to read.
     #[inline]
     pub fn is_readable(&self) -> bool {

quiche/src/tls/boringssl.rs

@@ -274,6 +274,12 @@ impl Handshake {
     pub fn is_in_early_data(&self) -> bool {
         unsafe { SSL_in_early_data(self.as_ptr()) == 1 }
     }
+
+    pub fn early_data_reason(&self) -> u32 {
+        let reuse_reason_status =
+            unsafe { SSL_get_early_data_reason(self.as_ptr()) };
+        reuse_reason_status.0
+    }
 }
 
 pub(super) fn get_session_bytes(session: *mut SSL_SESSION) -> Result<Vec<u8>> {
@@ -293,6 +299,10 @@ pub(super) fn get_session_bytes(session: *mut SSL_SESSION) -> Result<Vec<u8>> {
 }
 pub(super) const TLS_ERROR: c_int = 3;
 
+#[allow(non_camel_case_types)]
+#[repr(transparent)]
+#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq)]
+pub struct ssl_early_data_reason_t(pub ::std::os::raw::c_uint);
 extern "C" {
     // SSL_METHOD specific for boringssl.
     pub(super) fn SSL_CTX_set_tlsext_ticket_keys(
@@ -341,6 +351,8 @@ extern "C" {
 
     fn SSL_in_early_data(ssl: *const SSL) -> c_int;
 
+    fn SSL_get_early_data_reason(ssl: *const SSL) -> ssl_early_data_reason_t;
+
     fn SSL_SESSION_to_bytes(
         session: *const SSL_SESSION, out: *mut *mut u8, out_len: *mut usize,
     ) -> c_int;

quiche/src/tls/openssl_quictls.rs

@@ -141,6 +141,10 @@ impl Handshake {
         false
     }
 
+    pub fn early_data_reason(&self) -> u32 {
+        0
+    }
+
     pub fn set_session(&mut self, session: &[u8]) -> Result<()> {
         unsafe {
             let ctx = SSL_get_SSL_CTX(self.as_ptr());

tokio-quiche/examples/async_http3_server/server.rs

@@ -147,6 +147,7 @@ where
             ServerH3Event::Headers {
                 incoming_headers,
                 priority,
+                is_in_early_data: _,
             } => {
                 // Received headers for a new stream from the H3Driver.
                 self.handle_incoming_headers(incoming_headers, priority)

tokio-quiche/src/http3/driver/mod.rs

@@ -88,6 +88,7 @@ pub use self::client::ClientH3Driver;
 pub use self::client::ClientH3Event;
 pub use self::client::ClientRequestSender;
 pub use self::client::NewClientRequest;
+pub use self::server::IsInEarlyData;
 pub use self::server::RawPriorityValue;
 pub use self::server::ServerEventStream;
 pub use self::server::ServerH3Command;

tokio-quiche/src/http3/driver/server.rs

@@ -78,6 +78,24 @@ impl Deref for RawPriorityValue {
     }
 }
 
+/// Custom header info to track early data requests.
+#[derive(Clone, Debug)]
+pub struct IsInEarlyData(bool);
+
+impl IsInEarlyData {
+    fn new(is_in_early_data: bool) -> Self {
+        IsInEarlyData(is_in_early_data)
+    }
+}
+
+impl Deref for IsInEarlyData {
+    type Target = bool;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
 /// Events produced by [ServerH3Driver].
 #[derive(Debug)]
 pub enum ServerH3Event {
@@ -87,15 +105,24 @@ pub enum ServerH3Event {
         incoming_headers: IncomingH3Headers,
         /// The latest PRIORITY_UPDATE frame value, if any.
         priority: Option<RawPriorityValue>,
+        is_in_early_data: IsInEarlyData,
     },
 }
 
 impl From<H3Event> for ServerH3Event {
     fn from(ev: H3Event) -> Self {
         match ev {
-            H3Event::IncomingHeaders(incoming_headers) => Self::Headers {
-                incoming_headers,
-                priority: None,
+            H3Event::IncomingHeaders(incoming_headers) => {
+                // Requests are exclusively handled by `Self::headers_received`,
+                // which correctly serializers the RawPriorityValue and
+                // IsInEarlyData values.
+                //
+                // See `H3Driver::process_read_event` for implementation details.
+                Self::Headers {
+                    incoming_headers,
+                    priority: None,
+                    is_in_early_data: IsInEarlyData::new(false),
+                }
             },
             _ => Self::Core(ev),
         }
@@ -205,6 +232,7 @@ impl ServerHooks {
             .send(ServerH3Event::Headers {
                 incoming_headers: headers,
                 priority: latest_priority_update,
+                is_in_early_data: IsInEarlyData::new(qconn.is_in_early_data()),
             })
             .map_err(|_| H3ConnectionError::ControllerWentAway)?;
         driver.hooks.requests += 1;

tokio-quiche/src/quic/io/connection_stage.rs

@@ -122,7 +122,9 @@ impl ConnectionStage for Handshake {
         &mut self, qconn: &mut QuicheConnection,
         _ctx: &mut ConnectionStageContext<A>,
     ) -> ControlFlow<QuicResult<()>> {
-        if qconn.is_established() {
+        // Transition to RunningApplication if we have 1-RTT keys (handshake is
+        // complete) or if we have 0-RTT keys (in early data).
+        if qconn.is_established() || qconn.is_in_early_data() {
             ControlFlow::Break(Ok(()))
         } else {
             ControlFlow::Continue(())