GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feature/sphinx socket packet encoder (#245)

Browse Source 
* Ability to send sphinx packets of different sizes + more efficient decoding

* Closing connection on connection corruption

* Missing semicolons

* Missing license notices

* Default for packetsize
This commit is contained in:
Jędrzej Stuczyński
2020-05-27 10:08:08 +01:00
committed by GitHub
parent aeb1a4b22e
commit 860a69b246
30 changed files with 817 additions and 492 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+19 -13
View File
@@ -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]]
Cargo.toml
+1 -1
View File
@@ -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",
clients/desktop/Cargo.toml
+1 -1
View File
@@ -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" }
clients/desktop/src/client/mix_traffic.rs
+1 -1
View File
@@ -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),
common/client-libs/gateway-client/src/lib.rs
+2 -2
View File
@@ -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<u8>,
packet: SphinxPacket,
) -> Result<(), GatewayClientError> {
if !self.authenticated {
return Err(GatewayClientError::NotAuthenticated);
common/client-libs/multi-tcp-client/Cargo.toml → common/client-libs/mixnet-client/Cargo.toml
+5 -1
View File
@@ -1,5 +1,5 @@
[package]
name = "multi-tcp-client"
name = "mixnet-client"
version = "0.1.0"
authors = ["Jedrzej Stuczynski <andrew@nymtech.net>"]
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" }
common/client-libs/multi-tcp-client/src/connection_manager/mod.rs → common/client-libs/mixnet-client/src/connection_manager/mod.rs
+56 -16
View File
@@ -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<oneshot::Sender<io::Result<()>>>;
pub(crate) type ConnectionManagerSender = mpsc::UnboundedSender<(Vec<u8>, ResponseSender)>;
type ConnectionManagerReceiver = mpsc::UnboundedReceiver<(Vec<u8>, 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<SphinxPacket>,
}
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<u8>) -> 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<Vec<u8>>
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!();
}
common/client-libs/multi-tcp-client/src/connection_manager/reconnector.rs → common/client-libs/mixnet-client/src/connection_manager/reconnector.rs
View File
common/client-libs/mixnet-client/src/connection_manager/writer.rs
+52
View File
@@ -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<tokio::net::TcpStream, SphinxCodec>,
}
impl ConnectionWriter {
pub(crate) fn new(connection: tokio::net::TcpStream) -> Self {
ConnectionWriter {
framed_connection: Framed::new(connection, SphinxCodec),
}
}
}
impl Sink<SphinxPacket> for ConnectionWriter {
type Error = SphinxCodecError;
fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
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<Result<(), Self::Error>> {
Pin::new(&mut self.framed_connection).poll_flush(cx)
}
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Pin::new(&mut self.framed_connection).poll_close(cx)
}
}
common/client-libs/mixnet-client/src/lib.rs
+281
View File
@@ -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<SocketAddr, (ConnectionManagerSender, AbortHandle)>,
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<Vec<u8>>,
// 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<Vec<u8>> {
// 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);
// }
// }
common/client-libs/multi-tcp-client/src/connection_manager/writer.rs
-60
View File
@@ -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<io::Result<usize>> {
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<io::Result<()>> {
use tokio::io::AsyncWrite;
Pin::new(&mut self.connection).poll_flush(cx)
}
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
use tokio::io::AsyncWrite;
Pin::new(&mut self.connection).poll_shutdown(cx)
}
}
common/client-libs/multi-tcp-client/src/lib.rs
-274
View File
@@ -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<SocketAddr, (ConnectionManagerSender, AbortHandle)>,
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<u8>,
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<Vec<u8>>,
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<Vec<u8>> {
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);
}
}
common/healthcheck/Cargo.toml
+1 -1
View File
@@ -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" }
common/healthcheck/src/path_check.rs
+4 -4
View File
@@ -69,7 +69,7 @@ pub enum PathStatus {
pub(crate) struct PathChecker {
provider_clients: HashMap<[u8; 32], Option<ProviderClient>>,
mixnet_client: multi_tcp_client::Client,
mixnet_client: mixnet_client::Client,
paths_status: HashMap<Vec<u8>, 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) => {
common/nymsphinx/Cargo.toml
+2
View File
@@ -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" }
common/nymsphinx/src/chunking/mod.rs
+46
View File
@@ -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<Vec<u8>> {
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.
common/nymsphinx/src/framing/mod.rs
+147
View File
@@ -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<io::Error> for SphinxCodecError {
fn from(err: io::Error) -> Self {
SphinxCodecError::IoError(err)
}
}
impl Into<io::Error> 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<InvalidPacketSize> 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<SphinxPacket> 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<Option<Self::Item>, 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<Result<Option<SphinxPacket>, SphinxCodecError>> {
let mut result = Vec::new();
loop {
match codec.decode(bytes) {
Ok(None) => {
break;
}
output => result.push(output),
}
}
return result;
}
}
common/nymsphinx/src/lib.rs
+3
View File
@@ -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;
common/nymsphinx/src/packets.rs
+76
View File
@@ -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<u8> for PacketSize {
type Error = InvalidPacketSize;
fn try_from(value: u8) -> std::result::Result<Self, Self::Error> {
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<Self, InvalidPacketSize> {
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
}
}
gateway/Cargo.toml
+2 -1
View File
@@ -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" }
gateway/gateway-requests/src/types.rs
+23 -7
View File
@@ -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<String> for ServerResponse {
}
pub enum BinaryRequest {
ForwardSphinx { address: SocketAddr, data: Vec<u8> },
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<u8> {
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<u8>) -> BinaryRequest {
BinaryRequest::ForwardSphinx { address, data }
pub fn new_forward_request(address: SocketAddr, sphinx_packet: SphinxPacket) -> BinaryRequest {
BinaryRequest::ForwardSphinx {
address,
sphinx_packet,
}
}
}
gateway/src/node/client_handling/websocket/connection_handler.rs
+5 -2
View File
@@ -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 }
}
gateway/src/node/mixnet_handling/receiver/connection_handler.rs
+27 -38
View File
@@ -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<S: AsyncRead + AsyncWrite + Unpin> {
peer_address: SocketAddr,
socket_connection: S,
framed_connection: Framed<S, SphinxCodec>,
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);
}
}
gateway/src/node/mixnet_handling/receiver/packet_processing.rs
+3 -5
View File
@@ -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<u8>), 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())
gateway/src/node/mixnet_handling/sender/mod.rs
+10 -6
View File
@@ -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<u8>)>;
pub(crate) type OutboundMixMessageReceiver = mpsc::UnboundedReceiver<(SocketAddr, Vec<u8>)>;
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
}
}),
mixnode/Cargo.toml
+2 -1
View File
@@ -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"}
mixnode/src/node/listener.rs
+32 -42
View File
@@ -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<u8>)>,
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<u8>)>,
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<u8>)>,
forwarding_channel: mpsc::UnboundedSender<(SocketAddr, SphinxPacket)>,
) -> JoinHandle<io::Result<()>> {
let handle_clone = handle.clone();
handle.spawn(async move {
mixnode/src/node/mod.rs
+3 -2
View File
@@ -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<u8>)>,
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<u8>)> {
fn start_packet_forwarder(&mut self) -> mpsc::UnboundedSender<(SocketAddr, SphinxPacket)> {
info!("Starting packet forwarder...");
self.runtime
.enter(|| {
mixnode/src/node/packet_forwarding.rs
+10 -6
View File
@@ -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<u8>)>,
conn_rx: mpsc::UnboundedReceiver<(SocketAddr, Vec<u8>)>,
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<u8>)> {
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 {
mixnode/src/node/packet_processing.rs
+3 -8
View File
@@ -32,7 +32,7 @@ pub enum MixProcessingError {
}
pub enum MixProcessingResult {
ForwardHop(SocketAddr, Vec<u8>),
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<MixProcessingResult, MixProcessingError> {
// 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