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,23 +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 tokio::net::{TcpStream, UdpSocket};
+use std::time::{Duration, Instant};
+
 use tokio::time::sleep;
-use tokio_util::codec::Framed;
-use tokio_util::udp::UdpFramed;
+use tokio_util::codec::FramedWrite;
 
 pub struct Config {
     initial_reconnection_backoff: Duration,
@@ -58,98 +58,152 @@ pub trait SendWithoutResponse {
 }
 
 pub struct Client {
-    conn_new: Option<mpsc::Sender<(FramedNymPacket, SocketAddr)>>,
+    conn_new: HashMap<NymNodeRoutingAddress, ConnectionSender>,
     config: Config,
 }
 
+struct ConnectionSender {
+    channel: mpsc::Sender<FramedNymPacket>,
+    current_reconnection_attempt: Arc<AtomicU32>,
+    last_used: Instant,
+}
+
+impl ConnectionSender {
+    fn new(channel: mpsc::Sender<FramedNymPacket>) -> Self {
+        ConnectionSender {
+            channel,
+            current_reconnection_attempt: Arc::new(AtomicU32::new(0)),
+            last_used: Instant::now(),
+        }
+    }
+}
+
 impl Client {
     pub fn new(config: Config) -> Client {
         Client {
-            conn_new: None,
+            conn_new: HashMap::new(),
             config,
         }
     }
 
     async fn manage_connection(
         address: SocketAddr,
-        receiver: mpsc::Receiver<(FramedNymPacket, SocketAddr)>,
+        mut receiver: mpsc::Receiver<FramedNymPacket>,
+        connection_timeout: Duration,
+        current_reconnection: &AtomicU32,
     ) {
-        let socket = match UdpSocket::bind("0.0.0.0:0").await {
-            Ok(socket) => socket,
-            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?");
+        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,
+            endpoint.connect(address, "mixnode").unwrap(),
+        )
+        .await
+        {
+            Ok(stream_res) => match stream_res {
+                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);
+                    connection
+                }
+                Err(err) => {
+                    debug!(
+                        "failed to establish connection to {} (err: {})",
+                        address, err
+                    );
+                    return;
+                }
+            },
+            Err(_) => {
+                debug!(
+                    "failed to connect to {} within {:?}",
+                    address, connection_timeout
+                );
+
+                // we failed to connect - increase reconnection attempt
+                current_reconnection.fetch_add(1, Ordering::SeqCst);
                 return;
             }
         };
-        let framed_conn = UdpFramed::new(socket, NymCodec);
-        // Take whatever the receiver channel produces and put it on the connection.
-        // We could have as well used conn.send_all(receiver.map(Ok)), but considering we don't care
-        // about neither receiver nor the connection, it doesn't matter which one gets consumed
-        if let Err(err) = receiver.map(Ok).forward(framed_conn).await {
-            warn!("Failed to forward packets to {} - {err}", address);
+        loop {
+            let pkt = match receiver.next().await {
+                Some(pkt) => pkt,
+                None => {
+                    debug!("No more packet to send to {}", 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 is finished. Either the connection failed or mixnet client got dropped"
-        );
     }
 
     /// If we're trying to reconnect, determine how long we should wait.
-    // fn determine_backoff(&self, current_attempt: u32) -> Option<Duration> {
-    //     if current_attempt == 0 {
-    //         None
-    //     } else {
-    //         let exp = 2_u32.checked_pow(current_attempt);
-    //         let backoff = exp
-    //             .and_then(|exp| self.config.initial_reconnection_backoff.checked_mul(exp))
-    //             .unwrap_or(self.config.maximum_reconnection_backoff);
+    fn determine_backoff(&self, current_attempt: u32) -> Option<Duration> {
+        if current_attempt == 0 {
+            None
+        } else {
+            let exp = 2_u32.checked_pow(current_attempt);
+            let backoff = exp
+                .and_then(|exp| self.config.initial_reconnection_backoff.checked_mul(exp))
+                .unwrap_or(self.config.maximum_reconnection_backoff);
 
-    //         Some(std::cmp::min(
-    //             backoff,
-    //             self.config.maximum_reconnection_backoff,
-    //         ))
-    //     }
-    // }
+            Some(std::cmp::min(
+                backoff,
+                self.config.maximum_reconnection_backoff,
+            ))
+        }
+    }
 
     fn make_connection(&mut self, address: NymNodeRoutingAddress, pending_packet: FramedNymPacket) {
         let (mut sender, receiver) = mpsc::channel(self.config.maximum_connection_buffer_size);
 
         // this CAN'T fail because we just created the channel which has a non-zero capacity
         if self.config.maximum_connection_buffer_size > 0 {
-            sender.try_send((pending_packet, address.into())).unwrap();
+            sender.try_send(pending_packet).unwrap();
         }
-        self.conn_new = Some(sender);
 
         // if we already tried to connect to `address` before, grab the current attempt count
-        // let current_reconnection_attempt = if let Some(existing) = self.conn_new.get_mut(&address) {
-        //     existing.channel = sender;
-        //     Arc::clone(&existing.current_reconnection_attempt)
-        // } else {
-        //     let new_entry = ConnectionSender::new(sender);
-        //     let current_attempt = Arc::clone(&new_entry.current_reconnection_attempt);
-        //     self.conn_new.insert(address, new_entry);
-        //     current_attempt
-        // };
+        let current_reconnection_attempt = if let Some(existing) = self.conn_new.get_mut(&address) {
+            existing.channel = sender;
+            Arc::clone(&existing.current_reconnection_attempt)
+        } else {
+            let new_entry = ConnectionSender::new(sender);
+            let current_attempt = Arc::clone(&new_entry.current_reconnection_attempt);
+            self.conn_new.insert(address, new_entry);
+            current_attempt
+        };
 
         // load the actual value.
-        // let reconnection_attempt = current_reconnection_attempt.load(Ordering::Acquire);
-        // let backoff = self.determine_backoff(reconnection_attempt);
+        let reconnection_attempt = current_reconnection_attempt.load(Ordering::Acquire);
+        let backoff = self.determine_backoff(reconnection_attempt);
 
         // copy the value before moving into another task
-        // let initial_connection_timeout = self.config.initial_connection_timeout;
+        let initial_connection_timeout = self.config.initial_connection_timeout;
 
         tokio::spawn(async move {
             // before executing the manager, wait for what was specified, if anything
-            // if let Some(backoff) = backoff {
-            //     trace!("waiting for {:?} before attempting connection", backoff);
-            //     sleep(backoff).await;
-            // }
+            if let Some(backoff) = backoff {
+                trace!("waiting for {:?} before attempting connection", backoff);
+                sleep(backoff).await;
+            }
 
             Self::manage_connection(
                 address.into(),
                 receiver,
-                //initial_connection_timeout,
-                //&current_reconnection_attempt,
+                initial_connection_timeout,
+                &current_reconnection_attempt,
             )
             .await
         });
@@ -167,9 +221,24 @@ impl SendWithoutResponse for Client {
         let framed_packet =
             FramedNymPacket::new(packet, packet_type, self.config.use_legacy_version);
 
-        match &self.conn_new {
-            Some(sender) => {
-                if let Err(err) = sender.clone().try_send((framed_packet, address.into())) {
+        if let Some(sender) = self.conn_new.get_mut(&address) {
+            if sender.last_used.elapsed().as_millis() > 9_000 {
+                // default timeout is 10sec, let's take some margin for the operation to run.
+                //connection is near timeout, let's just recreate one
+                sender.channel.close_channel();
+                self.conn_new.remove(&address);
+                debug!(
+                    "connection near or past timemout. Reconnecting to {}",
+                    address
+                );
+                self.make_connection(address, framed_packet);
+                Err(io::Error::new(
+                    io::ErrorKind::NotConnected,
+                    "re-connection is in progress",
+                ))
+            } else {
+                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
@@ -185,7 +254,7 @@ impl SendWithoutResponse for Client {
                         );
                         // 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().0);
+                        self.make_connection(address, err.into_inner());
                         Err(io::Error::new(
                             io::ErrorKind::ConnectionAborted,
                             "reconnection attempt is in progress",
@@ -198,22 +267,19 @@ impl SendWithoutResponse for Client {
                         ))
                     }
                 } else {
-                    debug!("Sending packet to {:?}", address);
                     Ok(())
                 }
             }
-
-            None => {
-                // there was never a connection to begin with
-                debug!("establishing initial connection");
-                // it's not a 'big' error, but we did not manage to send the packet, but queue the packet
-                // for sending for as soon as the connection is created
-                self.make_connection(address, framed_packet);
-                Err(io::Error::new(
-                    io::ErrorKind::NotConnected,
-                    "connection is in progress",
-                ))
-            }
+        } else {
+            // there was never a connection to begin with
+            debug!("establishing initial connection to {}", address);
+            // it's not a 'big' error, but we did not manage to send the packet, but queue the packet
+            // for sending for as soon as the connection is created
+            self.make_connection(address, framed_packet);
+            Err(io::Error::new(
+                io::ErrorKind::NotConnected,
+                "connection is in progress",
+            ))
         }
     }
 }
@@ -260,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))
+}

common/nymsphinx/params/src/packet_sizes.rs

@@ -26,7 +26,7 @@ const ACK_IV_SIZE: usize = 16;
 
 const ACK_PACKET_SIZE: usize = ACK_IV_SIZE + FRAG_ID_LEN + SPHINX_PACKET_OVERHEAD;
 const REGULAR_PACKET_SIZE: usize = 2 * 1024 + SPHINX_PACKET_OVERHEAD;
-const EXTENDED_PACKET_SIZE_8: usize = 1 * 512 + SPHINX_PACKET_OVERHEAD;
+const EXTENDED_PACKET_SIZE_8: usize = 8 * 1024 + SPHINX_PACKET_OVERHEAD;
 const EXTENDED_PACKET_SIZE_16: usize = 16 * 1024 + SPHINX_PACKET_OVERHEAD;
 const EXTENDED_PACKET_SIZE_32: usize = 32 * 1024 + SPHINX_PACKET_OVERHEAD;
 

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::net::TcpStream;
-use tokio_util::codec::Framed;
+use tokio_util::codec::FramedRead;
 
 pub(crate) struct ConnectionHandler<St: Storage> {
     packet_processor: PacketProcessor,
@@ -155,7 +154,7 @@ impl<St: Storage> ConnectionHandler<St> {
         self.forward_ack(forward_ack, client_address);
     }
 
-    pub(crate) async fn handle_received_packet(&mut self, framed_sphinx_packet: FramedNymPacket) {
+    async fn handle_received_packet(&mut 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,
@@ -174,47 +173,58 @@ impl<St: Storage> ConnectionHandler<St> {
         self.handle_processed_packet(processed_final_hop).await
     }
 
-    // pub(crate) async fn handle_connection(
-    //     mut self,
-    //     conn: TcpStream,
-    //     remote: SocketAddr,
-    //     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 {
-    //                     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.
+    pub(crate) async fn handle_connection(mut self, conn: Connection, mut shutdown: TaskClient) {
+        debug!(
+            "Starting connection handler for {:?}",
+            conn.remote_address()
+        );
+        shutdown.mark_as_success();
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                biased;
+                _ = shutdown.recv() => {
+                    log::trace!("ConnectionHandler: received shutdown");
+                }
+                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;
+                                },
+                            }
+                        }
 
-    //                         // 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).await;
-    //                     }
-    //                     Some(Err(err)) => {
-    //                         error!(
-    //                             "The socket connection got corrupted with error: {err}. Closing the socket",
-    //                         );
-    //                         return;
-    //                     }
-    //                     None => break, // stream got closed by remote
-    //                 }
-    //             }
-    //         }
-    //     }
+                    };
+                    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.
 
-    //     info!(
-    //         "Closing connection from {:?}",
-    //         framed_conn.into_inner().peer_addr()
-    //     );
-    // }
+                            // 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).await;
+                        }
+                        Some(Err(err)) => {
+                            error!(
+                                "The socket connection got corrupted with error: {err}. Closing the socket",
+                            );
+                            return;
+                        }
+                        None => break, // stream got closed by remote
+                    }
+                }
+            }
+        }
+
+        info!("Closing connection from {:?}", conn.remote_address());
+    }
 }

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

