From a77881d2844718f0382ec01abeb208f3926569df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 23 Apr 2020 10:36:13 +0100 Subject: [PATCH] Intermediate gateway-heart surgery checkpoint (#199) * Initial draft for ClientsHandler * Created listener struct * typo * Stateful websocket connection handler * Exposing modules * Depdendencies updates * Moved listener to correct file + made start consume listener * Main starting new listener * Catching sigint * Copied client storage from provider into gateway * Exposed websocket listener type for nicer import path * Defined websocket message receiver concrete type * Client ledger struct without implementation * ClientsHandler using more concrete types * Mixnet sender + receiver and exposed listener type * Handling mix packets * Ability to forward mix packets * "starting" both listeners at main * Depedencies updates * Initial type definitions for client messages --- Cargo.lock | 7 +- gateway/Cargo.toml | 10 +- .../src/client_handling/clients_handler.rs | 65 +++++ gateway/src/client_handling/ledger.rs | 162 +++++++++++++ gateway/src/client_handling/mod.rs | 3 + .../websocket/connection_handler.rs | 74 ++++++ .../src/client_handling/websocket/listener.rs | 58 +++++ .../websocket/message_receiver.rs | 4 + gateway/src/client_handling/websocket/mod.rs | 6 + .../src/client_handling/websocket/types.rs | 42 ++++ gateway/src/listener.rs | 42 ---- gateway/src/main.rs | 51 ++-- gateway/src/mixnet_handling/mod.rs | 4 + .../receiver/connection_handler.rs | 88 +++++++ .../src/mixnet_handling/receiver/listener.rs | 51 ++++ gateway/src/mixnet_handling/receiver/mod.rs | 17 ++ .../receiver/packet_processing.rs | 217 +++++++++++++++++ gateway/src/mixnet_handling/sender/mod.rs | 63 +++++ gateway/src/storage/mod.rs | 222 ++++++++++++++++++ 19 files changed, 1114 insertions(+), 72 deletions(-) create mode 100644 gateway/src/client_handling/clients_handler.rs create mode 100644 gateway/src/client_handling/ledger.rs create mode 100644 gateway/src/client_handling/mod.rs create mode 100644 gateway/src/client_handling/websocket/connection_handler.rs create mode 100644 gateway/src/client_handling/websocket/listener.rs create mode 100644 gateway/src/client_handling/websocket/message_receiver.rs create mode 100644 gateway/src/client_handling/websocket/mod.rs create mode 100644 gateway/src/client_handling/websocket/types.rs delete mode 100644 gateway/src/listener.rs create mode 100644 gateway/src/mixnet_handling/mod.rs create mode 100644 gateway/src/mixnet_handling/receiver/connection_handler.rs create mode 100644 gateway/src/mixnet_handling/receiver/listener.rs create mode 100644 gateway/src/mixnet_handling/receiver/mod.rs create mode 100644 gateway/src/mixnet_handling/receiver/packet_processing.rs create mode 100644 gateway/src/mixnet_handling/sender/mod.rs create mode 100644 gateway/src/storage/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 13a9d7f4a2..e29462985d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -899,14 +899,17 @@ dependencies = [ name = "gateway" version = "0.1.0" dependencies = [ + "crypto", "dotenv", "futures 0.3.4", - "futures-channel", - "futures-util", "log 0.4.8", + "mix-client", "multi-tcp-client", "nymsphinx", "pretty_env_logger", + "rand 0.7.3", + "serde", + "sled", "tokio 0.2.16", "tokio-tungstenite", "tungstenite", diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index dcbed977b9..0fc5e0c9cb 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -9,15 +9,19 @@ edition = "2018" [dependencies] dotenv = "0.15.0" futures = "0.3" -futures-channel = "0.3" -futures-util = { version = "0.3", default-features = false, features = ["async-await", "sink", "std"] } log = "0.4" -multi-tcp-client = { path = "../common/client-libs/multi-tcp-client" } pretty_env_logger = "0.3" +rand = "0.7.2" +serde = { version = "1.0.104", features = ["derive"] } +#serde_json = "1.0.44" +sled = "0.31" tokio = { version = "0.2", features = ["full"] } tokio-tungstenite = "0.10.1" # internal +crypto = { path = "../common/crypto" } +mix-client = { path = "../common/client-libs/mix-client" } # to be removed very soon +multi-tcp-client = { path = "../common/client-libs/multi-tcp-client" } nymsphinx = { path = "../common/nymsphinx" } diff --git a/gateway/src/client_handling/clients_handler.rs b/gateway/src/client_handling/clients_handler.rs new file mode 100644 index 0000000000..91bd1aba0e --- /dev/null +++ b/gateway/src/client_handling/clients_handler.rs @@ -0,0 +1,65 @@ +use crate::client_handling::ledger::ClientLedger; +use crate::client_handling::websocket::message_receiver::MixMessageSender; +use futures::channel::{mpsc, oneshot}; +use futures::StreamExt; +use log::*; +use nymsphinx::DestinationAddressBytes; +use std::collections::HashMap; +use tokio::task::JoinHandle; + +type temp_AuthToken = String; + +pub(crate) type ClientsHandlerRequestSender = mpsc::UnboundedSender; +pub(crate) type ClientsHandlerRequestReceiver = mpsc::UnboundedReceiver; + +pub(crate) type ClientsHandlerResponseSender = oneshot::Sender; +pub(crate) type ClientsHandlerResponseReceiver = oneshot::Receiver; + +pub(crate) enum ClientsHandlerRequest { + // client + Register( + DestinationAddressBytes, + MixMessageSender, + ClientsHandlerResponseSender, + ), + Authenticate( + temp_AuthToken, + MixMessageSender, + ClientsHandlerResponseSender, + ), + + // mix + IsOnline(DestinationAddressBytes, ClientsHandlerResponseSender), +} + +pub(crate) enum ClientsHandlerResponse { + Register(Option), + Authenticate(bool), + IsOnline(Option), +} + +pub(crate) struct ClientsHandler { + open_connections: HashMap, // clients_ledger: unimplemented!(), + clients_ledger: ClientLedger, + request_receiver_channel: ClientsHandlerRequestReceiver, +} + +impl ClientsHandler { + pub(crate) fn new(request_receiver_channel: ClientsHandlerRequestReceiver) -> Self { + ClientsHandler { + open_connections: HashMap::new(), + request_receiver_channel, + clients_ledger: unimplemented!(), + } + } + + pub(crate) async fn run(&mut self) { + while let Some(request) = self.request_receiver_channel.next().await { + // handle request + } + } + + pub(crate) fn start(mut self) -> JoinHandle<()> { + tokio::spawn(async move { self.run().await }) + } +} diff --git a/gateway/src/client_handling/ledger.rs b/gateway/src/client_handling/ledger.rs new file mode 100644 index 0000000000..dd19b429c6 --- /dev/null +++ b/gateway/src/client_handling/ledger.rs @@ -0,0 +1,162 @@ +// 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 directory_client::presence::providers::MixProviderClient; +//use log::*; +//use nymsphinx::{DestinationAddressBytes, DESTINATION_ADDRESS_LENGTH}; +//use sfw_provider_requests::auth_token::{AuthToken, AUTH_TOKEN_SIZE}; +//use std::path::PathBuf; + +#[derive(Debug)] +pub(crate) enum ClientLedgerError { + DbReadError(sled::Error), + DbWriteError(sled::Error), + DbOpenError(sled::Error), +} + +#[derive(Debug, Clone)] +// Note: you should NEVER create more than a single instance of this using 'new()'. +// You should always use .clone() to create additional instances +pub(crate) struct ClientLedger { + db: sled::Db, +} +// +//impl ClientLedger { +// pub(crate) fn load(file: PathBuf) -> Result { +// let db = match sled::open(file) { +// Err(e) => return Err(ClientLedgerError::DbOpenError(e)), +// Ok(db) => db, +// }; +// +// let ledger = ClientLedger { db }; +// +// ledger.db.iter().keys().for_each(|key| { +// println!( +// "key: {:?}", +// ledger +// .read_destination_address_bytes(key.unwrap()) +// .to_base58_string() +// ); +// }); +// +// Ok(ledger) +// } +// +// fn read_auth_token(&self, raw_token: sled::IVec) -> AuthToken { +// let token_bytes_ref = raw_token.as_ref(); +// // if this fails it means we have some database corruption and we +// // absolutely can't continue +// if token_bytes_ref.len() != AUTH_TOKEN_SIZE { +// error!("CLIENT LEDGER DATA CORRUPTION - TOKEN HAS INVALID LENGTH"); +// panic!("CLIENT LEDGER DATA CORRUPTION - TOKEN HAS INVALID LENGTH"); +// } +// +// let mut token_bytes = [0u8; AUTH_TOKEN_SIZE]; +// token_bytes.copy_from_slice(token_bytes_ref); +// AuthToken::from_bytes(token_bytes) +// } +// +// fn read_destination_address_bytes( +// &self, +// raw_destination: sled::IVec, +// ) -> DestinationAddressBytes { +// let destination_ref = raw_destination.as_ref(); +// // if this fails it means we have some database corruption and we +// // absolutely can't continue +// if destination_ref.len() != DESTINATION_ADDRESS_LENGTH { +// error!("CLIENT LEDGER DATA CORRUPTION - CLIENT ADDRESS HAS INVALID LENGTH"); +// panic!("CLIENT LEDGER DATA CORRUPTION - CLIENT ADDRESS HAS INVALID LENGTH"); +// } +// +// let mut destination_bytes = [0u8; DESTINATION_ADDRESS_LENGTH]; +// destination_bytes.copy_from_slice(destination_ref); +// DestinationAddressBytes::from_bytes(destination_bytes) +// } +// +// pub(crate) fn verify_token( +// &self, +// auth_token: &AuthToken, +// client_address: &DestinationAddressBytes, +// ) -> Result { +// match self.db.get(&client_address.to_bytes()) { +// Err(e) => Err(ClientLedgerError::DbReadError(e)), +// Ok(token) => match token { +// Some(token_ivec) => Ok(&self.read_auth_token(token_ivec) == auth_token), +// None => Ok(false), +// }, +// } +// } +// +// pub(crate) fn insert_token( +// &mut self, +// auth_token: AuthToken, +// client_address: DestinationAddressBytes, +// ) -> Result, ClientLedgerError> { +// let insertion_result = match self +// .db +// .insert(&client_address.to_bytes(), &auth_token.to_bytes()) +// { +// Err(e) => Err(ClientLedgerError::DbWriteError(e)), +// Ok(existing_token) => { +// Ok(existing_token.map(|existing_token| self.read_auth_token(existing_token))) +// } +// }; +// +// // registration doesn't happen that often so might as well flush it to the disk to be sure +// self.db.flush().unwrap(); +// insertion_result +// } +// +// pub(crate) fn remove_token( +// &mut self, +// client_address: &DestinationAddressBytes, +// ) -> Result, ClientLedgerError> { +// let removal_result = match self.db.remove(&client_address.to_bytes()) { +// Err(e) => Err(ClientLedgerError::DbWriteError(e)), +// Ok(existing_token) => { +// Ok(existing_token.map(|existing_token| self.read_auth_token(existing_token))) +// } +// }; +// +// // removing of tokens happens extremely rarely, so flush is also fine here +// self.db.flush().unwrap(); +// removal_result +// } +// +// pub(crate) fn current_clients(&self) -> Result, ClientLedgerError> { +// let clients = self.db.iter().keys(); +// +// let mut client_vec = Vec::new(); +// for client in clients { +// match client { +// Err(e) => return Err(ClientLedgerError::DbWriteError(e)), +// Ok(client_entry) => client_vec.push(MixProviderClient { +// pub_key: self +// .read_destination_address_bytes(client_entry) +// .to_base58_string(), +// }), +// } +// } +// +// Ok(client_vec) +// } +// +// #[cfg(test)] +// pub(crate) fn create_temporary() -> Self { +// let cfg = sled::Config::new().temporary(true); +// ClientLedger { +// db: cfg.open().unwrap(), +// } +// } +//} diff --git a/gateway/src/client_handling/mod.rs b/gateway/src/client_handling/mod.rs new file mode 100644 index 0000000000..15706c869d --- /dev/null +++ b/gateway/src/client_handling/mod.rs @@ -0,0 +1,3 @@ +pub(crate) mod clients_handler; +pub(crate) mod ledger; +pub(crate) mod websocket; diff --git a/gateway/src/client_handling/websocket/connection_handler.rs b/gateway/src/client_handling/websocket/connection_handler.rs new file mode 100644 index 0000000000..fb83a1073d --- /dev/null +++ b/gateway/src/client_handling/websocket/connection_handler.rs @@ -0,0 +1,74 @@ +use crate::client_handling::clients_handler::ClientsHandlerRequestSender; +use log::*; +use tokio::{prelude::*, stream::StreamExt}; +use tokio_tungstenite::{ + tungstenite::{protocol::Message, Error as WsError}, + WebSocketStream, +}; + +enum SocketStream { + RawTCP(S), + UpgradedWebSocket(WebSocketStream), + Invalid, +} + +pub(crate) struct Handle { + socket_connection: SocketStream, + clients_handler_sender: ClientsHandlerRequestSender, +} + +impl Handle +where + S: AsyncRead + AsyncWrite + Unpin, +{ + // for time being we assume handle is always constructed from raw socket. + // if we decide we want to change it, that's not too difficult + pub(crate) fn new(conn: S, clients_handler_sender: ClientsHandlerRequestSender) -> Self { + Handle { + socket_connection: SocketStream::RawTCP(conn), + clients_handler_sender, + } + } + + async fn perform_websocket_handshake(&mut self) { + self.socket_connection = + match std::mem::replace(&mut self.socket_connection, SocketStream::Invalid) { + SocketStream::RawTCP(conn) => { + // TODO: perhaps in the future, rather than panic here (and uncleanly shut tcp stream) + // return a result with an error? + let ws_stream = tokio_tungstenite::accept_async(conn) + .await + .expect("Failed to perform websocket handshake"); + SocketStream::UpgradedWebSocket(ws_stream) + } + other => other, + } + } + + async fn next_websocket_request(&mut self) -> Option> { + match self.socket_connection { + SocketStream::UpgradedWebSocket(ref mut ws_stream) => ws_stream.next().await, + _ => panic!("impossible state - websocket handshake was somehow reverted"), + } + } + + async fn listen_for_requests(&mut self) { + trace!("Started listening for incoming requests..."); + + while let Some(msg) = self.next_websocket_request().await { + // start handling here + // let msg = msg?; + // if msg.is_binary() { + // mixnet_client::forward_to_mixnode(msg.into_data(), Arc::clone(&client_ref)).await; + // } + } + + trace!("The stream was closed!"); + } + + pub(crate) async fn start_handling(&mut self) { + self.perform_websocket_handshake().await; + trace!("Managed to perform websocket handshake!"); + self.listen_for_requests().await; + } +} diff --git a/gateway/src/client_handling/websocket/listener.rs b/gateway/src/client_handling/websocket/listener.rs new file mode 100644 index 0000000000..e7cef58a97 --- /dev/null +++ b/gateway/src/client_handling/websocket/listener.rs @@ -0,0 +1,58 @@ +// 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::client_handling::clients_handler::ClientsHandlerRequestSender; +use crate::client_handling::websocket::connection_handler::Handle; +use log::*; +use std::net::SocketAddr; +use tokio::task::JoinHandle; + +pub(crate) struct Listener { + address: SocketAddr, + clients_handler_sender: ClientsHandlerRequestSender, +} + +impl Listener { + pub(crate) fn new( + address: SocketAddr, + clients_handler_sender: ClientsHandlerRequestSender, + ) -> Self { + Listener { + address, + clients_handler_sender, + } + } + + pub(crate) async fn run(&mut self) { + info!("Starting websocket listener at {}", self.address); + let mut tcp_listener = tokio::net::TcpListener::bind(self.address) + .await + .expect("Failed to start websocket listener"); + + loop { + match tcp_listener.accept().await { + Ok((socket, remote_addr)) => { + trace!("received a socket connection from {}", remote_addr); + let mut handle = Handle::new(socket, self.clients_handler_sender.clone()); + tokio::spawn(async move { handle.start_handling().await }); + } + Err(e) => warn!("failed to get client: {:?}", e), + } + } + } + + pub(crate) fn start(mut self) -> JoinHandle<()> { + tokio::spawn(async move { self.run().await }) + } +} diff --git a/gateway/src/client_handling/websocket/message_receiver.rs b/gateway/src/client_handling/websocket/message_receiver.rs new file mode 100644 index 0000000000..568b0d367d --- /dev/null +++ b/gateway/src/client_handling/websocket/message_receiver.rs @@ -0,0 +1,4 @@ +use futures::channel::mpsc; + +pub(crate) type MixMessageSender = mpsc::UnboundedSender>>; +pub(crate) type MixMessageReceiver = mpsc::UnboundedReceiver>>; diff --git a/gateway/src/client_handling/websocket/mod.rs b/gateway/src/client_handling/websocket/mod.rs new file mode 100644 index 0000000000..4ff6f227a0 --- /dev/null +++ b/gateway/src/client_handling/websocket/mod.rs @@ -0,0 +1,6 @@ +pub(crate) mod connection_handler; +pub(crate) mod listener; +pub(crate) mod types; +pub(crate) mod message_receiver; + +pub(crate) use listener::Listener; diff --git a/gateway/src/client_handling/websocket/types.rs b/gateway/src/client_handling/websocket/types.rs new file mode 100644 index 0000000000..1fcf2d135c --- /dev/null +++ b/gateway/src/client_handling/websocket/types.rs @@ -0,0 +1,42 @@ +// 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 serde::{Deserialize, Serialize}; +use std::convert::TryFrom; +use tokio_tungstenite::tungstenite::protocol::Message; + +#[derive(Serialize, Deserialize, Debug)] +pub(crate) enum Request { + Send, + Register { + address: String + }, + Authenticate { + token: String + }, +} + +#[derive(Serialize, Deserialize, Debug)] +pub(crate) enum Response { + Send, + Register { + token: String + }, + Authenticate { + status: bool + }, + Error { + message: String + }, +} \ No newline at end of file diff --git a/gateway/src/listener.rs b/gateway/src/listener.rs deleted file mode 100644 index cb915667bc..0000000000 --- a/gateway/src/listener.rs +++ /dev/null @@ -1,42 +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 super::mixnet_client; -use futures::lock::Mutex; -use futures_util::StreamExt; -use log::*; -use multi_tcp_client::Client as MultiClient; -use std::net::SocketAddr; -use std::sync::Arc; -use tokio::net::TcpStream; -use tokio_tungstenite::accept_async; -use tungstenite::Result; - -pub async fn handle_connection( - peer: SocketAddr, - stream: TcpStream, - client_ref: Arc>, -) -> Result<()> { - let mut ws_stream = accept_async(stream).await.expect("Failed to accept"); - - info!("New WebSocket connection: {}", peer); - - while let Some(msg) = ws_stream.next().await { - let msg = msg?; - if msg.is_binary() { - mixnet_client::forward_to_mixnode(msg.into_data(), Arc::clone(&client_ref)).await; - } - } - Ok(()) -} diff --git a/gateway/src/main.rs b/gateway/src/main.rs index b4ee2b4f8e..dff9e54de7 100644 --- a/gateway/src/main.rs +++ b/gateway/src/main.rs @@ -12,44 +12,45 @@ // See the License for the specific language governing permissions and // limitations under the License. -use futures::lock::Mutex; +use crate::client_handling::websocket; +use crate::mixnet_handling::receiver::packet_processing::PacketProcessor; +use crate::storage::ClientStorage; use log::*; -use multi_tcp_client::Client as MultiClient; -use std::net::SocketAddr; -use std::sync::Arc; -use tokio::net::{TcpListener, TcpStream}; -use tokio_tungstenite::tungstenite::Error; -mod listener; +mod client_handling; mod mixnet_client; - -async fn accept_connection(peer: SocketAddr, stream: TcpStream, client: Arc>) { - if let Err(e) = listener::handle_connection(peer, stream, client).await { - match e { - Error::ConnectionClosed | Error::Protocol(_) | Error::Utf8 => (), - err => error!("Error processing connection: {}", err), - } - } -} +mod mixnet_handling; +pub(crate) mod storage; #[tokio::main] async fn main() { dotenv::dotenv().ok(); setup_logging(); - let addr = "127.0.0.1:1793"; - let mut listener = TcpListener::bind(&addr).await.expect("Can't listen"); + let addr = "127.0.0.1:1793".parse().unwrap(); info!("Listening on: {}", addr); - let client_ref = mixnet_client::new(); + let (dummy_clients_handler_tx, _) = futures::channel::mpsc::unbounded(); + let client_storage = ClientStorage::new(42, 42, "foomp".into()); + let dummy_keypair = crypto::encryption::KeyPair::new(); + let dummy_mix_packet_processor = PacketProcessor::new( + dummy_keypair.private_key().to_owned(), + dummy_clients_handler_tx.clone(), + client_storage, + ); - while let Ok((stream, _)) = listener.accept().await { - let peer = stream - .peer_addr() - .expect("connected streams should have a peer address"); - info!("Peer address: {}", peer); + websocket::Listener::new(addr, dummy_clients_handler_tx.clone()).start(); + mixnet_handling::Listener::new(addr).start(dummy_mix_packet_processor); - tokio::spawn(accept_connection(peer, stream, Arc::clone(&client_ref))); + if let Err(e) = tokio::signal::ctrl_c().await { + error!( + "There was an error while capturing SIGINT - {:?}. We will terminate regardless", + e + ); } + + println!( + "Received SIGINT - the provider will terminate now (threads are not YET nicely stopped)" + ); } fn setup_logging() { diff --git a/gateway/src/mixnet_handling/mod.rs b/gateway/src/mixnet_handling/mod.rs new file mode 100644 index 0000000000..2209256561 --- /dev/null +++ b/gateway/src/mixnet_handling/mod.rs @@ -0,0 +1,4 @@ +pub(crate) mod receiver; +pub(crate) mod sender; + +pub(crate) use receiver::listener::Listener; diff --git a/gateway/src/mixnet_handling/receiver/connection_handler.rs b/gateway/src/mixnet_handling/receiver/connection_handler.rs new file mode 100644 index 0000000000..4255f9beff --- /dev/null +++ b/gateway/src/mixnet_handling/receiver/connection_handler.rs @@ -0,0 +1,88 @@ +use crate::client_handling::clients_handler::{ + ClientsHandlerRequest, ClientsHandlerRequestSender, ClientsHandlerResponse, +}; +use crate::client_handling::websocket::message_receiver::MixMessageSender; +use crate::mixnet_handling::receiver::packet_processing::PacketProcessor; +use futures::channel::{mpsc, oneshot}; +use log::*; +use nymsphinx::DestinationAddressBytes; +use std::collections::HashMap; +use std::net::SocketAddr; +use tokio::{io::AsyncReadExt, prelude::*}; + +pub(crate) struct Handle { + peer_address: SocketAddr, + socket_connection: S, + packet_processor: PacketProcessor, +} + +impl Handle +where + S: AsyncRead + AsyncWrite + Unpin + 'static, +{ + // for time being we assume handle is always constructed from raw socket. + // if we decide we want to change it, that's not too difficult + pub(crate) fn new( + peer_address: SocketAddr, + conn: S, + packet_processor: PacketProcessor, + ) -> Self { + Handle { + peer_address, + socket_connection: conn, + packet_processor, + } + } + + async fn process_received_packet( + packet_data: [u8; nymsphinx::PACKET_SIZE], + mut packet_processor: PacketProcessor, + ) { + match packet_processor.process_sphinx_packet(packet_data).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 + tokio::spawn(Self::process_received_packet( + buf, + 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 + ); + } + return; + } + }; + } + } +} diff --git a/gateway/src/mixnet_handling/receiver/listener.rs b/gateway/src/mixnet_handling/receiver/listener.rs new file mode 100644 index 0000000000..52a5c13cb7 --- /dev/null +++ b/gateway/src/mixnet_handling/receiver/listener.rs @@ -0,0 +1,51 @@ +// 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::mixnet_handling::receiver::connection_handler::Handle; +use crate::mixnet_handling::receiver::packet_processing::PacketProcessor; +use log::*; +use std::net::SocketAddr; +use tokio::task::JoinHandle; + +pub(crate) struct Listener { + address: SocketAddr, +} + +impl Listener { + pub(crate) fn new(address: SocketAddr) -> Self { + Listener { address } + } + + pub(crate) async fn run(&mut self, packet_processor: PacketProcessor) { + info!("Starting mixnet listener at {}", self.address); + let mut tcp_listener = tokio::net::TcpListener::bind(self.address) + .await + .expect("Failed to start mixnet listener"); + + loop { + match tcp_listener.accept().await { + Ok((socket, remote_addr)) => { + trace!("received a socket connection from {}", remote_addr); + let mut handle = Handle::new(remote_addr, socket, packet_processor.clone()); + tokio::spawn(async move { handle.start_handling().await }); + } + Err(e) => warn!("failed to get client: {:?}", e), + } + } + } + + pub(crate) fn start(mut self, packet_processor: PacketProcessor) -> JoinHandle<()> { + tokio::spawn(async move { self.run(packet_processor).await }) + } +} diff --git a/gateway/src/mixnet_handling/receiver/mod.rs b/gateway/src/mixnet_handling/receiver/mod.rs new file mode 100644 index 0000000000..2f49616cb2 --- /dev/null +++ b/gateway/src/mixnet_handling/receiver/mod.rs @@ -0,0 +1,17 @@ +// 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. + +pub(crate) mod connection_handler; +pub(crate) mod listener; +pub(crate) mod packet_processing; diff --git a/gateway/src/mixnet_handling/receiver/packet_processing.rs b/gateway/src/mixnet_handling/receiver/packet_processing.rs new file mode 100644 index 0000000000..64a2092ec2 --- /dev/null +++ b/gateway/src/mixnet_handling/receiver/packet_processing.rs @@ -0,0 +1,217 @@ +// 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::client_handling::clients_handler::{ + ClientsHandlerRequest, ClientsHandlerRequestSender, ClientsHandlerResponse, +}; +use crate::client_handling::websocket::message_receiver::MixMessageSender; +use crate::storage::{ClientStorage, StoreData}; +use crypto::encryption; +use futures::channel::oneshot; +use futures::lock::Mutex; +use log::*; +use mix_client::packet::LOOP_COVER_MESSAGE_PAYLOAD; +use nymsphinx::{DestinationAddressBytes, Payload, ProcessedPacket, SURBIdentifier, SphinxPacket}; +use std::collections::HashMap; +use std::io; +use std::ops::Deref; +use std::sync::Arc; + +#[derive(Debug)] +pub enum MixProcessingError { + ReceivedForwardHopError, + NonMatchingRecipient, + InvalidPayload, + SphinxProcessingError, + IOError(String), +} + +pub enum MixProcessingResult { + #[allow(dead_code)] + ForwardHop, + FinalHop, +} + +impl From for MixProcessingError { + // for time being just have a single error instance for all possible results of nymsphinx::ProcessingError + fn from(_: nymsphinx::ProcessingError) -> Self { + use MixProcessingError::*; + + SphinxProcessingError + } +} + +impl From for MixProcessingError { + fn from(e: io::Error) -> Self { + use MixProcessingError::*; + + IOError(e.to_string()) + } +} + +// PacketProcessor contains all data required to correctly unwrap and store sphinx packets +#[derive(Clone)] +pub struct PacketProcessor { + secret_key: Arc, + // TODO: later investigate some concurrent hashmap solutions or perhaps RWLocks. + // Right now Mutex is the simplest and fastest to implement approach + available_socket_senders_cache: Arc>>, + client_store: ClientStorage, + clients_handler_sender: ClientsHandlerRequestSender, +} + +impl PacketProcessor { + pub(crate) fn new( + secret_key: encryption::PrivateKey, + clients_handler_sender: ClientsHandlerRequestSender, + client_store: ClientStorage, + ) -> Self { + PacketProcessor { + available_socket_senders_cache: Arc::new(Mutex::new(HashMap::new())), + clients_handler_sender, + client_store, + secret_key: Arc::new(secret_key), + } + } + + fn try_push_message_to_client( + &self, + sender_channel: Option, + message: Vec, + ) -> bool { + match sender_channel { + None => false, + Some(sender_channel) => sender_channel.unbounded_send(vec![message]).is_ok(), + } + } + + async fn try_to_obtain_client_ws_message_sender( + &mut self, + client_address: DestinationAddressBytes, + ) -> Option { + let mut cache_guard = self.available_socket_senders_cache.lock().await; + + if let Some(sender) = cache_guard.get(&client_address) { + if !sender.is_closed() { + return Some(sender.clone()); + } else { + cache_guard.remove(&client_address); + } + } + + // do not block other readers to the cache while we are doing some blocking work here + drop(cache_guard); + + // if we got here it means that either we have no sender channel for this client or it's closed + // so we must refresh it from the source, i.e. ClientsHandler + let (res_sender, res_receiver) = oneshot::channel(); + let clients_handler_request = + ClientsHandlerRequest::IsOnline(client_address.clone(), res_sender); + self.clients_handler_sender + .unbounded_send(clients_handler_request) + .unwrap(); // the receiver MUST BE alive + + let client_sender = match res_receiver.await.unwrap() { + ClientsHandlerResponse::IsOnline(client_sender) => client_sender, + _ => panic!("received response to wrong query!"), // again, this should NEVER happen + }; + + if client_sender.is_none() { + return None; + } + + let client_sender = client_sender.unwrap(); + // finally re-acquire the lock to update the cache + let mut cache_guard = self.available_socket_senders_cache.lock().await; + cache_guard.insert(client_address, client_sender.clone()); + + Some(client_sender) + } + + pub(crate) async fn store_processed_packet_payload( + &self, + client_address: DestinationAddressBytes, + message: Vec, + ) -> io::Result<()> { + trace!( + "Storing received packet for {:?} on the disk...", + client_address.to_base58_string() + ); + // we are temporarily ignoring and not storing obvious loop cover traffic messages to + // not cause our sfw-provider to run out of disk space too quickly. + // Eventually this is going to get removed and be replaced by a quota system described in: + // https://github.com/nymtech/nym/issues/137 + if message == LOOP_COVER_MESSAGE_PAYLOAD { + debug!("Received a loop cover message - not going to store it"); + return Ok(()); + } + + let store_data = StoreData::new(client_address, message); + self.client_store.store_processed_data(store_data).await + } + + pub(crate) fn unwrap_sphinx_packet( + &self, + raw_packet_data: [u8; nymsphinx::PACKET_SIZE], + ) -> 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"); + Err(MixProcessingError::ReceivedForwardHopError) + } + Ok(ProcessedPacket::ProcessedPacketFinalHop(client_address, surb_id, payload)) => { + // in our current design, we do not care about the 'surb_id' in the header + // as it will always be empty anyway + let (payload_destination, message) = payload + .try_recover_destination_and_plaintext() + .ok_or_else(|| MixProcessingError::InvalidPayload)?; + if client_address != payload_destination { + return Err(MixProcessingError::NonMatchingRecipient); + } + Ok((client_address, message)) + } + Err(e) => { + warn!("Failed to unwrap Sphinx packet: {:?}", e); + Err(MixProcessingError::SphinxProcessingError) + } + } + } + + pub(crate) async fn process_sphinx_packet( + &mut self, + raw_packet_data: [u8; nymsphinx::PACKET_SIZE], + ) -> Result<(), MixProcessingError> { + let (client_address, plaintext) = self.unwrap_sphinx_packet(raw_packet_data)?; + + let client_sender = self + .try_to_obtain_client_ws_message_sender(client_address.clone()) + .await; + + // TODO: think of a way to prevent having to clone the plaintext here, perhaps make channels use references? + // this will, again, take slightly more time, so it's an issue for later + if !self.try_push_message_to_client(client_sender, plaintext.clone()) { + Ok(self + .store_processed_packet_payload(client_address, plaintext) + .await?) + } else { + trace!( + "Managed to push received packet for {:?} to websocket connection!", + client_address.to_base58_string() + ); + Ok(()) + } + } +} diff --git a/gateway/src/mixnet_handling/sender/mod.rs b/gateway/src/mixnet_handling/sender/mod.rs new file mode 100644 index 0000000000..e6dc32e0f0 --- /dev/null +++ b/gateway/src/mixnet_handling/sender/mod.rs @@ -0,0 +1,63 @@ +// 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. + +// TODO: code is nearly identical to mixnode::node::packet_forwarding -> perhaps it should be put to common? + +use futures::channel::mpsc; +use futures::StreamExt; +use log::*; +use std::net::SocketAddr; +use std::time::Duration; + +pub(crate) struct PacketForwarder { + tcp_client: multi_tcp_client::Client, + conn_tx: mpsc::UnboundedSender<(SocketAddr, Vec)>, + conn_rx: mpsc::UnboundedReceiver<(SocketAddr, Vec)>, +} + +impl PacketForwarder { + pub(crate) fn new( + initial_reconnection_backoff: Duration, + maximum_reconnection_backoff: Duration, + initial_connection_timeout: Duration, + ) -> PacketForwarder { + let tcp_client_config = multi_tcp_client::Config::new( + initial_reconnection_backoff, + maximum_reconnection_backoff, + initial_connection_timeout, + ); + + let (conn_tx, conn_rx) = mpsc::unbounded(); + + PacketForwarder { + tcp_client: multi_tcp_client::Client::new(tcp_client_config), + conn_tx, + conn_rx, + } + } + + pub(crate) fn start(mut self) -> mpsc::UnboundedSender<(SocketAddr, Vec)> { + // TODO: what to do with the lost JoinHandle - do we even care? + let sender_channel = self.conn_tx.clone(); + tokio::spawn(async move { + while let Some((address, packet)) = self.conn_rx.next().await { + 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(); // if we're not waiting for response, we MUST get an Ok + } + }); + sender_channel + } +} diff --git a/gateway/src/storage/mod.rs b/gateway/src/storage/mod.rs new file mode 100644 index 0000000000..b488dea5c8 --- /dev/null +++ b/gateway/src/storage/mod.rs @@ -0,0 +1,222 @@ +// 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::lock::Mutex; +use futures::StreamExt; +use log::*; +use nymsphinx::{DestinationAddressBytes, SURBIdentifier}; +use rand::Rng; +//use sfw_provider_requests::DUMMY_MESSAGE_CONTENT; +use std::io; +use std::path::PathBuf; +use std::sync::Arc; +use tokio::fs; +use tokio::fs::File; +use tokio::prelude::*; + +const DUMMY_MESSAGE_CONTENT: &'static [u8] = b"foomp"; + +fn dummy_message() -> ClientFile { + ClientFile { + content: DUMMY_MESSAGE_CONTENT.to_vec(), + path: Default::default(), + } +} + +#[derive(Clone, Debug)] +pub struct ClientFile { + content: Vec, + path: PathBuf, +} + +impl ClientFile { + fn new(content: Vec, path: PathBuf) -> Self { + ClientFile { content, path } + } + + pub(crate) fn into_tuple(self) -> (Vec, PathBuf) { + (self.content, self.path) + } +} + +pub struct StoreData { + client_address: DestinationAddressBytes, + message: Vec, +} + +impl StoreData { + pub(crate) fn new(client_address: DestinationAddressBytes, message: Vec) -> Self { + StoreData { + client_address, + message, + } + } +} + +// TODO: replace with proper database... +// Note: you should NEVER create more than a single instance of this using 'new()'. +// You should always use .clone() to create additional instances +#[derive(Clone, Debug)] +pub struct ClientStorage { + inner: Arc>, +} + +// even though the data inside is extremely cheap to copy, we have to have a single mutex, +// so might as well store the data behind it +pub struct ClientStorageInner { + message_retrieval_limit: usize, + filename_length: u16, + main_store_path_dir: PathBuf, +} + +// TODO: change it to some generic implementation to inject fs (or even better - proper database) +impl ClientStorage { + pub(crate) fn new(message_limit: usize, filename_len: u16, main_store_dir: PathBuf) -> Self { + ClientStorage { + inner: Arc::new(Mutex::new(ClientStorageInner { + message_retrieval_limit: message_limit, + filename_length: filename_len, + main_store_path_dir: main_store_dir, + })), + } + } + + // TODO: does this method really require locking? + // The worst that can happen is client sending 2 requests: to pull messages and register + // if register does not lock, then under specific timing pull messages will fail, + // but can simply be retried with no issues + pub(crate) async fn create_storage_dir( + &self, + client_address: DestinationAddressBytes, + ) -> io::Result<()> { + let inner_data = self.inner.lock().await; + + let client_dir_name = client_address.to_base58_string(); + let full_store_dir = inner_data.main_store_path_dir.join(client_dir_name); + fs::create_dir_all(full_store_dir).await + } + + pub(crate) fn generate_random_file_name(length: usize) -> String { + rand::thread_rng() + .sample_iter(&rand::distributions::Alphanumeric) + .take(length) + .collect::() + } + + pub(crate) async fn store_processed_data(&self, store_data: StoreData) -> io::Result<()> { + let inner_data = self.inner.lock().await; + + let client_dir_name = store_data.client_address.to_base58_string(); + let full_store_dir = inner_data.main_store_path_dir.join(client_dir_name); + let full_store_path = full_store_dir.join(Self::generate_random_file_name( + inner_data.filename_length as usize, + )); + debug!( + "going to store: {:?} in file: {:?}", + store_data.message, full_store_path + ); + + let mut file = File::create(full_store_path).await?; + file.write_all(store_data.message.as_ref()).await + } + + pub(crate) async fn retrieve_client_files( + &self, + client_address: DestinationAddressBytes, + ) -> io::Result> { + let inner_data = self.inner.lock().await; + + let client_dir_name = client_address.to_base58_string(); + let full_store_dir = inner_data.main_store_path_dir.join(client_dir_name); + + trace!("going to lookup: {:?}!", full_store_dir); + if !full_store_dir.exists() { + return Err(io::Error::new( + io::ErrorKind::NotFound, + "Target client does not exist", + )); + } + + let mut msgs = Vec::new(); + let mut read_dir = fs::read_dir(full_store_dir).await?; + + while let Some(dir_entry) = read_dir.next().await { + if let Ok(dir_entry) = dir_entry { + if !Self::is_valid_file(&dir_entry).await { + continue; + } + // Do not delete the file itself here! + // Only do it after client has received it + let client_file = + ClientFile::new(fs::read(dir_entry.path()).await?, dir_entry.path()); + msgs.push(client_file) + } + if msgs.len() == inner_data.message_retrieval_limit { + break; + } + } + + let dummy_message = dummy_message(); + + // make sure we always return as many messages as we need + if msgs.len() != inner_data.message_retrieval_limit as usize { + msgs = msgs + .into_iter() + .chain(std::iter::repeat(dummy_message)) + .take(inner_data.message_retrieval_limit) + .collect(); + } + + Ok(msgs) + } + + async fn is_valid_file(entry: &fs::DirEntry) -> bool { + let metadata = match entry.metadata().await { + Ok(meta) => meta, + Err(e) => { + error!( + "potentially corrupted client inbox! ({:?} - failed to read its metadata - {:?}", + entry.path(), + e, + ); + return false; + } + }; + + let is_file = metadata.is_file(); + if !is_file { + error!( + "potentially corrupted client inbox! - found a non-file - {:?}", + entry.path() + ); + } + + is_file + } + + pub(crate) async fn delete_files(&self, file_paths: Vec) -> io::Result<()> { + let dummy_message = dummy_message(); + let _guard = self.inner.lock().await; + + for file_path in file_paths { + if file_path == dummy_message.path { + continue; + } + if let Err(e) = fs::remove_file(file_path).await { + error!("Failed to delete client message! - {:?}", e) + } + } + Ok(()) + } +}