fix(quic): fix quic accept issues

Author: mempirate

Cargo.lock

@@ -58,7 +58,7 @@ rand = "0.8"
 
 # networking
 quinn = "0.11.9"
-rcgen = "0.12"
+rcgen = "0.14"
 
 # benchmarking & profiling
 criterion = { version = "0.5", features = ["async_tokio"] }

Cargo.toml

@@ -78,14 +78,23 @@ pub trait TransportExt<A: Address>: Transport<A> {
     }
 }
 
-pub struct Acceptor<'a, T, A> {
+pub struct Acceptor<'a, T, A>
+where
+    T: Transport<A>,
+    A: Address,
+{
     inner: &'a mut T,
+    pending: Option<T::Accept>,
     _marker: PhantomData<A>,
 }
 
-impl<'a, T, A> Acceptor<'a, T, A> {
+impl<'a, T, A> Acceptor<'a, T, A>
+where
+    T: Transport<A>,
+    A: Address,
+{
     fn new(inner: &'a mut T) -> Self {
-        Self { inner, _marker: PhantomData }
+        Self { inner, pending: None, _marker: PhantomData }
     }
 }
 
@@ -97,13 +106,26 @@ where
     type Output = Result<T::Io, T::Error>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match Pin::new(&mut *self.get_mut().inner).poll_accept(cx) {
-            Poll::Ready(mut accept) => match accept.poll_unpin(cx) {
-                Poll::Ready(Ok(output)) => Poll::Ready(Ok(output)),
-                Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
-                Poll::Pending => Poll::Pending,
-            },
-            Poll::Pending => Poll::Pending,
+        let this = self.get_mut();
+
+        loop {
+            if let Some(pending) = this.pending.as_mut() {
+                match pending.poll_unpin(cx) {
+                    Poll::Ready(res) => {
+                        this.pending = None;
+                        return Poll::Ready(res);
+                    }
+                    Poll::Pending => return Poll::Pending,
+                }
+            }
+
+            match Pin::new(&mut *this.inner).poll_accept(cx) {
+                Poll::Ready(accept) => {
+                    this.pending = Some(accept);
+                    continue;
+                }
+                Poll::Pending => return Poll::Pending,
+            }
         }
     }
 }

msg-transport/src/lib.rs

@@ -113,24 +113,24 @@ impl Transport<SocketAddr> for Quic {
         let endpoint = if let Some(endpoint) = self.endpoint.clone() {
             endpoint
         } else {
-            let Ok(endpoint) = self.new_endpoint(None, None) else {
+            let Ok(mut endpoint) = self.new_endpoint(None, None) else {
                 return async_error(Error::ClosedEndpoint);
             };
 
+            endpoint.set_default_client_config(self.config.client_config.clone());
+
             self.endpoint = Some(endpoint.clone());
 
             endpoint
         };
 
-        let client_config = self.config.client_config.clone();
-
         Box::pin(async move {
             debug!(target = %addr, "Initiating connection");
 
             // This `"l"` seems necessary because an empty string is an invalid domain
             // name. While we don't use domain names, the underlying rustls library
             // is based upon the assumption that we do.
-            let connection = endpoint.connect_with(client_config, addr, "localhost")?.await?;
+            let connection = endpoint.connect(addr, "l")?.await?;
 
             debug!(target = %addr, "Connected, opening stream...");
 
@@ -152,15 +152,15 @@ impl Transport<SocketAddr> for Quic {
             if let Some(ref mut incoming) = this.incoming {
                 // Incoming channel and task are spawned, so we can poll it.
                 match ready!(incoming.poll_recv(cx)) {
-                    Some(Ok(connecting)) => {
-                        let peer = connecting.remote_address();
+                    Some(Ok(incoming)) => {
+                        let peer = incoming.remote_address();
 
                         debug!("New incoming connection from {}", peer);
 
                         // Return a future that resolves to the output.
                         return Poll::Ready(Box::pin(async move {
                             debug!(client = %peer, "Accepting connection...");
-                            let connection = connecting.await?;
+                            let connection = incoming.accept()?.await?;
                             debug!(
                                 "Accepted connection from {}, opening stream",
                                 connection.remote_address()

msg-transport/src/quic/mod.rs

@@ -5,7 +5,7 @@ use quinn::{
     rustls::{
         self, SignatureScheme,
         client::danger::{ServerCertVerified, ServerCertVerifier},
-        pki_types::{CertificateDer, PrivateKeyDer},
+        pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer},
     },
 };
 
@@ -81,7 +81,7 @@ pub(crate) fn unsafe_client_config() -> QuicClientConfig {
 
     let mut rustls_config = rustls::ClientConfig::builder_with_provider(provider)
         .with_protocol_versions(&[&rustls::version::TLS13])
-        .expect("aws_lc_rs provider supports TLS 1.3")
+        .expect("ring provider supports TLS 1.3")
         .dangerous()
         .with_custom_certificate_verifier(SkipServerVerification::new())
         .with_no_client_auth();
@@ -100,7 +100,7 @@ pub(crate) fn tls_server_config() -> QuicServerConfig {
 
     let mut rustls_config = rustls::ServerConfig::builder_with_provider(provider)
         .with_protocol_versions(&[&rustls::version::TLS13])
-        .expect("aws_lc_rs provider supports TLS 1.3")
+        .expect("ring provider supports TLS 1.3")
         .with_no_client_auth()
         .with_single_cert(cert_chain, key_der)
         .expect("Valid rustls config");
@@ -113,12 +113,13 @@ pub(crate) fn tls_server_config() -> QuicServerConfig {
 
 /// Generates a self-signed certificate chain and private key.
 pub(crate) fn self_signed_certificate() -> (Vec<CertificateDer<'static>>, PrivateKeyDer<'static>) {
-    let cert = rcgen::generate_simple_self_signed(vec![]).expect("Generates valid certificate");
-    let cert_der = cert.serialize_der().expect("Serializes certificate");
-    let priv_key =
-        PrivateKeyDer::try_from(cert.serialize_private_key_der()).expect("Serializes private key");
+    let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_string()])
+        .expect("Generates valid certificate");
 
-    (vec![CertificateDer::from(cert_der)], priv_key)
+    let cert_der = CertificateDer::from(cert.cert);
+    let priv_key = PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der());
+
+    (vec![CertificateDer::from(cert_der)], priv_key.into())
 }
 
 #[cfg(test)]