@@ -3,14 +3,13 @@
 
 use crate::node::mixnet_handling::receiver::connection_handler::ConnectionHandler;
 use crate::node::storage::Storage;
-use futures::StreamExt;
 use log::*;
-use nym_sphinx::framing::codec::NymCodec;
 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;
-use tokio_util::udp::UdpFramed;
 
 pub(crate) struct Listener {
     address: SocketAddr,
@@ -23,20 +22,12 @@ impl Listener {
         Listener { address, shutdown }
     }
 
-    pub(crate) async fn run<St>(&mut self, mut connection_handler: ConnectionHandler<St>)
+    pub(crate) async fn run<St>(&mut self, connection_handler: ConnectionHandler<St>)
     where
         St: Storage + Clone + 'static,
     {
         info!("Starting mixnet listener at {}", self.address);
-        let socket = match tokio::net::UdpSocket::bind(self.address).await {
-            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);
-            }
-        };
-
-        let mut framed_conn = UdpFramed::new(socket, NymCodec);
+        let endpoint = Endpoint::server(server_config(), self.address).unwrap();
 
         while !self.shutdown.is_shutdown() {
             tokio::select! {
@@ -44,25 +35,22 @@ impl Listener {
                 _ = self.shutdown.recv() => {
                     log::trace!("mixnet_handling::Listener: Received shutdown");
                 }
-                framed_sphinx_packet = framed_conn.next() => {
-                    match framed_sphinx_packet {
-                        Some(Ok((framed_sphinx_packet, remote))) => {
-                            // TODO: benchmark spawning tokio task with full processing vs just processing it
-                            // synchronously 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
-                            debug!("Handling packet from {remote:?}");
-                            connection_handler.handle_received_packet(framed_sphinx_packet).await;
+                connection = endpoint.accept() => {
+                    match connection {
+                        Some(connecting) => {
+                            match connecting.await {
+                                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:?}"),
+                            }
                         }
-                        Some(Err(err)) => {
-                            error!(
-                                "The socket connection got corrupted with error: {err}. Closing the socket",
-                            );
-                            return;
-                        }
-                        None => break, // stream got closed by remote
+                        None => {
+                            error!("Endpoint closed");
+                            break;
+                        }, // stream got closed by remote
                     }
                 }
             }
@@ -78,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

@@ -37,6 +37,10 @@ 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 tokio::net::TcpStream;
+use quinn::{Connection, ConnectionError};
 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))
     )]
