From 860a69b246c35f3bf15fa97cf16ca42b220b8089 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 27 May 2020 10:08:08 +0100 Subject: [PATCH] Feature/sphinx socket packet encoder (#245) * Ability to send sphinx packets of different sizes + more efficient decoding * Closing connection on connection corruption * Missing semicolons * Missing license notices * Default for packetsize --- Cargo.lock | 32 +- Cargo.toml | 2 +- clients/desktop/Cargo.toml | 2 +- clients/desktop/src/client/mix_traffic.rs | 2 +- common/client-libs/gateway-client/src/lib.rs | 4 +- .../Cargo.toml | 6 +- .../src/connection_manager/mod.rs | 72 ++++- .../src/connection_manager/reconnector.rs | 0 .../src/connection_manager/writer.rs | 52 ++++ common/client-libs/mixnet-client/src/lib.rs | 281 ++++++++++++++++++ .../src/connection_manager/writer.rs | 60 ---- .../client-libs/multi-tcp-client/src/lib.rs | 274 ----------------- common/healthcheck/Cargo.toml | 2 +- common/healthcheck/src/path_check.rs | 8 +- common/nymsphinx/Cargo.toml | 2 + common/nymsphinx/src/chunking/mod.rs | 46 +++ common/nymsphinx/src/framing/mod.rs | 147 +++++++++ common/nymsphinx/src/lib.rs | 3 + common/nymsphinx/src/packets.rs | 76 +++++ gateway/Cargo.toml | 3 +- gateway/gateway-requests/src/types.rs | 30 +- .../websocket/connection_handler.rs | 7 +- .../receiver/connection_handler.rs | 65 ++-- .../receiver/packet_processing.rs | 8 +- .../src/node/mixnet_handling/sender/mod.rs | 16 +- mixnode/Cargo.toml | 3 +- mixnode/src/node/listener.rs | 74 ++--- mixnode/src/node/mod.rs | 5 +- mixnode/src/node/packet_forwarding.rs | 16 +- mixnode/src/node/packet_processing.rs | 11 +- 30 files changed, 817 insertions(+), 492 deletions(-) rename common/client-libs/{multi-tcp-client => mixnet-client}/Cargo.toml (68%) rename common/client-libs/{multi-tcp-client => mixnet-client}/src/connection_manager/mod.rs (64%) rename common/client-libs/{multi-tcp-client => mixnet-client}/src/connection_manager/reconnector.rs (100%) create mode 100644 common/client-libs/mixnet-client/src/connection_manager/writer.rs create mode 100644 common/client-libs/mixnet-client/src/lib.rs delete mode 100644 common/client-libs/multi-tcp-client/src/connection_manager/writer.rs delete mode 100644 common/client-libs/multi-tcp-client/src/lib.rs create mode 100644 common/nymsphinx/src/framing/mod.rs create mode 100644 common/nymsphinx/src/packets.rs diff --git a/Cargo.lock b/Cargo.lock index 5f0bf0be50..ba9abdd109 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1014,7 +1014,7 @@ dependencies = [ "futures 0.3.4", "itertools", "log 0.4.8", - "multi-tcp-client", + "mixnet-client", "nymsphinx", "pretty_env_logger", "rand 0.7.3", @@ -1572,6 +1572,17 @@ dependencies = [ "winapi 0.3.8", ] +[[package]] +name = "mixnet-client" +version = "0.1.0" +dependencies = [ + "futures 0.3.4", + "log 0.4.8", + "nymsphinx", + "tokio 0.2.16", + "tokio-util", +] + [[package]] name = "mockito" version = "0.23.3" @@ -1596,15 +1607,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "41f5c9112cb662acd3b204077e0de5bc66305fa8df65c8019d5adb10e9ab6e58" -[[package]] -name = "multi-tcp-client" -version = "0.1.0" -dependencies = [ - "futures 0.3.4", - "log 0.4.8", - "tokio 0.2.16", -] - [[package]] name = "native-tls" version = "0.2.4" @@ -1680,7 +1682,7 @@ dependencies = [ "gateway-client", "gateway-requests", "log 0.4.8", - "multi-tcp-client", + "mixnet-client", "nymsphinx", "pem", "pemstore", @@ -1728,7 +1730,7 @@ dependencies = [ "gateway-requests", "hmac", "log 0.4.8", - "multi-tcp-client", + "mixnet-client", "nymsphinx", "pemstore", "pretty_env_logger", @@ -1739,6 +1741,7 @@ dependencies = [ "tempfile", "tokio 0.2.16", "tokio-tungstenite", + "tokio-util", "tungstenite", ] @@ -1757,13 +1760,14 @@ dependencies = [ "dotenv", "futures 0.3.4", "log 0.4.8", - "multi-tcp-client", + "mixnet-client", "nymsphinx", "pemstore", "pretty_env_logger", "serde", "tempfile", "tokio 0.2.16", + "tokio-util", "topology", ] @@ -1799,10 +1803,12 @@ dependencies = [ name = "nymsphinx" version = "0.1.0" dependencies = [ + "bytes 0.5.4", "log 0.4.8", "rand 0.7.3", "rand_distr", "sphinx", + "tokio-util", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 121ba955fd..639a30dc51 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,7 +9,7 @@ members = [ "clients/webassembly", "common/client-libs/directory-client", "common/client-libs/gateway-client", - "common/client-libs/multi-tcp-client", + "common/client-libs/mixnet-client", "common/client-libs/validator-client", "common/config", "common/crypto", diff --git a/clients/desktop/Cargo.toml b/clients/desktop/Cargo.toml index ef68fd20b2..4dfbff474f 100644 --- a/clients/desktop/Cargo.toml +++ b/clients/desktop/Cargo.toml @@ -34,7 +34,7 @@ crypto = {path = "../../common/crypto"} directory-client = { path = "../../common/client-libs/directory-client" } gateway-client = { path = "../../common/client-libs/gateway-client" } gateway-requests = { path = "../../gateway/gateway-requests" } -multi-tcp-client = { path = "../../common/client-libs/multi-tcp-client" } +mixnet-client = { path = "../../common/client-libs/mixnet-client" } nymsphinx = { path = "../../common/nymsphinx" } pemstore = {path = "../../common/pemstore"} topology = {path = "../../common/topology" } diff --git a/clients/desktop/src/client/mix_traffic.rs b/clients/desktop/src/client/mix_traffic.rs index e9d858a6df..3ccded4532 100644 --- a/clients/desktop/src/client/mix_traffic.rs +++ b/clients/desktop/src/client/mix_traffic.rs @@ -53,7 +53,7 @@ impl<'a> MixTrafficController<'static> { debug!("Got a mix_message for {:?}", mix_message.0); match self .gateway_client - .send_sphinx_packet(mix_message.0, mix_message.1.to_bytes()) + .send_sphinx_packet(mix_message.0, mix_message.1) .await { Err(e) => error!("Failed to send sphinx packet to the gateway! - {:?}", e), diff --git a/common/client-libs/gateway-client/src/lib.rs b/common/client-libs/gateway-client/src/lib.rs index 4b834fbe57..a8923ebc77 100644 --- a/common/client-libs/gateway-client/src/lib.rs +++ b/common/client-libs/gateway-client/src/lib.rs @@ -17,7 +17,7 @@ use futures::{channel::mpsc, future::BoxFuture, FutureExt, SinkExt, StreamExt}; use gateway_requests::auth_token::{AuthToken, AuthTokenConversionError}; use gateway_requests::{BinaryRequest, ClientControlRequest, ServerResponse}; use log::*; -use nymsphinx::DestinationAddressBytes; +use nymsphinx::{DestinationAddressBytes, SphinxPacket}; use std::convert::TryFrom; use std::fmt::{self, Error, Formatter}; use std::net::SocketAddr; @@ -439,7 +439,7 @@ where pub async fn send_sphinx_packet( &mut self, address: SocketAddr, - packet: Vec, + packet: SphinxPacket, ) -> Result<(), GatewayClientError> { if !self.authenticated { return Err(GatewayClientError::NotAuthenticated); diff --git a/common/client-libs/multi-tcp-client/Cargo.toml b/common/client-libs/mixnet-client/Cargo.toml similarity index 68% rename from common/client-libs/multi-tcp-client/Cargo.toml rename to common/client-libs/mixnet-client/Cargo.toml index 4887808519..4aa980a23f 100644 --- a/common/client-libs/multi-tcp-client/Cargo.toml +++ b/common/client-libs/mixnet-client/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "multi-tcp-client" +name = "mixnet-client" version = "0.1.0" authors = ["Jedrzej Stuczynski "] edition = "2018" @@ -10,3 +10,7 @@ edition = "2018" futures = "0.3" log = "0.4.8" tokio = { version = "0.2", features = ["full"] } +tokio-util = { version = "0.3.1", features = ["codec"] } + +# internal +nymsphinx = {path = "../../nymsphinx" } diff --git a/common/client-libs/multi-tcp-client/src/connection_manager/mod.rs b/common/client-libs/mixnet-client/src/connection_manager/mod.rs similarity index 64% rename from common/client-libs/multi-tcp-client/src/connection_manager/mod.rs rename to common/client-libs/mixnet-client/src/connection_manager/mod.rs index 0cb5cd3951..08eec122a6 100644 --- a/common/client-libs/multi-tcp-client/src/connection_manager/mod.rs +++ b/common/client-libs/mixnet-client/src/connection_manager/mod.rs @@ -17,8 +17,9 @@ use crate::connection_manager::writer::ConnectionWriter; use futures::channel::{mpsc, oneshot}; use futures::future::{abortable, AbortHandle}; use futures::task::Poll; -use futures::{AsyncWriteExt, StreamExt}; +use futures::{SinkExt, StreamExt}; use log::*; +use nymsphinx::SphinxPacket; use std::io; use std::net::SocketAddr; use std::time::Duration; @@ -29,8 +30,8 @@ mod writer; pub(crate) type ResponseSender = Option>>; -pub(crate) type ConnectionManagerSender = mpsc::UnboundedSender<(Vec, ResponseSender)>; -type ConnectionManagerReceiver = mpsc::UnboundedReceiver<(Vec, ResponseSender)>; +pub(crate) type ConnectionManagerSender = mpsc::UnboundedSender<(SphinxPacket, ResponseSender)>; +type ConnectionManagerReceiver = mpsc::UnboundedReceiver<(SphinxPacket, ResponseSender)>; enum ConnectionState<'a> { Writing(ConnectionWriter), @@ -47,6 +48,7 @@ pub(crate) struct ConnectionManager<'a> { reconnection_backoff: Duration, state: ConnectionState<'a>, + pending_messages_buffer: Vec, } impl<'a> Drop for ConnectionManager<'a> { @@ -90,13 +92,14 @@ impl<'a> ConnectionManager<'static> { maximum_reconnection_backoff, reconnection_backoff, state: initial_state, + pending_messages_buffer: Vec::new(), } } async fn run(mut self) { while let Some(msg) = self.conn_rx.next().await { let (msg_content, res_ch) = msg; - let res = self.handle_new_message(msg_content).await; + let res = self.handle_new_packet(msg_content).await; if let Some(res_ch) = res_ch { if let Err(e) = res_ch.send(res) { error!( @@ -118,15 +121,21 @@ impl<'a> ConnectionManager<'static> { (sender_clone, abort_handle) } - async fn handle_new_message(&mut self, msg: Vec) -> io::Result<()> { + // Possible future TODO: `Framed<...>` is both a Sink and a Stream, + // so it is possible to read any responses we might receive (it is also duplex, so that could be + // done while writing packets themselves). But it'd require slight additions to `SphinxCodec` + async fn handle_new_packet(&mut self, packet: SphinxPacket) -> io::Result<()> { if let ConnectionState::Reconnecting(conn_reconnector) = &mut self.state { // do a single poll rather than await for future to completely resolve let new_connection = match futures::poll(conn_reconnector).await { Poll::Pending => { + // make sure we don't lose the received packet + self.pending_messages_buffer.push(packet); return Err(io::Error::new( io::ErrorKind::BrokenPipe, "connection is broken - reconnection is in progress", - )) + ) + .into()); } Poll::Ready(conn) => conn, }; @@ -138,24 +147,55 @@ impl<'a> ConnectionManager<'static> { // we must be in writing state if we are here, either by being here from beginning or just // transitioning from reconnecting if let ConnectionState::Writing(conn_writer) = &mut self.state { - return match conn_writer.write_all(msg.as_ref()).await { - // if we failed to write to connection we should reconnect - // TODO: is this true? can we fail to write to a connection while it still remains open and valid? + // check if we have any pending writes + return if !self.pending_messages_buffer.is_empty() { + let pending_messages = + std::mem::replace(&mut self.pending_messages_buffer, Vec::new()); + let mut send_stream = futures::stream::iter( + pending_messages + .into_iter() + .chain(std::iter::once(packet)) + .map(|packet| Ok(packet)), + ); - // TODO: change connection writer to somehow also poll for responses and - // change return type of this method from io::Result<> to io::Result> - Ok(_) => Ok(()), - Err(e) => { - trace!("Creating connection reconnector!"); + match conn_writer.send_all(&mut send_stream).await { + Ok(_) => Ok(()), + Err(e) => { + // whatever wasn't successfully sent, put it back into the buffer + // (Looking at Future impl for SendAll, I *think* it does not consume + // items it failed to send) + self.pending_messages_buffer = send_stream + .filter_map(|x| async move { x.ok() }) // presumably all items will be an 'Ok'? + .collect() + .await; + + warn!( + "Failed to forward messages - {:?}. Starting reconnection procedure...", + e + ); + self.state = ConnectionState::Reconnecting(ConnectionReconnector::new( + self.address, + self.reconnection_backoff, + self.maximum_reconnection_backoff, + )); + Err(e.into()) + } + } + } else { + if let Err(e) = conn_writer.send(packet).await { + warn!( + "Failed to forward message - {:?}. Starting reconnection procedure...", + e + ); self.state = ConnectionState::Reconnecting(ConnectionReconnector::new( self.address, self.reconnection_backoff, self.maximum_reconnection_backoff, )); - Err(e) } + Ok(()) }; - }; + } unreachable!(); } diff --git a/common/client-libs/multi-tcp-client/src/connection_manager/reconnector.rs b/common/client-libs/mixnet-client/src/connection_manager/reconnector.rs similarity index 100% rename from common/client-libs/multi-tcp-client/src/connection_manager/reconnector.rs rename to common/client-libs/mixnet-client/src/connection_manager/reconnector.rs diff --git a/common/client-libs/mixnet-client/src/connection_manager/writer.rs b/common/client-libs/mixnet-client/src/connection_manager/writer.rs new file mode 100644 index 0000000000..1208049b01 --- /dev/null +++ b/common/client-libs/mixnet-client/src/connection_manager/writer.rs @@ -0,0 +1,52 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use futures::task::{Context, Poll}; +use futures::Sink; +use nymsphinx::framing::{SphinxCodec, SphinxCodecError}; +use nymsphinx::SphinxPacket; +use std::pin::Pin; +use tokio_util::codec::Framed; + +pub(crate) struct ConnectionWriter { + framed_connection: Framed, +} + +impl ConnectionWriter { + pub(crate) fn new(connection: tokio::net::TcpStream) -> Self { + ConnectionWriter { + framed_connection: Framed::new(connection, SphinxCodec), + } + } +} + +impl Sink for ConnectionWriter { + type Error = SphinxCodecError; + + fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.framed_connection).poll_ready(cx) + } + + fn start_send(mut self: Pin<&mut Self>, item: SphinxPacket) -> Result<(), Self::Error> { + Pin::new(&mut self.framed_connection).start_send(item) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.framed_connection).poll_flush(cx) + } + + fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.framed_connection).poll_close(cx) + } +} diff --git a/common/client-libs/mixnet-client/src/lib.rs b/common/client-libs/mixnet-client/src/lib.rs new file mode 100644 index 0000000000..d2b6bdbc82 --- /dev/null +++ b/common/client-libs/mixnet-client/src/lib.rs @@ -0,0 +1,281 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::connection_manager::{ConnectionManager, ConnectionManagerSender}; +use futures::channel::oneshot; +use futures::future::AbortHandle; +use log::*; +use nymsphinx::SphinxPacket; +use std::collections::HashMap; +use std::io; +use std::net::SocketAddr; +use std::time::Duration; +use tokio::runtime::Handle; + +mod connection_manager; + +pub struct Config { + initial_reconnection_backoff: Duration, + maximum_reconnection_backoff: Duration, + initial_connection_timeout: Duration, +} + +impl Config { + pub fn new( + initial_reconnection_backoff: Duration, + maximum_reconnection_backoff: Duration, + initial_connection_timeout: Duration, + ) -> Self { + Config { + initial_reconnection_backoff, + maximum_reconnection_backoff, + initial_connection_timeout, + } + } +} + +pub struct Client { + runtime_handle: Handle, + connections_managers: HashMap, + maximum_reconnection_backoff: Duration, + initial_reconnection_backoff: Duration, + initial_connection_timeout: Duration, +} + +impl Client { + pub fn new(config: Config) -> Client { + Client { + // if the function is not called within tokio runtime context, this will panic + // but perhaps the code should be better structured to completely avoid this call + runtime_handle: Handle::try_current() + .expect("The client MUST BE used within tokio runtime context"), + connections_managers: HashMap::new(), + initial_reconnection_backoff: config.initial_reconnection_backoff, + maximum_reconnection_backoff: config.maximum_reconnection_backoff, + initial_connection_timeout: config.initial_connection_timeout, + } + } + + async fn start_new_connection_manager( + &mut self, + address: SocketAddr, + ) -> (ConnectionManagerSender, AbortHandle) { + let (sender, abort_handle) = ConnectionManager::new( + address, + self.initial_reconnection_backoff, + self.maximum_reconnection_backoff, + self.initial_connection_timeout, + ) + .await + .start_abortable(&self.runtime_handle); + + (sender, abort_handle) + } + + // if wait_for_response is set to true, we will get information about any possible IO errors + // as well as (once implemented) received replies, however, this will also cause way longer + // waiting periods + pub async fn send( + &mut self, + address: SocketAddr, + packet: SphinxPacket, + wait_for_response: bool, + ) -> io::Result<()> { + trace!("Sending packet to {:?}", address); + if !self.connections_managers.contains_key(&address) { + debug!( + "There is no existing connection to {:?} - it will be established now", + address + ); + + let (new_manager_sender, abort_handle) = + self.start_new_connection_manager(address).await; + self.connections_managers + .insert(address, (new_manager_sender, abort_handle)); + } + + let manager = self.connections_managers.get_mut(&address).unwrap(); + + if wait_for_response { + let (res_tx, res_rx) = oneshot::channel(); + manager.0.unbounded_send((packet, Some(res_tx))).unwrap(); + res_rx.await.unwrap() + } else { + manager.0.unbounded_send((packet, None)).unwrap(); + Ok(()) + } + } +} + +impl Drop for Client { + fn drop(&mut self) { + for (_, abort_handle) in self.connections_managers.values() { + abort_handle.abort() + } + } +} + +/* + The below tests weren't extremely reliable to begin with, however, + to restore them as they were before, we'd need to expose some kind of 'SphinxPacket::test_fixture()` + function. +*/ +// +// #[cfg(test)] +// mod tests { +// use super::*; +// use std::str; +// use std::time; +// use tokio::prelude::*; +// +// const SERVER_MSG_LEN: usize = 16; +// const CLOSE_MESSAGE: [u8; SERVER_MSG_LEN] = [0; SERVER_MSG_LEN]; +// +// struct DummyServer { +// received_buf: Vec>, +// listener: tokio::net::TcpListener, +// } +// +// impl DummyServer { +// async fn new(address: SocketAddr) -> Self { +// DummyServer { +// received_buf: Vec::new(), +// listener: tokio::net::TcpListener::bind(address).await.unwrap(), +// } +// } +// +// fn get_received(&self) -> Vec> { +// self.received_buf.clone() +// } +// +// // this is only used in tests so slightly higher logging levels are fine +// async fn listen_until(mut self, close_message: &[u8]) -> Self { +// let (mut socket, _) = self.listener.accept().await.unwrap(); +// loop { +// let mut buf = [0u8; SERVER_MSG_LEN]; +// match socket.read(&mut buf).await { +// Ok(n) if n == 0 => { +// info!("Remote connection closed"); +// return self; +// } +// Ok(n) => { +// info!("received ({}) - {:?}", n, str::from_utf8(buf[..n].as_ref())); +// +// if buf[..n].as_ref() == close_message { +// info!("closing..."); +// socket.shutdown(std::net::Shutdown::Both).unwrap(); +// return self; +// } else { +// self.received_buf.push(buf[..n].to_vec()); +// } +// } +// Err(e) => { +// panic!("failed to read from socket; err = {:?}", e); +// } +// }; +// } +// } +// } +// +// #[test] +// fn client_reconnects_to_server_after_it_went_down() { +// let mut rt = tokio::runtime::Runtime::new().unwrap(); +// let addr = "127.0.0.1:6000".parse().unwrap(); +// let reconnection_backoff = Duration::from_secs(1); +// let timeout = Duration::from_secs(1); +// let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff, timeout); +// +// let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()]; +// +// let dummy_server = rt.block_on(DummyServer::new(addr)); +// let finished_dummy_server_future = rt.spawn(dummy_server.listen_until(&CLOSE_MESSAGE)); +// +// let mut c = rt.enter(|| Client::new(client_config)); +// +// for msg in &messages_to_send { +// rt.block_on(c.send(addr, msg.clone(), true)).unwrap(); +// } +// +// // kill server +// rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) +// .unwrap(); +// let received_messages = rt +// .block_on(finished_dummy_server_future) +// .unwrap() +// .get_received(); +// +// assert_eq!(received_messages, messages_to_send); +// +// // try to send - go into reconnection +// let post_kill_message = [3u8; SERVER_MSG_LEN].to_vec(); +// +// // we are trying to send to killed server +// assert!(rt +// .block_on(c.send(addr, post_kill_message.clone(), true)) +// .is_err()); +// +// let new_dummy_server = rt.block_on(DummyServer::new(addr)); +// let new_server_future = rt.spawn(new_dummy_server.listen_until(&CLOSE_MESSAGE)); +// +// // keep sending after we leave reconnection backoff and reconnect +// loop { +// if rt +// .block_on(c.send(addr, post_kill_message.clone(), true)) +// .is_ok() +// { +// break; +// } +// rt.block_on( +// async move { tokio::time::delay_for(time::Duration::from_millis(50)).await }, +// ); +// } +// +// // kill the server to ensure it actually got the message +// rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) +// .unwrap(); +// let new_received_messages = rt.block_on(new_server_future).unwrap().get_received(); +// assert_eq!(post_kill_message.to_vec(), new_received_messages[0]); +// } +// +// #[test] +// fn server_receives_all_sent_messages_when_up() { +// let mut rt = tokio::runtime::Runtime::new().unwrap(); +// let addr = "127.0.0.1:6001".parse().unwrap(); +// let reconnection_backoff = Duration::from_secs(2); +// let timeout = Duration::from_secs(1); +// let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff, timeout); +// +// let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()]; +// +// let dummy_server = rt.block_on(DummyServer::new(addr)); +// let finished_dummy_server_future = rt.spawn(dummy_server.listen_until(&CLOSE_MESSAGE)); +// +// let mut c = rt.enter(|| Client::new(client_config)); +// +// for msg in &messages_to_send { +// rt.block_on(c.send(addr, msg.clone(), true)).unwrap(); +// } +// +// rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) +// .unwrap(); +// +// // the server future should have already been resolved +// let received_messages = rt +// .block_on(finished_dummy_server_future) +// .unwrap() +// .get_received(); +// +// assert_eq!(received_messages, messages_to_send); +// } +// } diff --git a/common/client-libs/multi-tcp-client/src/connection_manager/writer.rs b/common/client-libs/multi-tcp-client/src/connection_manager/writer.rs deleted file mode 100644 index 90d3a3ff6b..0000000000 --- a/common/client-libs/multi-tcp-client/src/connection_manager/writer.rs +++ /dev/null @@ -1,60 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use futures::task::{Context, Poll}; -use futures::AsyncWrite; -use std::io; -use std::pin::Pin; -use tokio::prelude::*; - -pub(crate) struct ConnectionWriter { - connection: tokio::net::TcpStream, -} - -impl ConnectionWriter { - pub(crate) fn new(connection: tokio::net::TcpStream) -> Self { - ConnectionWriter { connection } - } -} - -impl AsyncWrite for ConnectionWriter { - fn poll_write( - mut self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - use tokio::io::AsyncWrite; - - let mut read_buf = [0; 1]; - match Pin::new(&mut self.connection).poll_read(cx, &mut read_buf) { - // at least try the obvious check for if connection is definitely down - // TODO: can we do anything else? - Poll::Ready(Ok(n)) if n == 0 => Poll::Ready(Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "trying to write to closed connection", - ))), - _ => Pin::new(&mut self.connection).poll_write(cx, buf), - } - } - - fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - use tokio::io::AsyncWrite; - Pin::new(&mut self.connection).poll_flush(cx) - } - - fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - use tokio::io::AsyncWrite; - Pin::new(&mut self.connection).poll_shutdown(cx) - } -} diff --git a/common/client-libs/multi-tcp-client/src/lib.rs b/common/client-libs/multi-tcp-client/src/lib.rs deleted file mode 100644 index 4d967f1431..0000000000 --- a/common/client-libs/multi-tcp-client/src/lib.rs +++ /dev/null @@ -1,274 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::connection_manager::{ConnectionManager, ConnectionManagerSender}; -use futures::channel::oneshot; -use futures::future::AbortHandle; -use log::*; -use std::collections::HashMap; -use std::io; -use std::net::SocketAddr; -use std::time::Duration; -use tokio::runtime::Handle; - -mod connection_manager; - -pub struct Config { - initial_reconnection_backoff: Duration, - maximum_reconnection_backoff: Duration, - initial_connection_timeout: Duration, -} - -impl Config { - pub fn new( - initial_reconnection_backoff: Duration, - maximum_reconnection_backoff: Duration, - initial_connection_timeout: Duration, - ) -> Self { - Config { - initial_reconnection_backoff, - maximum_reconnection_backoff, - initial_connection_timeout, - } - } -} - -pub struct Client { - runtime_handle: Handle, - connections_managers: HashMap, - maximum_reconnection_backoff: Duration, - initial_reconnection_backoff: Duration, - initial_connection_timeout: Duration, -} - -impl Client { - pub fn new(config: Config) -> Client { - Client { - // if the function is not called within tokio runtime context, this will panic - // but perhaps the code should be better structured to completely avoid this call - runtime_handle: Handle::try_current() - .expect("The client MUST BE used within tokio runtime context"), - connections_managers: HashMap::new(), - initial_reconnection_backoff: config.initial_reconnection_backoff, - maximum_reconnection_backoff: config.maximum_reconnection_backoff, - initial_connection_timeout: config.initial_connection_timeout, - } - } - - async fn start_new_connection_manager( - &mut self, - address: SocketAddr, - ) -> (ConnectionManagerSender, AbortHandle) { - let (sender, abort_handle) = ConnectionManager::new( - address, - self.initial_reconnection_backoff, - self.maximum_reconnection_backoff, - self.initial_connection_timeout, - ) - .await - .start_abortable(&self.runtime_handle); - - (sender, abort_handle) - } - - // if wait_for_response is set to true, we will get information about any possible IO errors - // as well as (once implemented) received replies, however, this will also cause way longer - // waiting periods - pub async fn send( - &mut self, - address: SocketAddr, - message: Vec, - wait_for_response: bool, - ) -> io::Result<()> { - trace!("Sending packet to {:?}", address); - if !self.connections_managers.contains_key(&address) { - debug!( - "There is no existing connection to {:?} - it will be established now", - address - ); - - let (new_manager_sender, abort_handle) = - self.start_new_connection_manager(address).await; - self.connections_managers - .insert(address, (new_manager_sender, abort_handle)); - } - - let manager = self.connections_managers.get_mut(&address).unwrap(); - - if wait_for_response { - let (res_tx, res_rx) = oneshot::channel(); - manager.0.unbounded_send((message, Some(res_tx))).unwrap(); - res_rx.await.unwrap() - } else { - manager.0.unbounded_send((message, None)).unwrap(); - Ok(()) - } - } -} - -impl Drop for Client { - fn drop(&mut self) { - for (_, abort_handle) in self.connections_managers.values() { - abort_handle.abort() - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::str; - use std::time; - use tokio::prelude::*; - - const SERVER_MSG_LEN: usize = 16; - const CLOSE_MESSAGE: [u8; SERVER_MSG_LEN] = [0; SERVER_MSG_LEN]; - - struct DummyServer { - received_buf: Vec>, - listener: tokio::net::TcpListener, - } - - impl DummyServer { - async fn new(address: SocketAddr) -> Self { - DummyServer { - received_buf: Vec::new(), - listener: tokio::net::TcpListener::bind(address).await.unwrap(), - } - } - - fn get_received(&self) -> Vec> { - self.received_buf.clone() - } - - // this is only used in tests so slightly higher logging levels are fine - async fn listen_until(mut self, close_message: &[u8]) -> Self { - let (mut socket, _) = self.listener.accept().await.unwrap(); - loop { - let mut buf = [0u8; SERVER_MSG_LEN]; - match socket.read(&mut buf).await { - Ok(n) if n == 0 => { - info!("Remote connection closed"); - return self; - } - Ok(n) => { - info!("received ({}) - {:?}", n, str::from_utf8(buf[..n].as_ref())); - - if buf[..n].as_ref() == close_message { - info!("closing..."); - socket.shutdown(std::net::Shutdown::Both).unwrap(); - return self; - } else { - self.received_buf.push(buf[..n].to_vec()); - } - } - Err(e) => { - panic!("failed to read from socket; err = {:?}", e); - } - }; - } - } - } - - #[test] - fn client_reconnects_to_server_after_it_went_down() { - let mut rt = tokio::runtime::Runtime::new().unwrap(); - let addr = "127.0.0.1:6000".parse().unwrap(); - let reconnection_backoff = Duration::from_secs(1); - let timeout = Duration::from_secs(1); - let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff, timeout); - - let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()]; - - let dummy_server = rt.block_on(DummyServer::new(addr)); - let finished_dummy_server_future = rt.spawn(dummy_server.listen_until(&CLOSE_MESSAGE)); - - let mut c = rt.enter(|| Client::new(client_config)); - - for msg in &messages_to_send { - rt.block_on(c.send(addr, msg.clone(), true)).unwrap(); - } - - // kill server - rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) - .unwrap(); - let received_messages = rt - .block_on(finished_dummy_server_future) - .unwrap() - .get_received(); - - assert_eq!(received_messages, messages_to_send); - - // try to send - go into reconnection - let post_kill_message = [3u8; SERVER_MSG_LEN].to_vec(); - - // we are trying to send to killed server - assert!(rt - .block_on(c.send(addr, post_kill_message.clone(), true)) - .is_err()); - - let new_dummy_server = rt.block_on(DummyServer::new(addr)); - let new_server_future = rt.spawn(new_dummy_server.listen_until(&CLOSE_MESSAGE)); - - // keep sending after we leave reconnection backoff and reconnect - loop { - if rt - .block_on(c.send(addr, post_kill_message.clone(), true)) - .is_ok() - { - break; - } - rt.block_on( - async move { tokio::time::delay_for(time::Duration::from_millis(50)).await }, - ); - } - - // kill the server to ensure it actually got the message - rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) - .unwrap(); - let new_received_messages = rt.block_on(new_server_future).unwrap().get_received(); - assert_eq!(post_kill_message.to_vec(), new_received_messages[0]); - } - - #[test] - fn server_receives_all_sent_messages_when_up() { - let mut rt = tokio::runtime::Runtime::new().unwrap(); - let addr = "127.0.0.1:6001".parse().unwrap(); - let reconnection_backoff = Duration::from_secs(2); - let timeout = Duration::from_secs(1); - let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff, timeout); - - let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()]; - - let dummy_server = rt.block_on(DummyServer::new(addr)); - let finished_dummy_server_future = rt.spawn(dummy_server.listen_until(&CLOSE_MESSAGE)); - - let mut c = rt.enter(|| Client::new(client_config)); - - for msg in &messages_to_send { - rt.block_on(c.send(addr, msg.clone(), true)).unwrap(); - } - - rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true)) - .unwrap(); - - // the server future should have already been resolved - let received_messages = rt - .block_on(finished_dummy_server_future) - .unwrap() - .get_received(); - - assert_eq!(received_messages, messages_to_send); - } -} diff --git a/common/healthcheck/Cargo.toml b/common/healthcheck/Cargo.toml index 1c52152184..90e82c59c6 100644 --- a/common/healthcheck/Cargo.toml +++ b/common/healthcheck/Cargo.toml @@ -20,7 +20,7 @@ tokio = { version = "0.2", features = ["full"] } ## internal crypto = { path = "../crypto" } directory-client = { path = "../client-libs/directory-client" } -multi-tcp-client = { path = "../client-libs/multi-tcp-client" } +mixnet-client = { path = "../client-libs/mixnet-client" } nymsphinx = {path = "../nymsphinx" } topology = {path = "../topology" } diff --git a/common/healthcheck/src/path_check.rs b/common/healthcheck/src/path_check.rs index 60cc6bb845..b61f921df8 100644 --- a/common/healthcheck/src/path_check.rs +++ b/common/healthcheck/src/path_check.rs @@ -69,7 +69,7 @@ pub enum PathStatus { pub(crate) struct PathChecker { provider_clients: HashMap<[u8; 32], Option>, - mixnet_client: multi_tcp_client::Client, + mixnet_client: mixnet_client::Client, paths_status: HashMap, PathStatus>, our_destination: Destination, check_id: CheckId, @@ -114,7 +114,7 @@ impl PathChecker { } // there's no reconnection allowed - if it fails, then it fails. - let mixnet_client_config = multi_tcp_client::Config::new( + let mixnet_client_config = mixnet_client::Config::new( Duration::from_secs(1_000_000_000), Duration::from_secs(1_000_000_000), connection_timeout, @@ -122,7 +122,7 @@ impl PathChecker { PathChecker { provider_clients, - mixnet_client: multi_tcp_client::Client::new(mixnet_client_config), + mixnet_client: mixnet_client::Client::new(mixnet_client_config), our_destination: Destination::new(address, Default::default()), paths_status: HashMap::new(), check_id, @@ -293,7 +293,7 @@ impl PathChecker { match self .mixnet_client - .send(first_node_address, packet.to_bytes(), true) + .send(first_node_address, packet, true) .await { Err(err) => { diff --git a/common/nymsphinx/Cargo.toml b/common/nymsphinx/Cargo.toml index e22379f4a7..df40720199 100644 --- a/common/nymsphinx/Cargo.toml +++ b/common/nymsphinx/Cargo.toml @@ -7,9 +7,11 @@ edition = "2018" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +bytes = "0.5" log = "0.4.8" rand = {version = "0.7.3", features = ["wasm-bindgen"]} rand_distr = "0.2.2" +tokio-util = { version = "0.3.1", features = ["codec"] } ## will be moved to proper dependencies once released sphinx = { git = "https://github.com/nymtech/sphinx", rev="fcd17932acbd21a76c42a4bf2a42b9f8668f7e1f" } diff --git a/common/nymsphinx/src/chunking/mod.rs b/common/nymsphinx/src/chunking/mod.rs index d0db789eff..5f0238bace 100644 --- a/common/nymsphinx/src/chunking/mod.rs +++ b/common/nymsphinx/src/chunking/mod.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::chunking::set::split_into_sets; +use crate::packets::PacketSize; pub mod fragment; pub mod reconstruction; @@ -57,6 +58,51 @@ pub enum ChunkingError { UnexpectedFragmentCount, } +pub struct MessageChunker { + packet_size: PacketSize, + reply_surbs: bool, + surb_acks: bool, +} + +impl MessageChunker { + pub fn new() -> Self { + Default::default() + } + + pub fn with_reply_surbs(mut self, reply_surbs: bool) -> Self { + self.reply_surbs = reply_surbs; + self + } + + pub fn with_surb_acks(mut self, surb_acks: bool) -> Self { + self.surb_acks = surb_acks; + self + } + + pub fn with_packet_size(mut self, packet_size: PacketSize) -> Self { + self.packet_size = packet_size; + self + } + + pub fn finalize() -> Vec> { + todo!() + } + + pub fn attach_surb_acks() {} + + pub fn attach_reply_surbs() {} +} + +impl Default for MessageChunker { + fn default() -> Self { + MessageChunker { + packet_size: Default::default(), + reply_surbs: false, + surb_acks: false, + } + } +} + /// Takes the entire message and splits it into bytes chunks that will fit into sphinx packets /// directly. After receiving they can be combined using `reconstruction::MessageReconstructor` /// to obtain the original message back. diff --git a/common/nymsphinx/src/framing/mod.rs b/common/nymsphinx/src/framing/mod.rs new file mode 100644 index 0000000000..4add544ffd --- /dev/null +++ b/common/nymsphinx/src/framing/mod.rs @@ -0,0 +1,147 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::packets::{InvalidPacketSize, PacketSize}; +use bytes::{Buf, BufMut, BytesMut}; +use sphinx::SphinxPacket; +use std::convert::TryFrom; +use std::io; +use tokio_util::codec::{Decoder, Encoder}; + +#[derive(Debug)] +pub enum SphinxCodecError { + InvalidPacketSize, + MalformedSphinxPacket, + IoError(io::Error), +} + +impl From for SphinxCodecError { + fn from(err: io::Error) -> Self { + SphinxCodecError::IoError(err) + } +} + +impl Into for SphinxCodecError { + fn into(self) -> io::Error { + match self { + SphinxCodecError::InvalidPacketSize => { + io::Error::new(io::ErrorKind::InvalidInput, "invalid packet size") + } + SphinxCodecError::MalformedSphinxPacket => { + io::Error::new(io::ErrorKind::InvalidData, "malformed packet") + } + SphinxCodecError::IoError(err) => err, + } + } +} + +impl From for SphinxCodecError { + fn from(_: InvalidPacketSize) -> Self { + SphinxCodecError::InvalidPacketSize + } +} + +// The SphinxCodec is an extremely simple one, u8 representing one of valid packet +// lengths followed by the actual framed packet +pub struct SphinxCodec; + +impl Encoder for SphinxCodec { + type Error = SphinxCodecError; + + fn encode(&mut self, item: SphinxPacket, dst: &mut BytesMut) -> Result<(), Self::Error> { + let packet_bytes = item.to_bytes(); + let packet_length = packet_bytes.len(); + let packet_size = PacketSize::get_type(packet_length)?; + dst.reserve(1 + packet_size.size()); + dst.put_u8(packet_size as u8); + dst.put(packet_bytes.as_ref()); + Ok(()) + } +} + +impl Decoder for SphinxCodec { + type Item = SphinxPacket; + type Error = SphinxCodecError; + + //https://docs.rs/tokio-util/0.3.1/tokio_util/codec/trait.Decoder.html + fn decode(&mut self, src: &mut BytesMut) -> Result, Self::Error> { + if src.is_empty() { + // can't do anything if we have no bytes + return Ok(None); + } + // we at least have a single byte in the buffer, so we can read the expected + // length of the sphinx packet + let packet_len_flag = src[0]; + let packet_len = PacketSize::try_from(packet_len_flag)?; + + let frame_len = packet_len.size() + 1; // one is due to the flag taking the space + if src.len() < frame_len { + // we don't have enough bytes to read the entire frame + src.reserve(frame_len); + return Ok(None); + } + // we advance the buffer beyond the flag + src.advance(1); + let sphinx_packet_bytes = src.split_to(packet_len.size()); + let sphinx_packet = match SphinxPacket::from_bytes(&sphinx_packet_bytes) { + Ok(sphinx_packet) => sphinx_packet, + // here it could be debatable whether stream is corrupt or not, + // but let's go with the safer approach and assume it is. + Err(_) => return Err(SphinxCodecError::MalformedSphinxPacket), + }; + + // As per docs: + // Before returning from the function, implementations should ensure that the buffer + // has appropriate capacity in anticipation of future calls to decode. + // Failing to do so leads to inefficiency. + + // if we have at least one more byte available, we can reserve enough bytes for + // the entire next frame + if !src.is_empty() { + let next_packet_len = match PacketSize::try_from(src[0]) { + Ok(next_packet_len) => next_packet_len, + // the next frame will be malformed but let's leave handling the error to the next + // call to 'decode', as presumably, the current sphinx packet is still valid + Err(_) => return Ok(Some(sphinx_packet)), + }; + let next_frame_len = next_packet_len.size() + 1; + src.reserve(next_frame_len); + } + + Ok(Some(sphinx_packet)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[cfg(test)] + #[allow(dead_code)] + fn consume( + codec: &mut SphinxCodec, + bytes: &mut BytesMut, + ) -> Vec, SphinxCodecError>> { + let mut result = Vec::new(); + loop { + match codec.decode(bytes) { + Ok(None) => { + break; + } + output => result.push(output), + } + } + return result; + } +} diff --git a/common/nymsphinx/src/lib.rs b/common/nymsphinx/src/lib.rs index 747a255baf..e3ec920bb9 100644 --- a/common/nymsphinx/src/lib.rs +++ b/common/nymsphinx/src/lib.rs @@ -14,6 +14,8 @@ pub mod addressing; pub mod chunking; +pub mod framing; +pub mod packets; pub mod utils; // Future consideration: currently in a lot of places, the payloads have randomised content @@ -36,4 +38,5 @@ pub use sphinx::{ // re-exporting this separately to remember to put special attention to below // modules/types/constants when refactoring sphinx crate itself +// TODO: replace with sphinx::PublicKey once merged pub use sphinx::key; diff --git a/common/nymsphinx/src/packets.rs b/common/nymsphinx/src/packets.rs new file mode 100644 index 0000000000..34b22071a7 --- /dev/null +++ b/common/nymsphinx/src/packets.rs @@ -0,0 +1,76 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::convert::TryFrom; + +// it's up to the smart people to figure those values out : ) +const REGULAR_PACKET_SIZE: usize = 2 * 1024; +const ACK_PACKET_SIZE: usize = 512; +const EXTENDED_PACKET_SIZE: usize = 32 * 1024; + +pub struct InvalidPacketSize; + +#[repr(u8)] +pub enum PacketSize { + RegularPacket = 1, // for example instant messaging use case + ACKPacket = 2, // for sending SURB-ACKs + ExtendedPacket = 3, // for example for streaming fast and furious in uncompressed 10bit 4K HDR quality + + PreSURBChanges = 0, +} + +impl TryFrom for PacketSize { + type Error = InvalidPacketSize; + + fn try_from(value: u8) -> std::result::Result { + match value { + _ if value == (PacketSize::RegularPacket as u8) => Ok(Self::RegularPacket), + _ if value == (PacketSize::ACKPacket as u8) => Ok(Self::ACKPacket), + _ if value == (PacketSize::ExtendedPacket as u8) => Ok(Self::ExtendedPacket), + _ if value == (PacketSize::PreSURBChanges as u8) => Ok(Self::PreSURBChanges), + _ => Err(InvalidPacketSize), + } + } +} + +impl PacketSize { + pub fn size(&self) -> usize { + match &self { + PacketSize::RegularPacket => REGULAR_PACKET_SIZE, + PacketSize::ACKPacket => ACK_PACKET_SIZE, + PacketSize::ExtendedPacket => EXTENDED_PACKET_SIZE, + PacketSize::PreSURBChanges => crate::PACKET_SIZE, + } + } + + pub fn get_type(size: usize) -> std::result::Result { + if PacketSize::RegularPacket.size() == size { + Ok(PacketSize::RegularPacket) + } else if PacketSize::ACKPacket.size() == size { + Ok(PacketSize::ACKPacket) + } else if PacketSize::ExtendedPacket.size() == size { + Ok(PacketSize::ExtendedPacket) + } else if PacketSize::PreSURBChanges.size() == size { + Ok(PacketSize::PreSURBChanges) + } else { + Err(InvalidPacketSize) + } + } +} + +impl Default for PacketSize { + fn default() -> Self { + PacketSize::RegularPacket + } +} diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index 3b3989975a..612959470a 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -20,6 +20,7 @@ serde = { version = "1.0.104", features = ["derive"] } sha2 = "0.8.1" sled = "0.31" tokio = { version = "0.2", features = ["full"] } +tokio-util = { version = "0.3.1", features = ["codec"] } tokio-tungstenite = "0.10.1" # internal @@ -27,7 +28,7 @@ config = { path = "../common/config" } crypto = { path = "../common/crypto" } directory-client = { path = "../common/client-libs/directory-client" } gateway-requests = { path = "gateway-requests" } -multi-tcp-client = { path = "../common/client-libs/multi-tcp-client" } +mixnet-client = { path = "../common/client-libs/mixnet-client" } nymsphinx = { path = "../common/nymsphinx" } pemstore = { path = "../common/pemstore" } diff --git a/gateway/gateway-requests/src/types.rs b/gateway/gateway-requests/src/types.rs index 1271a3913c..de7672aff2 100644 --- a/gateway/gateway-requests/src/types.rs +++ b/gateway/gateway-requests/src/types.rs @@ -15,7 +15,7 @@ use crate::auth_token::AuthToken; use crate::types::BinaryRequest::ForwardSphinx; use nymsphinx::addressing::nodes::{NymNodeRoutingAddress, NymNodeRoutingAddressError}; -use nymsphinx::DestinationAddressBytes; +use nymsphinx::{DestinationAddressBytes, SphinxPacket}; use serde::{Deserialize, Serialize}; use std::{ convert::{TryFrom, TryInto}, @@ -28,6 +28,7 @@ use tokio_tungstenite::tungstenite::protocol::Message; pub enum GatewayRequestsError { IncorrectlyEncodedAddress, RequestOfInvalidSize(usize, usize), + MalformedSphinxPacket, } // to use it as `std::error::Error`, and we don't want to just derive is because we want @@ -42,6 +43,7 @@ impl fmt::Display for GatewayRequestsError { "received request had invalid size. (actual: {}, expected: {})", actual, expected ), + MalformedSphinxPacket => write!(f, "received sphinx packet was malformed"), } } } @@ -149,7 +151,10 @@ impl TryFrom for ServerResponse { } pub enum BinaryRequest { - ForwardSphinx { address: SocketAddr, data: Vec }, + ForwardSphinx { + address: SocketAddr, + sphinx_packet: SphinxPacket, + }, } impl BinaryRequest { @@ -165,16 +170,24 @@ impl BinaryRequest { nymsphinx::PACKET_SIZE, )) } else { + let sphinx_packet = match SphinxPacket::from_bytes(&raw_req[addr_offset..]) { + Ok(packet) => packet, + Err(_) => return Err(GatewayRequestsError::MalformedSphinxPacket), + }; + Ok(ForwardSphinx { address: address.into(), - data: raw_req[addr_offset..].into(), + sphinx_packet, }) } } pub fn into_bytes(self) -> Vec { match self { - BinaryRequest::ForwardSphinx { address, data } => { + BinaryRequest::ForwardSphinx { + address, + sphinx_packet, + } => { // TODO: using intermediate `NymNodeRoutingAddress` here is just temporary, because // it happens to do exactly what we needed, but we don't really want to be // dependant on what it does @@ -182,14 +195,17 @@ impl BinaryRequest { wrapped_address .as_bytes() .into_iter() - .chain(data.into_iter()) + .chain(sphinx_packet.to_bytes().into_iter()) .collect() } } } - pub fn new_forward_request(address: SocketAddr, data: Vec) -> BinaryRequest { - BinaryRequest::ForwardSphinx { address, data } + pub fn new_forward_request(address: SocketAddr, sphinx_packet: SphinxPacket) -> BinaryRequest { + BinaryRequest::ForwardSphinx { + address, + sphinx_packet, + } } } diff --git a/gateway/src/node/client_handling/websocket/connection_handler.rs b/gateway/src/node/client_handling/websocket/connection_handler.rs index 75c83fda98..92780d2f20 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler.rs @@ -145,10 +145,13 @@ where Err(e) => ServerResponse::new_error(e.to_string()), Ok(request) => match request { // currently only a single type exists - BinaryRequest::ForwardSphinx { address, data } => { + BinaryRequest::ForwardSphinx { + address, + sphinx_packet, + } => { // we know data has correct size (but nothing else besides of it) self.outbound_mix_sender - .unbounded_send((address, data)) + .unbounded_send((address, sphinx_packet)) .unwrap(); ServerResponse::Send { status: true } } diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs index 2eb1107a54..51f23d93db 100644 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs @@ -14,12 +14,16 @@ use crate::node::mixnet_handling::receiver::packet_processing::PacketProcessor; use log::*; +use nymsphinx::framing::SphinxCodec; +use nymsphinx::SphinxPacket; use std::net::SocketAddr; -use tokio::{io::AsyncReadExt, prelude::*}; +use tokio::prelude::*; +use tokio::stream::StreamExt; +use tokio_util::codec::Framed; pub(crate) struct Handle { peer_address: SocketAddr, - socket_connection: S, + framed_connection: Framed, packet_processor: PacketProcessor, } @@ -34,62 +38,47 @@ where conn: S, packet_processor: PacketProcessor, ) -> Self { + // we expect only to receive sphinx packets on this socket, so let's frame it here + let framed = Framed::new(conn, SphinxCodec); Handle { peer_address, - socket_connection: conn, + framed_connection: framed, packet_processor, } } async fn process_received_packet( - packet_data: [u8; nymsphinx::PACKET_SIZE], + sphinx_packet: SphinxPacket, mut packet_processor: PacketProcessor, ) { - match packet_processor.process_sphinx_packet(packet_data).await { + match packet_processor.process_sphinx_packet(sphinx_packet).await { Ok(_) => trace!("successfully processed [and forwarded/stored] a final hop packet"), Err(e) => debug!("We failed to process received sphinx packet - {:?}", e), } } pub(crate) async fn start_handling(&mut self) { - let mut buf = [0u8; nymsphinx::PACKET_SIZE]; - loop { - match self.socket_connection.read_exact(&mut buf).await { - // socket closed - Ok(n) if n == 0 => { - trace!("Remote connection closed."); - return; - } - Ok(n) => { - // If I understand it correctly, this if should never be executed as if `read_exact` - // does not fill buffer, it will throw UnexpectedEof? - if n != nymsphinx::PACKET_SIZE { - error!("read data of different length than expected sphinx packet size - {} (expected {})", n, nymsphinx::PACKET_SIZE); - continue; - } - - // we must be able to handle multiple packets from same connection independently - // TODO: but WE NEED to have some worker pool so that we do not spawn too many - // tasks + while let Some(sphinx_packet) = self.framed_connection.next().await { + match sphinx_packet { + Ok(sphinx_packet) => { + // we *really* need a worker pool here, because if we receive too many packets, + // we will spawn too many tasks and starve CPU due to context switching. + // (because presumably tokio has some concept of context switching in its + // scheduler) tokio::spawn(Self::process_received_packet( - buf, + sphinx_packet, self.packet_processor.clone(), - )) + )); } - Err(e) => { - if e.kind() == io::ErrorKind::UnexpectedEof { - debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\ - Also closing the connection on this end.", self.peer_address) - } else { - warn!( - "failed to read from socket (source: {:?}). Closing the connection; err = {:?}", - self.peer_address, - e - ); - } + Err(err) => { + error!( + "The socket connection got corrupted with error: {:?}. Closing the socket", + err + ); return; } - }; + } } + info!("Closing connection from {:?}", self.peer_address); } } diff --git a/gateway/src/node/mixnet_handling/receiver/packet_processing.rs b/gateway/src/node/mixnet_handling/receiver/packet_processing.rs index a62f6a77ea..8142724aa5 100644 --- a/gateway/src/node/mixnet_handling/receiver/packet_processing.rs +++ b/gateway/src/node/mixnet_handling/receiver/packet_processing.rs @@ -159,10 +159,8 @@ impl PacketProcessor { pub(crate) fn unwrap_sphinx_packet( &self, - raw_packet_data: [u8; nymsphinx::PACKET_SIZE], + packet: SphinxPacket, ) -> Result<(DestinationAddressBytes, Vec), MixProcessingError> { - let packet = SphinxPacket::from_bytes(&raw_packet_data)?; - match packet.process(self.secret_key.deref().inner()) { Ok(ProcessedPacket::ProcessedPacketForwardHop(_, _, _)) => { warn!("Received a forward hop message - those are not implemented for providers"); @@ -188,9 +186,9 @@ impl PacketProcessor { pub(crate) async fn process_sphinx_packet( &mut self, - raw_packet_data: [u8; nymsphinx::PACKET_SIZE], + sphinx_packet: SphinxPacket, ) -> Result<(), MixProcessingError> { - let (client_address, plaintext) = self.unwrap_sphinx_packet(raw_packet_data)?; + let (client_address, plaintext) = self.unwrap_sphinx_packet(sphinx_packet)?; let client_sender = self .try_to_obtain_client_ws_message_sender(client_address.clone()) diff --git a/gateway/src/node/mixnet_handling/sender/mod.rs b/gateway/src/node/mixnet_handling/sender/mod.rs index 7329f524b3..c974dc1d7c 100644 --- a/gateway/src/node/mixnet_handling/sender/mod.rs +++ b/gateway/src/node/mixnet_handling/sender/mod.rs @@ -17,15 +17,16 @@ use futures::channel::mpsc; use futures::StreamExt; use log::*; +use nymsphinx::SphinxPacket; use std::net::SocketAddr; use std::time::Duration; use tokio::task::JoinHandle; -pub(crate) type OutboundMixMessageSender = mpsc::UnboundedSender<(SocketAddr, Vec)>; -pub(crate) type OutboundMixMessageReceiver = mpsc::UnboundedReceiver<(SocketAddr, Vec)>; +pub(crate) type OutboundMixMessageSender = mpsc::UnboundedSender<(SocketAddr, SphinxPacket)>; +pub(crate) type OutboundMixMessageReceiver = mpsc::UnboundedReceiver<(SocketAddr, SphinxPacket)>; pub(crate) struct PacketForwarder { - tcp_client: multi_tcp_client::Client, + mixnet_client: mixnet_client::Client, conn_tx: OutboundMixMessageSender, conn_rx: OutboundMixMessageReceiver, } @@ -36,7 +37,7 @@ impl PacketForwarder { maximum_reconnection_backoff: Duration, initial_connection_timeout: Duration, ) -> PacketForwarder { - let tcp_client_config = multi_tcp_client::Config::new( + let tcp_client_config = mixnet_client::Config::new( initial_reconnection_backoff, maximum_reconnection_backoff, initial_connection_timeout, @@ -45,7 +46,7 @@ impl PacketForwarder { let (conn_tx, conn_rx) = mpsc::unbounded(); PacketForwarder { - tcp_client: multi_tcp_client::Client::new(tcp_client_config), + mixnet_client: mixnet_client::Client::new(tcp_client_config), conn_tx, conn_rx, } @@ -59,7 +60,10 @@ impl PacketForwarder { trace!("Going to forward packet to {:?}", address); // as a mix node we don't care about responses, we just want to fire packets // as quickly as possible - self.tcp_client.send(address, packet, false).await.unwrap(); + self.mixnet_client + .send(address, packet, false) + .await + .unwrap(); // if we're not waiting for response, we MUST get an Ok } }), diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml index c1a593eed5..3e01c1b34a 100644 --- a/mixnode/Cargo.toml +++ b/mixnode/Cargo.toml @@ -18,12 +18,13 @@ log = "0.4" pretty_env_logger = "0.3" serde = { version = "1.0.104", features = ["derive"] } tokio = { version = "0.2", features = ["full"] } +tokio-util = { version = "0.3.1", features = ["codec"] } ## internal config = {path = "../common/config"} crypto = {path = "../common/crypto"} directory-client = { path = "../common/client-libs/directory-client" } -multi-tcp-client = { path = "../common/client-libs/multi-tcp-client" } +mixnet-client = { path = "../common/client-libs/mixnet-client" } nymsphinx = {path = "../common/nymsphinx" } pemstore = {path = "../common/pemstore"} topology = {path = "../common/topology"} diff --git a/mixnode/src/node/listener.rs b/mixnode/src/node/listener.rs index 0dece62b6e..ef0e0525e0 100644 --- a/mixnode/src/node/listener.rs +++ b/mixnode/src/node/listener.rs @@ -15,29 +15,32 @@ use crate::node::packet_processing::{MixProcessingResult, PacketProcessor}; use futures::channel::mpsc; use log::*; +use nymsphinx::framing::SphinxCodec; +use nymsphinx::SphinxPacket; use std::io; use std::net::SocketAddr; -use tokio::prelude::*; use tokio::runtime::Handle; +use tokio::stream::StreamExt; use tokio::task::JoinHandle; +use tokio_util::codec::Framed; async fn process_received_packet( - packet_data: [u8; nymsphinx::PACKET_SIZE], + sphinx_packet: SphinxPacket, packet_processor: PacketProcessor, - forwarding_channel: mpsc::UnboundedSender<(SocketAddr, Vec)>, + forwarding_channel: mpsc::UnboundedSender<(SocketAddr, SphinxPacket)>, ) { // all processing incl. delay was done, the only thing left is to forward it - match packet_processor.process_sphinx_packet(packet_data).await { + match packet_processor.process_sphinx_packet(sphinx_packet).await { Err(e) => debug!("We failed to process received sphinx packet - {:?}", e), Ok(res) => match res { - MixProcessingResult::ForwardHop(hop_address, hop_data) => { + MixProcessingResult::ForwardHop(hop_address, forward_packet) => { // send our data to tcp client for forwarding. If forwarding fails, then it fails, // it's not like we can do anything about it // // in unbounded_send() failed it means that the receiver channel was disconnected // and hence something weird must have happened without a way of recovering forwarding_channel - .unbounded_send((hop_address, hop_data)) + .unbounded_send((hop_address, forward_packet)) .unwrap(); packet_processor.report_sent(hop_address); } @@ -49,57 +52,44 @@ async fn process_received_packet( } async fn process_socket_connection( - mut socket: tokio::net::TcpStream, + socket: tokio::net::TcpStream, packet_processor: PacketProcessor, - forwarding_channel: mpsc::UnboundedSender<(SocketAddr, Vec)>, + forwarding_channel: mpsc::UnboundedSender<(SocketAddr, SphinxPacket)>, ) { - let mut buf = [0u8; nymsphinx::PACKET_SIZE]; - loop { - match socket.read_exact(&mut buf).await { - // socket closed - Ok(n) if n == 0 => { - trace!("Remote connection closed."); - return; - } - Ok(n) => { - // If I understand it correctly, this if should never be executed as if `read_exact` - // does not fill buffer, it will throw UnexpectedEof? - if n != nymsphinx::PACKET_SIZE { - warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, nymsphinx::PACKET_SIZE); - continue; - } - // we must be able to handle multiple packets from same connection independently + let mut framed = Framed::new(socket, SphinxCodec); + while let Some(sphinx_packet) = framed.next().await { + match sphinx_packet { + Ok(sphinx_packet) => { + // we *really* need a worker pool here, because if we receive too many packets, + // we will spawn too many tasks and starve CPU due to context switching. + // (because presumably tokio has some concept of context switching in its + // scheduler) tokio::spawn(process_received_packet( - buf, - // note: processing_data is relatively cheap (and safe) to clone - - // it contains arc to private key and metrics reporter (which is just - // a single mpsc unbounded sender) + sphinx_packet, packet_processor.clone(), forwarding_channel.clone(), - )) + )); } - Err(e) => { - if e.kind() == io::ErrorKind::UnexpectedEof { - debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\ - Also closing the connection on this end.", socket.peer_addr()) - } else { - warn!( - "failed to read from socket (source: {:?}). Closing the connection; err = {:?}", - socket.peer_addr(), - e - ); - } + Err(err) => { + error!( + "The socket connection got corrupted with error: {:?}. Closing the socket", + err + ); return; } - }; + } } + info!( + "Closing connection from {:?}", + framed.into_inner().peer_addr() + ); } pub(crate) fn run_socket_listener( handle: &Handle, addr: SocketAddr, packet_processor: PacketProcessor, - forwarding_channel: mpsc::UnboundedSender<(SocketAddr, Vec)>, + forwarding_channel: mpsc::UnboundedSender<(SocketAddr, SphinxPacket)>, ) -> JoinHandle> { let handle_clone = handle.clone(); handle.spawn(async move { diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 7002ea2d49..34089d3474 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -18,6 +18,7 @@ use crypto::encryption; use directory_client::presence::Topology; use futures::channel::mpsc; use log::*; +use nymsphinx::SphinxPacket; use std::net::SocketAddr; use tokio::runtime::Runtime; use topology::NymTopology; @@ -71,7 +72,7 @@ impl MixNode { fn start_socket_listener( &self, metrics_reporter: metrics::MetricsReporter, - forwarding_channel: mpsc::UnboundedSender<(SocketAddr, Vec)>, + forwarding_channel: mpsc::UnboundedSender<(SocketAddr, SphinxPacket)>, ) { info!("Starting socket listener..."); // this is the only location where our private key is going to be copied @@ -87,7 +88,7 @@ impl MixNode { ); } - fn start_packet_forwarder(&mut self) -> mpsc::UnboundedSender<(SocketAddr, Vec)> { + fn start_packet_forwarder(&mut self) -> mpsc::UnboundedSender<(SocketAddr, SphinxPacket)> { info!("Starting packet forwarder..."); self.runtime .enter(|| { diff --git a/mixnode/src/node/packet_forwarding.rs b/mixnode/src/node/packet_forwarding.rs index f67e814f91..71f9bd972f 100644 --- a/mixnode/src/node/packet_forwarding.rs +++ b/mixnode/src/node/packet_forwarding.rs @@ -15,14 +15,15 @@ use futures::channel::mpsc; use futures::StreamExt; use log::*; +use nymsphinx::SphinxPacket; use std::net::SocketAddr; use std::time::Duration; use tokio::runtime::Handle; pub(crate) struct PacketForwarder { - tcp_client: multi_tcp_client::Client, - conn_tx: mpsc::UnboundedSender<(SocketAddr, Vec)>, - conn_rx: mpsc::UnboundedReceiver<(SocketAddr, Vec)>, + tcp_client: mixnet_client::Client, + conn_tx: mpsc::UnboundedSender<(SocketAddr, SphinxPacket)>, + conn_rx: mpsc::UnboundedReceiver<(SocketAddr, SphinxPacket)>, } impl PacketForwarder { @@ -31,7 +32,7 @@ impl PacketForwarder { maximum_reconnection_backoff: Duration, initial_connection_timeout: Duration, ) -> PacketForwarder { - let tcp_client_config = multi_tcp_client::Config::new( + let tcp_client_config = mixnet_client::Config::new( initial_reconnection_backoff, maximum_reconnection_backoff, initial_connection_timeout, @@ -40,13 +41,16 @@ impl PacketForwarder { let (conn_tx, conn_rx) = mpsc::unbounded(); PacketForwarder { - tcp_client: multi_tcp_client::Client::new(tcp_client_config), + tcp_client: mixnet_client::Client::new(tcp_client_config), conn_tx, conn_rx, } } - pub(crate) fn start(mut self, handle: &Handle) -> mpsc::UnboundedSender<(SocketAddr, Vec)> { + pub(crate) fn start( + mut self, + handle: &Handle, + ) -> mpsc::UnboundedSender<(SocketAddr, SphinxPacket)> { // TODO: what to do with the lost JoinHandle? let sender_channel = self.conn_tx.clone(); handle.spawn(async move { diff --git a/mixnode/src/node/packet_processing.rs b/mixnode/src/node/packet_processing.rs index adba39dc64..e859f4001f 100644 --- a/mixnode/src/node/packet_processing.rs +++ b/mixnode/src/node/packet_processing.rs @@ -32,7 +32,7 @@ pub enum MixProcessingError { } pub enum MixProcessingResult { - ForwardHop(SocketAddr, Vec), + ForwardHop(SocketAddr, SphinxPacket), #[allow(dead_code)] LoopMessage, } @@ -87,21 +87,16 @@ impl PacketProcessor { // Delay packet for as long as required tokio::time::delay_for(delay.to_duration()).await; - Ok(MixProcessingResult::ForwardHop( - next_hop_address, - packet.to_bytes(), - )) + Ok(MixProcessingResult::ForwardHop(next_hop_address, packet)) } pub(crate) async fn process_sphinx_packet( &self, - raw_packet_data: [u8; nymsphinx::PACKET_SIZE], + packet: SphinxPacket, ) -> Result { // we received something resembling a sphinx packet, report it! self.metrics_reporter.report_received(); - let packet = SphinxPacket::from_bytes(&raw_packet_data)?; - match packet.process(self.secret_key.deref().inner()) { Ok(ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay)) => { self.process_forward_hop(packet, address, delay).await