diff --git a/Cargo.lock b/Cargo.lock index 3a65ed0106..13f17a9101 100644 --- a/Cargo.lock +++ b/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" diff --git a/common/bin-common/src/logging/mod.rs b/common/bin-common/src/logging/mod.rs index 135736ac2b..9898b84589 100644 --- a/common/bin-common/src/logging/mod.rs +++ b/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(); } diff --git a/common/client-libs/mixnet-client/Cargo.toml b/common/client-libs/mixnet-client/Cargo.toml index fabcca892d..b6f58a8997 100644 --- a/common/client-libs/mixnet-client/Cargo.toml +++ b/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" } diff --git a/common/client-libs/mixnet-client/src/client.rs b/common/client-libs/mixnet-client/src/client.rs index 4da308144f..03ff610b61 100644 --- a/common/client-libs/mixnet-client/src/client.rs +++ b/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 tokio::net::TcpStream; +use std::time::{Duration, Instant}; + 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, current_reconnection_attempt: Arc, + 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, + mut receiver: mpsc::Receiver, connection_timeout: Duration, current_reconnection: &AtomicU32, ) { - let connection_fut = TcpStream::connect(address); + let mut endpoint = Endpoint::client("0.0.0.0:0".parse::().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; } }; - - // 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(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 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 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", - )) - } + 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 { - 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 { + 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, + _ocsp_response: &[u8], + _now: std::time::SystemTime, + ) -> Result { + 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)) +} diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index 76f78fc189..0d2c070eca 100644 --- a/gateway/Cargo.toml +++ b/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" } diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs index dd2bc5d402..2eac9dbcdb 100644 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ b/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 { packet_processor: PacketProcessor, @@ -174,23 +173,37 @@ impl ConnectionHandler { 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); + 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(); - 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 under higher load in single and multi-threaded situation. @@ -212,9 +225,6 @@ impl ConnectionHandler { } } - info!( - "Closing connection from {:?}", - framed_conn.into_inner().peer_addr() - ); + info!("Closing connection from {:?}", conn.remote_address()); } } diff --git a/gateway/src/node/mixnet_handling/receiver/listener.rs b/gateway/src/node/mixnet_handling/receiver/listener.rs index 38fd8ef43b..dd3abdc174 100644 --- a/gateway/src/node/mixnet_handling/receiver/listener.rs +++ b/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 { - 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 endpoint = Endpoint::server(server_config(), self.address).unwrap(); 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)) => { - let handler = connection_handler.clone(); - tokio::spawn(handler.handle_connection(socket, remote_addr, self.shutdown.clone())); + 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:?}"), + } } - 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> { + 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") +} diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml index c12e9259b3..952188a681 100644 --- a/mixnode/Cargo.toml +++ b/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 } diff --git a/mixnode/src/node/listener/connection_handler/mod.rs b/mixnode/src/node/listener/connection_handler/mod.rs index 733d2660f8..7b6c57d663 100644 --- a/mixnode/src/node/listener/connection_handler/mod.rs +++ b/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)) )] - 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( - self, - conn: TcpStream, - remote: SocketAddr, - mut shutdown: TaskClient, - ) { - debug!("Starting connection handler for {:?}", remote); + pub(crate) async fn handle_connection(self, conn: Connection, mut shutdown: TaskClient) { + debug!( + "Starting connection handler for {:?}", + conn.remote_address() + ); 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, - // delay is moved to a global DelayQueue) - // under higher load in single and multi-threaded situation. + // 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); + 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!( - "Closing connection from {:?}", - framed_conn.into_inner().peer_addr() - ); + info!("Closing connection from {:?}", conn.remote_address()); log::trace!("ConnectionHandler: Exiting"); } } diff --git a/mixnode/src/node/listener/mod.rs b/mixnode/src/node/listener/mod.rs index f8fad07b23..1d4beb718f 100644 --- a/mixnode/src/node/listener/mod.rs +++ b/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 { - 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 endpoint = Endpoint::server(server_config(), self.address).unwrap(); 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)) => { - let handler = connection_handler.clone(); - tokio::spawn(handler.handle_connection(socket, remote_addr, self.shutdown.clone())); + 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:?}"), + } } - 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> { + 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") +}