From 386a877f131364f833f92da3ffeefac12fa4dfcc Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 20 Dec 2019 16:34:58 +0000 Subject: [PATCH] TCP socket for nym client --- src/clients/mod.rs | 52 +++++-- src/commands/tcpsocket.rs | 12 +- src/commands/websocket.rs | 10 +- src/sockets/tcp.rs | 304 +++++++++++++++++++++++++++++++++++++- 4 files changed, 356 insertions(+), 22 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index b9595974fa..b521d86080 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -1,6 +1,7 @@ use crate::clients::directory::presence::Topology; use crate::clients::mix::MixClient; use crate::clients::provider::ProviderClient; +use crate::sockets::tcp; use crate::sockets::ws; use crate::utils; use crate::utils::topology::get_topology; @@ -23,9 +24,11 @@ pub mod mix; pub mod provider; pub mod validator; -const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; // seconds -const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 1.0; // seconds; -const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds; +const LOOP_COVER_AVERAGE_DELAY: f64 = 100.0; +// seconds +const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0; +// seconds; +const FETCH_MESSAGES_DELAY: f64 = 10.0; // seconds; // provider-poller sends polls service provider; receives messages // provider-poller sends (TX) to ReceivedBufferController (RX) @@ -107,6 +110,12 @@ impl ReceivedMessagesBuffer { } } +pub enum SocketType { + TCP, + WebSocket, + None, +} + pub struct NymClient { // to be replaced by something else I guess address: DestinationAddressBytes, @@ -116,6 +125,7 @@ pub struct NymClient { socket_listening_address: SocketAddr, is_local: bool, auth_token: Option, + socket_type: SocketType, } #[derive(Debug)] @@ -127,6 +137,7 @@ impl NymClient { socket_listening_address: SocketAddr, is_local: bool, auth_token: Option, + socket_type: SocketType, ) -> Self { let (input_tx, input_rx) = mpsc::unbounded::(); @@ -137,6 +148,7 @@ impl NymClient { socket_listening_address, is_local, auth_token, + socket_type, } } @@ -180,7 +192,8 @@ impl NymClient { let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, &topology); mix_tx.send(MixMessage(cover_message.0, cover_message.1)).await.unwrap(); } - }; + } + ; let delay_duration = Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY); tokio::time::delay_for(delay_duration).await; @@ -273,13 +286,28 @@ impl NymClient { provider_client, poller_input_tx, )); - let websocket_future = rt.spawn(ws::start_websocket( - self.socket_listening_address, - self.input_tx, - received_messages_buffer_output_tx, - self.address, - topology, - )); + + match self.socket_type { + SocketType::WebSocket => { + rt.spawn(ws::start_websocket( + self.socket_listening_address, + self.input_tx, + received_messages_buffer_output_tx, + self.address, + topology, + )); + } + SocketType::TCP => { + rt.spawn(tcp::start_tcpsocket( + self.socket_listening_address, + self.input_tx, + received_messages_buffer_output_tx, + self.address, + topology + )); + } + SocketType::None => (), + } rt.block_on(async { let future_results = join!( @@ -289,7 +317,6 @@ impl NymClient { loop_cover_traffic_future, out_queue_control_future, provider_polling_future, - websocket_future, ); assert!( @@ -299,7 +326,6 @@ impl NymClient { && future_results.3.is_ok() && future_results.4.is_ok() && future_results.5.is_ok() - && future_results.6.is_ok() ); }); diff --git a/src/commands/tcpsocket.rs b/src/commands/tcpsocket.rs index 4ba652bcd6..8c160a5ee0 100644 --- a/src/commands/tcpsocket.rs +++ b/src/commands/tcpsocket.rs @@ -1,5 +1,5 @@ use crate::banner; -use crate::clients::NymClient; +use crate::clients::{NymClient, SocketType}; use crate::persistence::pemstore; use crate::sockets::tcp; @@ -26,13 +26,15 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to extract the socket address from the iterator"); let keypair = pemstore::read_keypair_from_disk(id); - let _client = NymClient::new( + let auth_token = None; + + let client = NymClient::new( keypair.public_bytes(), socket_address.clone(), is_local, - None, + auth_token, + SocketType::TCP, ); - // Question: should we be passing the client into the TCP socket somehow next? - tcp::start(socket_address); + client.start().unwrap(); } diff --git a/src/commands/websocket.rs b/src/commands/websocket.rs index 36a88ea9ed..913417c31e 100644 --- a/src/commands/websocket.rs +++ b/src/commands/websocket.rs @@ -1,5 +1,5 @@ use crate::banner; -use crate::clients::NymClient; +use crate::clients::{NymClient, SocketType}; use crate::persistence::pemstore; use clap::ArgMatches; @@ -27,7 +27,13 @@ pub fn execute(matches: &ArgMatches) { let keypair = pemstore::read_keypair_from_disk(id); // TODO: reading auth_token from disk (if exists); let auth_token = None; - let client = NymClient::new(keypair.public_bytes(), socket_address, is_local, auth_token); + let client = NymClient::new( + keypair.public_bytes(), + socket_address, + is_local, + auth_token, + SocketType::WebSocket, + ); client.start().unwrap(); } diff --git a/src/sockets/tcp.rs b/src/sockets/tcp.rs index a9475ab1cb..5262475092 100644 --- a/src/sockets/tcp.rs +++ b/src/sockets/tcp.rs @@ -1,5 +1,305 @@ +use crate::clients::directory::presence::Topology; +use crate::clients::BufferResponse; +use crate::clients::InputMessage; +use futures::channel::{mpsc, oneshot}; +use futures::future::FutureExt; +use futures::io::Error; +use futures::SinkExt; +use sphinx::route::{Destination, DestinationAddressBytes}; +use std::io; use std::net::SocketAddr; +use tokio::prelude::*; +use std::convert::TryFrom; +use std::sync::Arc; +use std::borrow::Borrow; -pub fn start(_socket_address: SocketAddr) { - println!("TCP server time!!!"); +const SEND_REQUEST_PREFIX: u8 = 1; +const FETCH_REQUEST_PREFIX: u8 = 2; +const GET_CLIENTS_REQUEST_PREFIX: u8 = 3; +const OWN_DETAILS_REQUEST_PREFIX: u8 = 4; + +#[derive(Debug)] +pub enum TCPSocketError { + FailedToStartSocketError, + UnknownSocketError, + IncompleteDataError, + UnknownRequestError, } + +impl From for TCPSocketError { + fn from(err: Error) -> Self { + use TCPSocketError::*; + match err.kind() { + io::ErrorKind::ConnectionRefused => FailedToStartSocketError, + io::ErrorKind::ConnectionReset => FailedToStartSocketError, + io::ErrorKind::ConnectionAborted => FailedToStartSocketError, + io::ErrorKind::NotConnected => FailedToStartSocketError, + + io::ErrorKind::AddrInUse => FailedToStartSocketError, + io::ErrorKind::AddrNotAvailable => FailedToStartSocketError, + _ => UnknownSocketError, + } + } +} + + +enum ClientRequest { + Send { + message: Vec, + recipient_address: DestinationAddressBytes, + }, + Fetch, + GetClients, + OwnDetails, +} + +impl TryFrom<&[u8]> for ClientRequest { + type Error = TCPSocketError; + + fn try_from(data: &[u8]) -> Result { + use TCPSocketError::*; + if data.is_empty() { + return Err(IncompleteDataError); + } + + match data[0] { + SEND_REQUEST_PREFIX => parse_send_request(data), + FETCH_REQUEST_PREFIX => Ok(ClientRequest::Fetch), + GET_CLIENTS_REQUEST_PREFIX => Ok(ClientRequest::GetClients), + OWN_DETAILS_REQUEST_PREFIX => Ok(ClientRequest::OwnDetails), + _ => Err(UnknownRequestError) + } + } +} + +fn parse_send_request(data: &[u8]) -> Result { + if data.len() < 1 + 32 + 1 { + // make sure it has the prefix, destination and at least single byte of data + return Err(TCPSocketError::IncompleteDataError); + } + + let mut recipient_address = [0u8; 32]; + recipient_address.copy_from_slice(&data[1..33]); + + let message = data[33..].to_vec(); + + Ok(ClientRequest::Send { + message, + recipient_address, + }) +} + +impl ClientRequest { + async fn handle_send( + msg: Vec, + recipient_address: DestinationAddressBytes, + mut input_tx: mpsc::UnboundedSender, + ) -> ServerResponse { + println!("send handle. sending to: {:?}, msg: {:?}", recipient_address, msg); + let dummy_surb = [0; 16]; + let input_msg = InputMessage(Destination::new(recipient_address, dummy_surb), msg); + + println!("ALMOST ABOUT TO SOMEDAY SEND {:?}", input_msg); + input_tx.send(input_msg).await.unwrap(); + + ServerResponse::Send + } + + async fn handle_fetch(mut msg_query: mpsc::UnboundedSender) -> ServerResponse { + println!("fetch handle"); + let (res_tx, res_rx) = oneshot::channel(); + if msg_query.send(res_tx).await.is_err() { + return ServerResponse::Error { + message: "Server failed to receive messages".to_string(), + }; + } + + let messages = res_rx.map(|msg| msg).await; + + if messages.is_err() { + return ServerResponse::Error { + message: "Server failed to receive messages".to_string(), + }; + } + + let messages = messages.unwrap(); + println!("fetched {} messages", messages.len()); + ServerResponse::Fetch { + messages, + } + } + + async fn handle_get_clients(topology: &Topology) -> ServerResponse { + println!("get clients handle"); + let clients = topology + .mix_provider_nodes + .iter() + .flat_map(|provider| provider.registered_clients.iter()) + .map(|client| base64::decode_config(&client.pub_key, base64::URL_SAFE).unwrap()) // TODO: this can potentially throw an error + .collect(); + ServerResponse::GetClients { clients } + } + + async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse { + println!("own details handle"); + ServerResponse::OwnDetails { + address: self_address_bytes.to_vec(), + } + } +} + + +enum ServerResponse { + Send, + Fetch { messages: Vec> }, + GetClients { clients: Vec> }, + OwnDetails { address: Vec }, + Error { message: String }, +} + +impl Into> for ServerResponse { + fn into(self) -> Vec { + match self { + ServerResponse::Send => b"ok".to_vec(), + ServerResponse::Fetch {messages} =>encode_fetched_messages(messages), + ServerResponse::GetClients {clients} => encode_list_of_clients(clients), + ServerResponse::OwnDetails {address} => address, + ServerResponse::Error { message } => message.as_bytes().to_vec() + } + } +} + +// num_msgs || len1 || len2 || ... || msg1 || msg2 || ... +fn encode_fetched_messages(messages: Vec>) -> Vec { + // for reciprocal of this look into sfw-provider-requests::responses::PullResponse::from_bytes() + + let num_msgs = messages.len() as u16; + let msgs_lens: Vec = messages.iter().map(|msg| msg.len() as u16).collect(); + + num_msgs + .to_be_bytes() + .to_vec() + .into_iter() + .chain( + msgs_lens + .into_iter() + .flat_map(|len| len.to_be_bytes().to_vec().into_iter()), + ) + .chain(messages.iter().flat_map(|msg| msg.clone().into_iter())) + .collect() +} + +fn encode_list_of_clients(clients: Vec>) -> Vec { + println!("clients: {:?}", clients); + // we can just concat all clients since all of them got to be 32 bytes long + // (if not, then we have bigger problem somewhere up the line) + + // converts [[1,2,3],[4,5,6],...] into [1,2,3,4,5,6,...] + clients.into_iter().flatten().collect() +} + +impl ServerResponse { + fn new_error(message: String) -> ServerResponse { + ServerResponse::Error { message } + } +} + + +async fn handle_connection(data: &[u8], request_handling_data: RequestHandlingData) -> Result { + let request = ClientRequest::try_from(data)?; + let response = match request { + ClientRequest::Send { + message, + recipient_address + } => ClientRequest::handle_send(message, recipient_address, request_handling_data.msg_input).await, + ClientRequest::Fetch => ClientRequest::handle_fetch(request_handling_data.msg_query).await, + ClientRequest::GetClients => ClientRequest::handle_get_clients(request_handling_data.topology.borrow()).await, + ClientRequest::OwnDetails => ClientRequest::handle_own_details(request_handling_data.self_address).await, + }; + + Ok(response) +} + +struct RequestHandlingData { + msg_input: mpsc::UnboundedSender, + msg_query: mpsc::UnboundedSender, + self_address: DestinationAddressBytes, + topology: Arc, +} + +async fn accept_connection( + mut socket: tokio::net::TcpStream, + msg_input: mpsc::UnboundedSender, + msg_query: mpsc::UnboundedSender, + self_address: DestinationAddressBytes, + topology: Topology, +) { + let address = socket + .peer_addr() + .expect("connected streams should have a peer address"); + println!("Peer address: {}", address); + + let topology = Arc::new(topology); + + let mut buf = [0u8; 2048]; + + // In a loop, read data from the socket and write the data back. + loop { + // TODO: shutdowns? + + let response = match socket.read(&mut buf).await { + // socket closed + Ok(n) if n == 0 => { + println!("Remote connection closed."); + return; + } + Ok(n) => { + let request_handling_data = RequestHandlingData { + topology: topology.clone(), + msg_input: msg_input.clone(), + msg_query: msg_query.clone(), + self_address: self_address.clone(), + }; + match handle_connection(&buf[..n], request_handling_data).await { + Ok(res) => res, + Err(e) => ServerResponse::new_error(format!("{:?}", e)) + } + } + Err(e) => { + eprintln!("failed to read from socket; err = {:?}", e); + return; + } + }; + + let response_vec: Vec = response.into(); + if let Err(e) = socket.write_all(&response_vec).await { + eprintln!("failed to write reply to socket; err = {:?}", e); + return; + } + } +} + +pub async fn start_tcpsocket( + address: SocketAddr, + message_tx: mpsc::UnboundedSender, + received_messages_query_tx: mpsc::UnboundedSender, + self_address: DestinationAddressBytes, + topology: Topology, +) -> Result<(), TCPSocketError> { + let mut listener = tokio::net::TcpListener::bind(address).await?; + + while let Ok((stream, _)) = listener.accept().await { + // it's fine to be cloning the channel on all new connection, because in principle + // this server should only EVER have a single client connected + tokio::spawn(accept_connection( + stream, + message_tx.clone(), + received_messages_query_tx.clone(), + self_address, + topology.clone(), + )); + } + + eprintln!("The tcpsocket went kaput..."); + Ok(()) +} \ No newline at end of file