From c02a44fbce5eccf88e2484cd0e069fc536272de2 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 17 Dec 2019 15:21:48 +0000 Subject: [PATCH] Ability to send mix messages from websocket --- src/clients/mod.rs | 31 ++----- src/commands/run.rs | 16 ++-- src/commands/websocket.rs | 12 ++- src/lib.rs | 1 + src/sockets/ws.rs | 179 ++++++++++++++++++++++++-------------- 5 files changed, 141 insertions(+), 98 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index eedf26b14c..cd0ccc099a 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -9,7 +9,9 @@ use sphinx::route::{Destination, DestinationAddressBytes, NodeAddressBytes}; use sphinx::SphinxPacket; use std::time::Duration; use tokio::runtime::Runtime; -use futures::future::join4; +use futures::future::join5; +use crate::sockets::ws; +use std::net::SocketAddr; pub mod directory; pub mod mix; @@ -17,7 +19,7 @@ pub mod provider; pub mod validator; // TODO: put that in config once it exists -const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5; +const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; // assume seconds const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0; // assume seconds; @@ -134,30 +136,14 @@ impl NymClient { } } - pub fn start(self) -> Result<(), Box> { + pub fn start(self, socket_address: SocketAddr) -> Result<(), Box> { println!("starting nym client"); let (mix_tx, mix_rx) = mpsc::unbounded(); let mut rt = Runtime::new()?; - let topology = get_topology(self.is_local); - -// let mut input_channel = self.input_tx.clone(); -// let bar = self.address.clone(); -// rt.spawn(async move { -// loop { -// let foomp = Destination::new(bar, Default::default()); -// -// let test_message = b"foomp".to_vec(); -// let recipient = foomp; -// let input_message = InputMessage(recipient, test_message); -// input_channel.send(input_message).await.unwrap(); -// tokio::time::delay_for(Duration::from_secs(1)).await; -// } -// }); - let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx)); let loop_cover_traffic_future = rt.spawn(NymClient::start_loop_cover_traffic_stream( mix_tx.clone(), @@ -173,13 +159,12 @@ impl NymClient { )); let provider_polling_future = rt.spawn(NymClient::start_provider_polling()); - - // TODO: websocket handler future + let websocket_future = rt.spawn(ws::start_websocket(socket_address, self.input_tx)); rt.block_on(async { - let future_results = join4(mix_traffic_future, loop_cover_traffic_future, out_queue_control_future, provider_polling_future).await; + let future_results = join5(mix_traffic_future, loop_cover_traffic_future, out_queue_control_future, provider_polling_future, websocket_future).await; assert!( - future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok() && future_results.3.is_ok() + future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok() && future_results.3.is_ok() && future_results.4.is_ok() ); }); diff --git a/src/commands/run.rs b/src/commands/run.rs index 81afdad7aa..6999237eaf 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -17,14 +17,16 @@ use tokio::runtime::Runtime; use tokio::time::{interval_at, Instant}; pub fn execute(matches: &ArgMatches) { - let is_local = matches.is_present("local"); - println!("Starting client, local: {:?}", is_local); + unimplemented!() // currently the 'run' starts websocket! - // todo: to be taken from config or something - let my_address = [42u8; 32]; - let is_local = true; - let client = NymClient::new(my_address, is_local); - client.start().unwrap(); + // let is_local = matches.is_present("local"); + // println!("Starting client, local: {:?}", is_local); + // + // // todo: to be taken from config or something + // let my_address = [42u8; 32]; + // let is_local = true; + // let client = NymClient::new(my_address, is_local); + // client.start().unwrap(); // Grab the network topology from the remote directory server // let topology = get_topology(is_local); diff --git a/src/commands/websocket.rs b/src/commands/websocket.rs index 7d9889019c..59a2e7492b 100644 --- a/src/commands/websocket.rs +++ b/src/commands/websocket.rs @@ -1,5 +1,5 @@ use crate::banner; -use crate::sockets::ws; +use crate::clients::NymClient; use clap::ArgMatches; use std::net::ToSocketAddrs; @@ -19,5 +19,13 @@ pub fn execute(matches: &ArgMatches) { .next() .expect("Failed to extract the socket address from the iterator"); - ws::start(socket_address); + let is_local = matches.is_present("local"); + println!("Starting client, local: {:?}", is_local); + + // todo: to be taken from config or something + let my_address = [42u8; 32]; + let is_local = true; + let client = NymClient::new(my_address, is_local); + + client.start(socket_address).unwrap(); } diff --git a/src/lib.rs b/src/lib.rs index 0574ba4a4e..5f1dc2a7c5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +1,5 @@ pub mod clients; pub mod identity; pub mod persistence; +pub mod sockets; pub mod utils; diff --git a/src/sockets/ws.rs b/src/sockets/ws.rs index 832f5d5da9..b4778a15ae 100644 --- a/src/sockets/ws.rs +++ b/src/sockets/ws.rs @@ -1,98 +1,148 @@ use crate::clients::InputMessage; use futures::channel::mpsc; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; -use futures::stream::Stream; -use futures::Future; use futures::{SinkExt, StreamExt}; -use hex::FromHexError; use serde::{Deserialize, Serialize}; -use serde_json::Value; use sphinx::route::Destination; use std::net::SocketAddr; -use tokio::runtime::Runtime; -use tokio_tungstenite::accept_async; use tungstenite::protocol::Message; -use std::time::Duration; +use std::io; +use futures::io::Error; struct Connection { address: SocketAddr, rx: UnboundedReceiver, tx: UnboundedSender, + + msg_input: mpsc::UnboundedSender, } #[derive(Debug)] -enum WebSocketError { - InvalidDestinationEncoding, - InvalidDestinationLength, +pub enum WebSocketError { + FailedToStartSocketError, + UnknownSocketError, } -impl From for WebSocketError { - fn from(_: FromHexError) -> Self { +impl From for WebSocketError { + fn from(err: Error) -> Self { use WebSocketError::*; + match err.kind() { + io::ErrorKind::ConnectionRefused => FailedToStartSocketError, + io::ErrorKind::ConnectionReset => FailedToStartSocketError, + io::ErrorKind::ConnectionAborted => FailedToStartSocketError, + io::ErrorKind::NotConnected => FailedToStartSocketError, - InvalidDestinationEncoding + io::ErrorKind::AddrInUse => FailedToStartSocketError, + io::ErrorKind::AddrNotAvailable => FailedToStartSocketError, + _ => UnknownSocketError, + } } } + #[derive(Serialize, Deserialize, Debug)] -struct ClientMessageJSON { - message: String, - recipient_address: String, +#[serde(tag = "type", rename_all = "camelCase")] +enum ClientRequest { + Send { message: String, recipient_address: String }, + Fetch, + GetClients, + OwnDetails, +} + +impl From for ClientRequest { + fn from(msg: Message) -> Self { + let text_msg = match msg { + Message::Text(msg) => msg, + Message::Binary(_) => panic!("binary messages are not supported!"), + Message::Close(_) => panic!("todo: handle close!"), + _ => panic!("Other types of messages are also unsupported!"), + }; + serde_json::from_str(&text_msg).unwrap() + } +} + +impl ClientRequest { + async fn handle_send(msg: String, recipient_address: String, mut input_tx: mpsc::UnboundedSender) -> ServerResponse { + let address_vec = match hex::decode(recipient_address) { + Err(e) => return ServerResponse::Error { message: e.to_string() }, + Ok(hex) => hex, + }; + + if address_vec.len() != 32 { + return ServerResponse::Error { message: "InvalidDestinationLength".to_string() }; + } + + let mut address = [0; 32]; + address.copy_from_slice(&address_vec); + + let dummy_surb = [0; 16]; + + let input_msg = InputMessage( + Destination::new(address, dummy_surb), + msg.into_bytes(), + ); + + input_tx.send(input_msg).await.unwrap(); + + ServerResponse::Send + } + + async fn handle_fetch() -> ServerResponse { + ServerResponse::Error { message: "NOT IMPLEMENTED".to_string() } + } + + async fn handle_get_clients() -> ServerResponse { + ServerResponse::Error { message: "NOT IMPLEMENTED".to_string() } + } + + async fn handle_own_details() -> ServerResponse { + ServerResponse::Error { message: "NOT IMPLEMENTED".to_string() } + } +} + +#[derive(Serialize, Deserialize, Debug)] +#[serde(tag = "type", rename_all = "camelCase")] +enum ServerResponse { + Send, + Fetch { messages: Vec> }, + GetClients { clients: Vec }, + OwnDetails { address: String }, + Error { message: String }, } fn dummy_response() -> Message { Message::Text("foomp".to_string()) } +impl Into for ServerResponse { + fn into(self) -> Message { + dummy_response() + } +} + + async fn handle_connection(conn: Connection) { let mut conn = conn; while let Some(msg) = conn.rx.next().await { println!("Received a message from {}: {}", conn.address, msg); - // TODO: currently only hardcoded sends are supported - let parsed_message = parse_message(msg); - println!("parsed: {:?}", parsed_message); - println!("test async pre wait"); + let request: ClientRequest = msg.into(); - tokio::time::delay_for(Duration::from_secs(2)).await; + let response = match request { + ClientRequest::Send { message, recipient_address } => ClientRequest::handle_send(message, recipient_address, conn.msg_input.clone()).await, + ClientRequest::Fetch => ClientRequest::handle_fetch().await, + ClientRequest::GetClients => ClientRequest::handle_get_clients().await, + ClientRequest::OwnDetails => ClientRequest::handle_own_details().await, + }; - println!("test async post wait"); conn .tx - .unbounded_send(dummy_response()) + .unbounded_send(response.into()) .expect("Failed to forward message"); } } -// Proves we can call Rust methods from the websocket listener. Re-route it to wherever JS puts -// the `send_message` functionality. -fn parse_message(msg: Message) -> Result { - let text_msg = match msg { - Message::Text(msg) => msg, - Message::Binary(_) => panic!("binary messages are not supported!"), - Message::Close(_) => panic!("todo: handle close!"), - _ => panic!("Other types of messages are also unsupported!"), - }; - - let raw_msg: ClientMessageJSON = serde_json::from_str(&text_msg).unwrap(); - let address_vec = hex::decode(raw_msg.recipient_address)?; - - if address_vec.len() != 32 { - return Err(WebSocketError::InvalidDestinationLength); - } - - let mut address = [0; 32]; - address.copy_from_slice(&address_vec); - - let dummy_surb = [0; 16]; - - Ok(InputMessage( - Destination::new(address, dummy_surb), - raw_msg.message.into_bytes(), - )) -} - -async fn accept_connection(stream: tokio::net::TcpStream) { +async fn accept_connection(stream: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender) { let address = stream .peer_addr() .expect("connected streams should have a peer address"); @@ -113,6 +163,8 @@ async fn accept_connection(stream: tokio::net::TcpStream) { address, rx: msg_rx, tx: response_tx, + + msg_input, }; tokio::spawn(handle_connection(conn)); @@ -127,21 +179,16 @@ async fn accept_connection(stream: tokio::net::TcpStream) { } } +pub async fn start_websocket(address: SocketAddr, message_tx: mpsc::UnboundedSender) -> Result<(), WebSocketError> { + 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())); + } -pub fn start(address: SocketAddr) -> Result<(), Box> { - let mut rt = Runtime::new()?; - - rt.block_on(async { - let mut listener = tokio::net::TcpListener::bind(address).await?; - - while let Ok((stream, _)) = listener.accept().await { - // TODO: should it rather be rt.spawn? - tokio::spawn(accept_connection(stream)); - } - - eprintln!("The websocket went kaput..."); - Ok(()) - }) + eprintln!("The websocket went kaput..."); + Ok(()) }