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

Intermediate gateway-heart surgery checkpoint (#199)

Browse Source 
* 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
This commit is contained in:
Jędrzej Stuczyński
2020-04-23 10:36:13 +01:00
committed by GitHub
parent 8baa236d81
commit a77881d284
19 changed files with 1114 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+5 -2
View File
@@ -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",
gateway/Cargo.toml
+7 -3
View File
@@ -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" }
gateway/src/client_handling/clients_handler.rs
+65
View File
@@ -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<ClientsHandlerRequest>;
pub(crate) type ClientsHandlerRequestReceiver = mpsc::UnboundedReceiver<ClientsHandlerRequest>;
pub(crate) type ClientsHandlerResponseSender = oneshot::Sender<ClientsHandlerResponse>;
pub(crate) type ClientsHandlerResponseReceiver = oneshot::Receiver<ClientsHandlerResponse>;
pub(crate) enum ClientsHandlerRequest {
// client
Register(
DestinationAddressBytes,
MixMessageSender,
ClientsHandlerResponseSender,
),
Authenticate(
temp_AuthToken,
MixMessageSender,
ClientsHandlerResponseSender,
),
// mix
IsOnline(DestinationAddressBytes, ClientsHandlerResponseSender),
}
pub(crate) enum ClientsHandlerResponse {
Register(Option<temp_AuthToken>),
Authenticate(bool),
IsOnline(Option<MixMessageSender>),
}
pub(crate) struct ClientsHandler {
open_connections: HashMap<DestinationAddressBytes, MixMessageSender>, // 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 })
}
}
gateway/src/client_handling/ledger.rs
+162
View File
@@ -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<Self, ClientLedgerError> {
// 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<bool, ClientLedgerError> {
// 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<Option<AuthToken>, 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<Option<AuthToken>, 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<Vec<MixProviderClient>, 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(),
// }
// }
//}
gateway/src/client_handling/mod.rs
+3
View File
@@ -0,0 +1,3 @@
pub(crate) mod clients_handler;
pub(crate) mod ledger;
pub(crate) mod websocket;
gateway/src/client_handling/websocket/connection_handler.rs
+74
View File
@@ -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<S: AsyncRead + AsyncWrite + Unpin> {
RawTCP(S),
UpgradedWebSocket(WebSocketStream<S>),
Invalid,
}
pub(crate) struct Handle<S: AsyncRead + AsyncWrite + Unpin> {
socket_connection: SocketStream<S>,
clients_handler_sender: ClientsHandlerRequestSender,
}
impl<S> Handle<S>
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<Result<Message, WsError>> {
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;
}
}
gateway/src/client_handling/websocket/listener.rs
+58
View File
@@ -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 })
}
}
gateway/src/client_handling/websocket/message_receiver.rs
+4
View File
@@ -0,0 +1,4 @@
use futures::channel::mpsc;
pub(crate) type MixMessageSender = mpsc::UnboundedSender<Vec<Vec<u8>>>;
pub(crate) type MixMessageReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
gateway/src/client_handling/websocket/mod.rs
+6
View File
@@ -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;
gateway/src/client_handling/websocket/types.rs
+42
View File
@@ -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
},
}
gateway/src/listener.rs
-42
View File
@@ -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<Mutex<MultiClient>>,
) -> 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(())
}
gateway/src/main.rs
+26 -25
View File
@@ -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<Mutex<MultiClient>>) {
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() {
gateway/src/mixnet_handling/mod.rs
+4
View File
@@ -0,0 +1,4 @@
pub(crate) mod receiver;
pub(crate) mod sender;
pub(crate) use receiver::listener::Listener;
gateway/src/mixnet_handling/receiver/connection_handler.rs
+88
View File
@@ -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<S: AsyncRead + AsyncWrite + Unpin> {
peer_address: SocketAddr,
socket_connection: S,
packet_processor: PacketProcessor,
}
impl<S> Handle<S>
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;
}
};
}
}
}
gateway/src/mixnet_handling/receiver/listener.rs
+51
View File
@@ -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 })
}
}
gateway/src/mixnet_handling/receiver/mod.rs
+17
View File
@@ -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;
gateway/src/mixnet_handling/receiver/packet_processing.rs
+217
View File
@@ -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<nymsphinx::ProcessingError> 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<io::Error> 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<encryption::PrivateKey>,
// 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<Mutex<HashMap<DestinationAddressBytes, MixMessageSender>>>,
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<MixMessageSender>,
message: Vec<u8>,
) -> 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<MixMessageSender> {
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<u8>,
) -> 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<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");
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(())
}
}
}
gateway/src/mixnet_handling/sender/mod.rs
+63
View File
@@ -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<u8>)>,
conn_rx: mpsc::UnboundedReceiver<(SocketAddr, Vec<u8>)>,
}
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<u8>)> {
// 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
}
}
gateway/src/storage/mod.rs
+222
View File
@@ -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<u8>,
path: PathBuf,
}
impl ClientFile {
fn new(content: Vec<u8>, path: PathBuf) -> Self {
ClientFile { content, path }
}
pub(crate) fn into_tuple(self) -> (Vec<u8>, PathBuf) {
(self.content, self.path)
}
}
pub struct StoreData {
client_address: DestinationAddressBytes,
message: Vec<u8>,
}
impl StoreData {
pub(crate) fn new(client_address: DestinationAddressBytes, message: Vec<u8>) -> 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<Mutex<ClientStorageInner>>,
}
// 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::<String>()
}
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<Vec<ClientFile>> {
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<PathBuf>) -> 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(())
}
}