squashing before rebasing

Cargo.lock

@@ -1275,7 +1275,7 @@ dependencies = [
  "openssl-probe",
  "openssl-sys",
  "schannel",
- "socket2",
+ "socket2 0.4.9",
  "winapi",
 ]
 
@@ -2518,7 +2518,7 @@ dependencies = [
  "httpdate",
  "itoa",
  "pin-project-lite",
- "socket2",
+ "socket2 0.4.9",
  "tokio",
  "tower-service",
  "tracing",
@@ -2537,7 +2537,7 @@ dependencies = [
  "http",
  "hyper",
  "hyper-rustls",
- "rustls-native-certs",
+ "rustls-native-certs 0.5.0",
  "tokio",
  "tokio-rustls 0.22.0",
  "tower-service",
@@ -2555,7 +2555,7 @@ dependencies = [
  "hyper",
  "log",
  "rustls 0.19.1",
- "rustls-native-certs",
+ "rustls-native-certs 0.5.0",
  "tokio",
  "tokio-rustls 0.22.0",
  "webpki 0.21.4",
@@ -3804,6 +3804,7 @@ dependencies = [
  "atty",
  "bip39",
  "bs58",
+ "bytes",
  "clap 4.2.7",
  "colored",
  "dashmap 4.0.2",
@@ -3831,7 +3832,10 @@ dependencies = [
  "nym-validator-client",
  "once_cell",
  "pretty_env_logger",
+ "quinn",
  "rand 0.7.3",
+ "rcgen",
+ "rustls 0.21.2",
  "serde",
  "serde_json",
  "sqlx 0.5.13",
@@ -3919,10 +3923,13 @@ dependencies = [
 name = "nym-mixnet-client"
 version = "0.1.0"
 dependencies = [
+ "bytes",
  "futures",
  "log",
  "nym-sphinx",
  "nym-task",
+ "quinn",
+ "rustls 0.21.2",
  "tokio",
  "tokio-util",
 ]
@@ -3952,6 +3959,7 @@ version = "1.1.23"
 dependencies = [
  "anyhow",
  "bs58",
+ "bytes",
  "cfg-if",
  "clap 4.2.7",
  "colored",
@@ -3979,8 +3987,11 @@ dependencies = [
  "nym-validator-client",
  "opentelemetry",
  "pretty_env_logger",
+ "quinn",
  "rand 0.7.3",
+ "rcgen",
  "rocket",
+ "rustls 0.21.2",
  "serde",
  "serde_json",
  "sysinfo",
@@ -4197,7 +4208,7 @@ dependencies = [
 name = "nym-pemstore"
 version = "0.3.0"
 dependencies = [
- "pem",
+ "pem 0.8.3",
 ]
 
 [[package]]
@@ -5060,6 +5071,16 @@ dependencies = [
  "regex",
 ]
 
+[[package]]
+name = "pem"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6b13fe415cdf3c8e44518e18a7c95a13431d9bdf6d15367d82b23c377fdd441a"
+dependencies = [
+ "base64 0.21.2",
+ "serde",
+]
+
 [[package]]
 name = "percent-encoding"
 version = "2.2.0"
@@ -5399,6 +5420,54 @@ version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3"
 
+[[package]]
+name = "quinn"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "21252f1c0fc131f1b69182db8f34837e8a69737b8251dff75636a9be0518c324"
+dependencies = [
+ "bytes",
+ "pin-project-lite",
+ "quinn-proto",
+ "quinn-udp",
+ "rustc-hash",
+ "rustls 0.21.2",
+ "thiserror",
+ "tokio",
+ "tracing",
+]
+
+[[package]]
+name = "quinn-proto"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85af4ed6ee5a89f26a26086e9089a6643650544c025158449a3626ebf72884b3"
+dependencies = [
+ "bytes",
+ "rand 0.8.5",
+ "ring",
+ "rustc-hash",
+ "rustls 0.21.2",
+ "rustls-native-certs 0.6.3",
+ "slab",
+ "thiserror",
+ "tinyvec",
+ "tracing",
+]
+
+[[package]]
+name = "quinn-udp"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6df19e284d93757a9fb91d63672f7741b129246a669db09d1c0063071debc0c0"
+dependencies = [
+ "bytes",
+ "libc",
+ "socket2 0.5.3",
+ "tracing",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "quote"
 version = "1.0.27"
@@ -5641,6 +5710,18 @@ dependencies = [
  "num_cpus",
 ]
 
+[[package]]
+name = "rcgen"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4954fbc00dcd4d8282c987710e50ba513d351400dbdd00e803a05172a90d8976"
+dependencies = [
+ "pem 2.0.1",
+ "ring",
+ "time 0.3.21",
+ "yasna",
+]
+
 [[package]]
 name = "rdrand"
 version = "0.4.0"
@@ -5932,6 +6013,12 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "rustc-hash"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2"
+
 [[package]]
 name = "rustc_version"
 version = "0.2.3"
@@ -5989,6 +6076,18 @@ dependencies = [
  "webpki 0.22.0",
 ]
 
+[[package]]
+name = "rustls"
+version = "0.21.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e32ca28af694bc1bbf399c33a516dbdf1c90090b8ab23c2bc24f834aa2247f5f"
+dependencies = [
+ "log",
+ "ring",
+ "rustls-webpki",
+ "sct 0.7.0",
+]
+
 [[package]]
 name = "rustls-native-certs"
 version = "0.5.0"
@@ -6001,6 +6100,18 @@ dependencies = [
  "security-framework",
 ]
 
+[[package]]
+name = "rustls-native-certs"
+version = "0.6.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00"
+dependencies = [
+ "openssl-probe",
+ "rustls-pemfile",
+ "schannel",
+ "security-framework",
+]
+
 [[package]]
 name = "rustls-pemfile"
 version = "1.0.2"
@@ -6010,6 +6121,16 @@ dependencies = [
  "base64 0.21.2",
 ]
 
+[[package]]
+name = "rustls-webpki"
+version = "0.100.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6207cd5ed3d8dca7816f8f3725513a34609c0c765bf652b8c3cb4cfd87db46b"
+dependencies = [
+ "ring",
+ "untrusted",
+]
+
 [[package]]
 name = "rustversion"
 version = "1.0.12"
@@ -6483,6 +6604,16 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "socket2"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2538b18701741680e0322a2302176d3253a35388e2e62f172f64f4f16605f877"
+dependencies = [
+ "libc",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "sphinx-packet"
 version = "0.1.0"
@@ -7132,7 +7263,7 @@ dependencies = [
  "parking_lot 0.12.1",
  "pin-project-lite",
  "signal-hook-registry",
- "socket2",
+ "socket2 0.4.9",
  "tokio-macros",
  "tracing",
  "windows-sys 0.48.0",
@@ -8215,6 +8346,15 @@ version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09041cd90cf85f7f8b2df60c646f853b7f535ce68f85244eb6731cf89fa498ec"
 
+[[package]]
+name = "yasna"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd"
+dependencies = [
+ "time 0.3.21",
+]
+
 [[package]]
 name = "zeroize"
 version = "1.6.0"

common/bin-common/src/logging/mod.rs

@@ -40,6 +40,8 @@ pub fn setup_logging() {
         .filter_module("tokio_tungstenite", log::LevelFilter::Warn)
         .filter_module("handlebars", log::LevelFilter::Warn)
         .filter_module("sled", log::LevelFilter::Warn)
+        .filter_module("quinn", log::LevelFilter::Warn)
+        .filter_module("rustls", log::LevelFilter::Warn)
         .init();
 }
 

common/client-libs/mixnet-client/Cargo.toml

@@ -11,6 +11,9 @@ futures = "0.3"
 log = { workspace = true }
 tokio = { version = "1.24.1", features = ["time", "net", "rt"] }
 tokio-util = { version = "0.7.4", features = ["codec"] }
+rustls = {version = "0.21.1", features = ["dangerous_configuration", "quic"]}
+quinn = "0.10.1"
+bytes = "*"
 
 # internal
 nym-sphinx = { path = "../../nymsphinx" }

common/client-libs/mixnet-client/src/client.rs

@@ -2,22 +2,23 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use futures::channel::mpsc;
-use futures::StreamExt;
+use futures::{SinkExt, StreamExt};
 use log::*;
 use nym_sphinx::addressing::nodes::NymNodeRoutingAddress;
 use nym_sphinx::framing::codec::NymCodec;
 use nym_sphinx::framing::packet::FramedNymPacket;
 use nym_sphinx::params::PacketType;
 use nym_sphinx::NymPacket;
+use quinn::{ClientConfig, Endpoint};
 use std::collections::HashMap;
 use std::io;
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
-use tokio::net::TcpStream;
+
 use tokio::time::sleep;
-use tokio_util::codec::Framed;
+use tokio_util::codec::FramedWrite;
 
 pub struct Config {
     initial_reconnection_backoff: Duration,
@@ -64,6 +65,7 @@ pub struct Client {
 struct ConnectionSender {
     channel: mpsc::Sender<FramedNymPacket>,
     current_reconnection_attempt: Arc<AtomicU32>,
+    last_used: Instant,
 }
 
 impl ConnectionSender {
@@ -71,6 +73,7 @@ impl ConnectionSender {
         ConnectionSender {
             channel,
             current_reconnection_attempt: Arc::new(AtomicU32::new(0)),
+            last_used: Instant::now(),
         }
     }
 }
@@ -85,19 +88,25 @@ impl Client {
 
     async fn manage_connection(
         address: SocketAddr,
-        receiver: mpsc::Receiver<FramedNymPacket>,
+        mut receiver: mpsc::Receiver<FramedNymPacket>,
         connection_timeout: Duration,
         current_reconnection: &AtomicU32,
     ) {
-        let connection_fut = TcpStream::connect(address);
+        let mut endpoint = Endpoint::client("0.0.0.0:0".parse::<SocketAddr>().unwrap()).unwrap();
+        endpoint.set_default_client_config(configure_client());
 
-        let conn = match tokio::time::timeout(connection_timeout, connection_fut).await {
+        let conn = match tokio::time::timeout(
+            connection_timeout,
+            endpoint.connect(address, "mixnode").unwrap(),
+        )
+        .await
+        {
             Ok(stream_res) => match stream_res {
-                Ok(stream) => {
+                Ok(connection) => {
                     debug!("Managed to establish connection to {}", address);
                     // if we managed to connect, reset the reconnection count (whatever it might have been)
                     current_reconnection.store(0, Ordering::Release);
-                    Framed::new(stream, NymCodec)
+                    connection
                 }
                 Err(err) => {
                     debug!(
@@ -118,18 +127,26 @@ impl Client {
                 return;
             }
         };
-
+        loop {
-        // Take whatever the receiver channel produces and put it on the connection.
+            let pkt = match receiver.next().await {
-        // We could have as well used conn.send_all(receiver.map(Ok)), but considering we don't care
+                Some(pkt) => pkt,
-        // about neither receiver nor the connection, it doesn't matter which one gets consumed
+                None => {
-        if let Err(err) = receiver.map(Ok).forward(conn).await {
+                    debug!("No more packet to send to {}", address);
-            warn!("Failed to forward packets to {} - {err}", address);
+                    return;
+                }
+            };
+            let send = match conn.open_uni().await {
+                Ok(send_stream) => send_stream,
+                Err(err) => {
+                    error!("Failed to open uni stream, dropping packet - {err:?}");
+                    return; //We shouldn't get a time out here, it should be handled higher
+                }
+            };
+            let mut framed_stream = FramedWrite::new(send, NymCodec);
+            if let Err(err) = framed_stream.send(pkt).await {
+                warn!("Failed to forward packets to {} - {err}", address);
+            }
         }
-
-        debug!(
-            "connection manager to {} is finished. Either the connection failed or mixnet client got dropped",
-            address
-        );
     }
 
     /// If we're trying to reconnect, determine how long we should wait.
@@ -205,36 +222,53 @@ impl SendWithoutResponse for Client {
             FramedNymPacket::new(packet, packet_type, self.config.use_legacy_version);
 
         if let Some(sender) = self.conn_new.get_mut(&address) {
-            if let Err(err) = sender.channel.try_send(framed_packet) {
+            if sender.last_used.elapsed().as_millis() > 9_000 {
-                if err.is_full() {
+                // default timeout is 10sec, let's take some margin for the operation to run.
-                    debug!("Connection to {} seems to not be able to handle all the traffic - dropping the current packet", address);
+                //connection is near timeout, let's just recreate one
-                    // it's not a 'big' error, but we did not manage to send the packet
+                sender.channel.close_channel();
-                    // if the queue is full, we can't really do anything but to drop the packet
+                self.conn_new.remove(&address);
-                    Err(io::Error::new(
+                debug!(
-                        io::ErrorKind::WouldBlock,
+                    "connection near or past timemout. Reconnecting to {}",
-                        "connection queue is full",
+                    address
-                    ))
+                );
-                } else if err.is_disconnected() {
+                self.make_connection(address, framed_packet);
-                    debug!(
+                Err(io::Error::new(
-                        "Connection to {} seems to be dead. attempting to re-establish it...",
+                    io::ErrorKind::NotConnected,
-                        address
+                    "re-connection is in progress",
-                    );
+                ))
-                    // it's not a 'big' error, but we did not manage to send the packet, but queue
-                    // it up to send it as soon as the connection is re-established
-                    self.make_connection(address, err.into_inner());
-                    Err(io::Error::new(
-                        io::ErrorKind::ConnectionAborted,
-                        "reconnection attempt is in progress",
-                    ))
-                } else {
-                    // this can't really happen, but let's safe-guard against it in case something changes in futures library
-                    Err(io::Error::new(
-                        io::ErrorKind::Other,
-                        "unknown connection buffer error",
-                    ))
-                }
             } else {
-                Ok(())
+                sender.last_used = Instant::now();
+                if let Err(err) = sender.channel.try_send(framed_packet) {
+                    if err.is_full() {
+                        debug!("Connection to {} seems to not be able to handle all the traffic - dropping the current packet", address);
+                        // it's not a 'big' error, but we did not manage to send the packet
+                        // if the queue is full, we can't really do anything but to drop the packet
+                        Err(io::Error::new(
+                            io::ErrorKind::WouldBlock,
+                            "connection queue is full",
+                        ))
+                    } else if err.is_disconnected() {
+                        debug!(
+                            "Connection to {} seems to be dead. attempting to re-establish it...",
+                            address
+                        );
+                        // it's not a 'big' error, but we did not manage to send the packet, but queue
+                        // it up to send it as soon as the connection is re-established
+                        self.make_connection(address, err.into_inner());
+                        Err(io::Error::new(
+                            io::ErrorKind::ConnectionAborted,
+                            "reconnection attempt is in progress",
+                        ))
+                    } else {
+                        // this can't really happen, but let's safe-guard against it in case something changes in futures library
+                        Err(io::Error::new(
+                            io::ErrorKind::Other,
+                            "unknown connection buffer error",
+                        ))
+                    }
+                } else {
+                    Ok(())
+                }
             }
         } else {
             // there was never a connection to begin with
@@ -292,3 +326,35 @@ mod tests {
         );
     }
 }
+
+// Implementation of `ServerCertVerifier` that verifies everything as trustworthy.
+struct SkipServerVerification;
+
+impl SkipServerVerification {
+    fn new() -> Arc<Self> {
+        Arc::new(Self)
+    }
+}
+
+impl rustls::client::ServerCertVerifier for SkipServerVerification {
+    fn verify_server_cert(
+        &self,
+        _end_entity: &rustls::Certificate,
+        _intermediates: &[rustls::Certificate],
+        _server_name: &rustls::ServerName,
+        _scts: &mut dyn Iterator<Item = &[u8]>,
+        _ocsp_response: &[u8],
+        _now: std::time::SystemTime,
+    ) -> Result<rustls::client::ServerCertVerified, rustls::Error> {
+        Ok(rustls::client::ServerCertVerified::assertion())
+    }
+}
+
+fn configure_client() -> ClientConfig {
+    let crypto = rustls::ClientConfig::builder()
+        .with_safe_defaults()
+        .with_custom_certificate_verifier(SkipServerVerification::new())
+        .with_no_client_auth();
+
+    ClientConfig::new(Arc::new(crypto))
+}

gateway/Cargo.toml

@@ -43,6 +43,10 @@ tokio-tungstenite = "0.14"
 tokio-util = { version = "0.7.4", features = ["codec"] }
 url = { version = "2.2", features = ["serde"] }
 zeroize = { workspace = true }
+quinn = "0.10.1"
+rustls = {version = "0.21.1", features = ["quic"]}
+rcgen = "*"
+bytes = "*"
 
 # internal
 nym-api-requests = { path = "../nym-api/nym-api-requests" }

gateway/src/node/mixnet_handling/receiver/connection_handler.rs

@@ -15,10 +15,9 @@ use nym_sphinx::framing::codec::NymCodec;
 use nym_sphinx::framing::packet::FramedNymPacket;
 use nym_sphinx::DestinationAddressBytes;
 use nym_task::TaskClient;
+use quinn::{Connection, ConnectionError};
 use std::collections::HashMap;
-use std::net::SocketAddr;
+use tokio_util::codec::FramedRead;
-use tokio::net::TcpStream;
-use tokio_util::codec::Framed;
 
 pub(crate) struct ConnectionHandler<St: Storage> {
     packet_processor: PacketProcessor,
@@ -174,23 +173,37 @@ impl<St: Storage> ConnectionHandler<St> {
         self.handle_processed_packet(processed_final_hop).await
     }
 
-    pub(crate) async fn handle_connection(
+    pub(crate) async fn handle_connection(mut self, conn: Connection, mut shutdown: TaskClient) {
-        mut self,
+        debug!(
-        conn: TcpStream,
+            "Starting connection handler for {:?}",
-        remote: SocketAddr,
+            conn.remote_address()
-        mut shutdown: TaskClient,
+        );
-    ) {
-        debug!("Starting connection handler for {:?}", remote);
         shutdown.mark_as_success();
-        let mut framed_conn = Framed::new(conn, NymCodec);
         while !shutdown.is_shutdown() {
             tokio::select! {
                 biased;
                 _ = shutdown.recv() => {
                     log::trace!("ConnectionHandler: received shutdown");
                 }
-                framed_sphinx_packet = framed_conn.next() => {
+                recv = conn.accept_uni() => {
-                    match framed_sphinx_packet {
+                    let recv_stream = match recv {
+                        Ok(recv_stream) => recv_stream,
+                        Err(err) => {
+                            match err {
+                                ConnectionError::TimedOut => {
+                                    //normal timeout, we just need to drop the connection
+                                    break;
+                                },
+                                _ => {
+                                    error!("Error accepting uni stream - {err:?}");
+                                    break;
+                                },
+                            }
+                        }
+
+                    };
+                    let mut framed_stream = FramedRead::new(recv_stream, NymCodec);
+                    match framed_stream.next().await {
                         Some(Ok(framed_sphinx_packet)) => {
                             // TODO: benchmark spawning tokio task with full processing vs just processing it
                             // synchronously under higher load in single and multi-threaded situation.
@@ -212,9 +225,6 @@ impl<St: Storage> ConnectionHandler<St> {
             }
         }
 
-        info!(
+        info!("Closing connection from {:?}", conn.remote_address());
-            "Closing connection from {:?}",
-            framed_conn.into_inner().peer_addr()
-        );
     }
 }

gateway/src/node/mixnet_handling/receiver/listener.rs

@@ -5,8 +5,10 @@ use crate::node::mixnet_handling::receiver::connection_handler::ConnectionHandle
 use crate::node::storage::Storage;
 use log::*;
 use nym_task::TaskClient;
+use quinn::{Endpoint, ServerConfig};
+use rcgen::generate_simple_self_signed;
+use rustls::{Certificate, PrivateKey};
 use std::net::SocketAddr;
-use std::process;
 use tokio::task::JoinHandle;
 
 pub(crate) struct Listener {
@@ -25,13 +27,7 @@ impl Listener {
         St: Storage + Clone + 'static,
     {
         info!("Starting mixnet listener at {}", self.address);
-        let tcp_listener = match tokio::net::TcpListener::bind(self.address).await {
+        let endpoint = Endpoint::server(server_config(), self.address).unwrap();
-            Ok(listener) => listener,
-            Err(err) => {
-                error!("Failed to bind to {} - {err}. Are you sure nothing else is running on the specified port and your user has sufficient permission to bind to the requested address?", self.address);
-                process::exit(1);
-            }
-        };
 
         while !self.shutdown.is_shutdown() {
             tokio::select! {
@@ -39,13 +35,22 @@ impl Listener {
                 _ = self.shutdown.recv() => {
                     log::trace!("mixnet_handling::Listener: Received shutdown");
                 }
-                connection = tcp_listener.accept() => {
+                connection = endpoint.accept() => {
                     match connection {
-                        Ok((socket, remote_addr)) => {
+                        Some(connecting) => {
-                            let handler = connection_handler.clone();
+                            match connecting.await {
-                            tokio::spawn(handler.handle_connection(socket, remote_addr, self.shutdown.clone()));
+                                Ok(conn) => {
+                                    debug!("Handling connection from {:?}", conn.remote_address());
+                                    let handler = connection_handler.clone();
+                                    tokio::spawn(handler.handle_connection(conn, self.shutdown.clone()));
+                                },
+                                Err(err) => error!("Failed to establish connection - {err:?}"),
+                            }
                         }
-                        Err(err) => warn!("failed to get client: {err}"),
+                        None => {
+                            error!("Endpoint closed");
+                            break;
+                        }, // stream got closed by remote
                     }
                 }
             }
@@ -61,3 +66,13 @@ impl Listener {
         tokio::spawn(async move { self.run(connection_handler).await })
     }
 }
+
+fn generate_self_signed_cert() -> Result<(Certificate, PrivateKey), Box<dyn std::error::Error>> {
+    let cert = generate_simple_self_signed(vec!["mixnode".to_string()])?;
+    let key = PrivateKey(cert.serialize_private_key_der());
+    Ok((Certificate(cert.serialize_der()?), key))
+}
+fn server_config() -> ServerConfig {
+    let (cert, key) = generate_self_signed_cert().expect("Failed to generate certificate");
+    ServerConfig::with_single_cert(vec![cert], key).expect("Failed to generate server config")
+}

mixnode/Cargo.toml

@@ -33,10 +33,14 @@ serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 sysinfo = "0.27.7"
 tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal"] }
-tokio-util = { version = "0.7.3", features = ["codec"] }
+tokio-util = { version = "0.7.3", features = ["codec", "net"] }
 toml = "0.5.8"
 url = { version = "2.2", features = ["serde"] }
 cfg-if = "1.0.0"
+quinn = "0.10.1"
+rustls = {version = "0.21.1", features = ["quic"]}
+rcgen = "*"
+bytes = "*"
 
 ## tracing
 tracing = { version = "0.1.37", optional = true }

mixnode/src/node/listener/connection_handler/mod.rs

@@ -12,10 +12,10 @@ use nym_sphinx::forwarding::packet::MixPacket;
 use nym_sphinx::framing::codec::NymCodec;
 use nym_sphinx::framing::packet::FramedNymPacket;
 use nym_sphinx::Delay as SphinxDelay;
-use std::net::SocketAddr;
+use quinn::{Connection, ConnectionError};
-use tokio::net::TcpStream;
 use tokio::time::Instant;
-use tokio_util::codec::Framed;
+use tokio_util::codec::FramedRead;
+
 #[cfg(feature = "cpucycles")]
 use tracing::{error, info, instrument};
 
@@ -54,7 +54,7 @@ impl ConnectionHandler {
         feature = "cpucycles",
         instrument(skip(self, framed_sphinx_packet), fields(cpucycles))
     )]
-    fn handle_received_packet(&self, framed_sphinx_packet: FramedNymPacket) {
+    async fn handle_received_packet(&self, framed_sphinx_packet: FramedNymPacket) {
         //
         // TODO: here be replay attack detection - it will require similar key cache to the one in
         // packet processor for vpn packets,
@@ -78,37 +78,51 @@ impl ConnectionHandler {
         })
     }
 
-    pub(crate) async fn handle_connection(
+    pub(crate) async fn handle_connection(self, conn: Connection, mut shutdown: TaskClient) {
-        self,
+        debug!(
-        conn: TcpStream,
+            "Starting connection handler for {:?}",
-        remote: SocketAddr,
+            conn.remote_address()
-        mut shutdown: TaskClient,
+        );
-    ) {
-        debug!("Starting connection handler for {:?}", remote);
         shutdown.mark_as_success();
-        let mut framed_conn = Framed::new(conn, NymCodec);
         while !shutdown.is_shutdown() {
             tokio::select! {
                 biased;
                 _ = shutdown.recv() => {
                     log::trace!("ConnectionHandler: received shutdown");
                 }
-                framed_sphinx_packet = framed_conn.next() => {
+
-                    match framed_sphinx_packet {
+                recv = conn.accept_uni() => {
+                    let recv_stream = match recv {
+                        Ok(recv_stream) => recv_stream,
+                        Err(err) => {
+                            match err {
+                                ConnectionError::TimedOut => {
+                                    //normal timeout, we just need to drop the connection
+                                    break;
+                                },
+                                _ => {
+                                    error!("Error accepting uni stream - {err:?}");
+                                    break;
+                                },
+                            }
+                        }
+
+                    };
+
+                    let mut framed_stream = FramedRead::new(recv_stream, NymCodec);
+                    match framed_stream.next().await {
                         Some(Ok(framed_sphinx_packet)) => {
                             // TODO: benchmark spawning tokio task with full processing vs just processing it
-                            // synchronously (without delaying inside of course,
+                            // synchronously under higher load in single and multi-threaded situation.
-                            // delay is moved to a global DelayQueue)
-                            // under higher load in single and multi-threaded situation.
 
                             // in theory we could process multiple sphinx packet from the same connection in parallel,
                             // but we already handle multiple concurrent connections so if anything, making
                             // that change would only slow things down
-                            self.handle_received_packet(framed_sphinx_packet);
+                            self.handle_received_packet(framed_sphinx_packet).await;
                         }
                         Some(Err(err)) => {
                             error!(
-                                "{remote:?} - The socket connection got corrupted with error: {err}. Closing the socket",
+                                "The socket connection got corrupted with error: {err}. Closing the socket",
                             );
                             return;
                         }
@@ -118,10 +132,7 @@ impl ConnectionHandler {
             }
         }
 
-        info!(
+        info!("Closing connection from {:?}", conn.remote_address());
-            "Closing connection from {:?}",
-            framed_conn.into_inner().peer_addr()
-        );
         log::trace!("ConnectionHandler: Exiting");
     }
 }

mixnode/src/node/listener/mod.rs

@@ -2,9 +2,10 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::node::listener::connection_handler::ConnectionHandler;
+use quinn::{Endpoint, ServerConfig};
+use rcgen::generate_simple_self_signed;
+use rustls::{Certificate, PrivateKey};
 use std::net::SocketAddr;
-use std::process;
-use tokio::net::TcpListener;
 use tokio::task::JoinHandle;
 #[cfg(feature = "cpucycles")]
 use tracing::error;
@@ -25,27 +26,36 @@ impl Listener {
 
     async fn run(&mut self, connection_handler: ConnectionHandler) {
         log::trace!("Starting Listener");
-        let listener = match TcpListener::bind(self.address).await {
+        let endpoint = Endpoint::server(server_config(), self.address).unwrap();
-            Ok(listener) => listener,
-            Err(err) => {
-                error!("Failed to bind to {} - {err}. Are you sure nothing else is running on the specified port and your user has sufficient permission to bind to the requested address?", self.address);
-                process::exit(1);
-            }
-        };
 
         while !self.shutdown.is_shutdown() {
             tokio::select! {
                 biased;
                 _ = self.shutdown.recv() => {
                     log::trace!("Listener: Received shutdown");
-                }
+                },
-                connection = listener.accept() => {
+                connection = endpoint.accept() => {
                     match connection {
-                        Ok((socket, remote_addr)) => {
+                        Some(connecting) => {
-                            let handler = connection_handler.clone();
+                           match connecting.await {
-                            tokio::spawn(handler.handle_connection(socket, remote_addr, self.shutdown.clone()));
+                                Ok(conn) => {
+                                    debug!("Handling connection from {:?}", conn.remote_address());
+                                    let handler = connection_handler.clone();
+                                    tokio::spawn(handler.handle_connection(conn, self.shutdown.clone()));
+                                },
+                                Err(err) => error!("Failed to establish connection - {err:?}"),
+                            }
                         }
-                        Err(err) => warn!("Failed to accept incoming connection - {err}"),
+                        // Some(Err(err)) => {
+                        //     error!(
+                        //         "The socket connection got corrupted with error: {err}. Closing the socket",
+                        //     );
+                        //     return;
+                        // }
+                        None => {
+                            error!("Endpoint closed");
+                            break;
+                        }, // stream got closed by remote
                     }
                 },
             };
@@ -59,3 +69,14 @@ impl Listener {
         tokio::spawn(async move { self.run(connection_handler).await })
     }
 }
+
+fn generate_self_signed_cert(
+) -> Result<(rustls::Certificate, rustls::PrivateKey), Box<dyn std::error::Error>> {
+    let cert = generate_simple_self_signed(vec!["mixnode".to_string()])?;
+    let key = PrivateKey(cert.serialize_private_key_der());
+    Ok((Certificate(cert.serialize_der()?), key))
+}
+fn server_config() -> ServerConfig {
+    let (cert, key) = generate_self_signed_cert().expect("Failed to generate certificate");
+    ServerConfig::with_single_cert(vec![cert], key).expect("Failed to generate server config")
+}