-    pub(crate) async 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,50 +78,61 @@ impl ConnectionHandler {
         })
     }
 
-    // pub(crate) async fn handle_connection(
-    //     self,
-    //     conn: TcpStream,
-    //     remote: SocketAddr,
-    //     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 {
-    //                     Some(Ok(framed_sphinx_packet)) => {
-    //                         // TODO: benchmark spawning tokio task with full processing vs just processing it
-    //                         // synchronously (without delaying inside of course,
-    //                         // delay is moved to a global DelayQueue)
-    //                         // under higher load in single and multi-threaded situation.
+    pub(crate) async fn handle_connection(self, conn: Connection, mut shutdown: TaskClient) {
+        debug!(
+            "Starting connection handler for {:?}",
+            conn.remote_address()
+        );
+        shutdown.mark_as_success();
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                biased;
+                _ = shutdown.recv() => {
+                    log::trace!("ConnectionHandler: received shutdown");
+                }
 
-    //                         // 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);
-    //                     }
-    //                     Some(Err(err)) => {
-    //                         error!(
-    //                             "{remote:?} - The socket connection got corrupted with error: {err}. Closing the socket",
-    //                         );
-    //                         return;
-    //                     }
-    //                     None => break, // stream got closed by remote
-    //                 }
-    //             },
-    //         }
-    //     }
+                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;
+                                },
+                            }
+                        }
 
-    //     info!(
-    //         "Closing connection from {:?}",
-    //         framed_conn.into_inner().peer_addr()
-    //     );
-    //     log::trace!("ConnectionHandler: Exiting");
-    // }
+                    };
+
+                    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.
+
+                            // 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).await;
+                        }
+                        Some(Err(err)) => {
+                            error!(
+                                "The socket connection got corrupted with error: {err}. Closing the socket",
+                            );
+                            return;
+                        }
+                        None => break, // stream got closed by remote
+                    }
+                },
+            }
+        }
+
+        info!("Closing connection from {:?}", conn.remote_address());
+        log::trace!("ConnectionHandler: Exiting");
+    }
 }

mixnode/src/node/listener/mod.rs

@@ -2,13 +2,11 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::node::listener::connection_handler::ConnectionHandler;
-use futures::StreamExt;
-use nym_sphinx::framing::codec::NymCodec;
+use quinn::{Endpoint, ServerConfig};
+use rcgen::generate_simple_self_signed;
+use rustls::{Certificate, PrivateKey};
 use std::net::SocketAddr;
-use std::process;
-use tokio::net::UdpSocket;
 use tokio::task::JoinHandle;
-use tokio_util::udp::UdpFramed;
 #[cfg(feature = "cpucycles")]
 use tracing::error;
 
@@ -28,15 +26,7 @@ impl Listener {
 
     async fn run(&mut self, connection_handler: ConnectionHandler) {
         log::trace!("Starting Listener");
-        let socket = match UdpSocket::bind(self.address).await {
-            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);
-            }
-        };
-
-        let mut framed_conn = UdpFramed::new(socket, NymCodec);
+        let endpoint = Endpoint::server(server_config(), self.address).unwrap();
 
         while !self.shutdown.is_shutdown() {
             tokio::select! {
@@ -44,28 +34,28 @@ impl Listener {
                 _ = self.shutdown.recv() => {
                     log::trace!("Listener: Received shutdown");
                 },
-                framed_sphinx_packet = framed_conn.next() => {
-                    match framed_sphinx_packet {
-                        Some(Ok((framed_sphinx_packet, remote))) => {
-                            // TODO: benchmark spawning tokio task with full processing vs just processing it
-                            // synchronously (without delaying inside of course,
-                            // 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
-                            debug!("Handling packet from {remote:?}");
-                            let connection_handler_clone = connection_handler.clone();
-                            tokio::spawn(async move {connection_handler_clone.handle_received_packet(framed_sphinx_packet).await });
+                connection = endpoint.accept() => {
+                    match connection {
+                        Some(connecting) => {
+                           match connecting.await {
+                                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:?}"),
+                            }
                         }
-                        Some(Err(err)) => {
-                            error!(
-                                "The socket connection got corrupted with error: {err}. Closing the socket",
-                            );
-                            return;
-                        }
-                        None => break, // stream got closed by remote
+                        // 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
                     }
                 },
             };
@@ -79,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")
+}