From d0a07ac006996b01ca23d9dcc4cdaf133961cede Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 17 Dec 2019 10:53:06 +0000 Subject: [PATCH] stash pop --- Cargo.lock | 29 ++++++---- Cargo.toml | 4 +- src/clients/mod.rs | 109 +++++++++++++++++++++++++------------- src/commands/websocket.rs | 2 +- src/sockets/ws.rs | 78 +++++++++++++++++++++++---- 5 files changed, 162 insertions(+), 60 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2b061f193c..1db7bd75cb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -72,7 +72,7 @@ name = "assert-json-diff" version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -292,7 +292,7 @@ dependencies = [ "idna 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", "publicsuffix 1.5.4 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", "time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)", "try_from 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)", @@ -671,6 +671,11 @@ dependencies = [ "libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "hex" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "hkdf" version = "0.8.0" @@ -1030,12 +1035,13 @@ dependencies = [ "curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)", "dirs 2.0.2 (registry+https://github.com/rust-lang/crates.io-index)", "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "hex 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)", "mockito 0.22.0 (registry+https://github.com/rust-lang/crates.io-index)", "pem 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)", "rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)", "rand_distr 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)", "reqwest 0.9.22 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", "sfw-provider-requests 0.1.0", "sphinx 0.1.0", @@ -1393,7 +1399,7 @@ dependencies = [ "mime 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)", "mime_guess 2.0.1 (registry+https://github.com/rust-lang/crates.io-index)", "native-tls 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", "serde_urlencoded 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)", "time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1483,15 +1489,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" [[package]] name = "serde" -version = "1.0.103" +version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "serde_derive 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] name = "serde_derive" -version = "1.0.103" +version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1506,7 +1512,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)", "ryu 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -1516,7 +1522,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "dtoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)", "itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "url 1.7.2 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -2111,6 +2117,7 @@ dependencies = [ "checksum getrandom 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "e7db7ca94ed4cd01190ceee0d8a8052f08a247aa1b469a7f68c6a3b71afcf407" "checksum h2 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)" = "a5b34c246847f938a410a03c5458c7fee2274436675e76d8b903c08efc29c462" "checksum hermit-abi 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "307c3c9f937f38e3534b1d6447ecf090cafcc9744e4a6360e8b037b2cf5af120" +"checksum hex 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "023b39be39e3a2da62a94feb433e91e8bcd37676fbc8bea371daf52b7a769a3e" "checksum hkdf 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "3fa08a006102488bd9cd5b8013aabe84955cf5ae22e304c2caf655b633aefae3" "checksum hmac 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)" = "5dcb5e64cda4c23119ab41ba960d1e170a774c8e4b9d9e6a9bc18aabf5e59695" "checksum http 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)" = "d6ccf5ede3a895d8856620237b2f02972c1bbc78d2965ad7fe8838d4a0ed41f0" @@ -2198,8 +2205,8 @@ dependencies = [ "checksum security-framework-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "e31493fc37615debb8c5090a7aeb4a9730bc61e77ab10b9af59f1a202284f895" "checksum semver 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403" "checksum semver-parser 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3" -"checksum serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)" = "1217f97ab8e8904b57dd22eb61cde455fa7446a9c1cf43966066da047c1f3702" -"checksum serde_derive 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)" = "a8c6faef9a2e64b0064f48570289b4bf8823b7581f1d6157c1b52152306651d0" +"checksum serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)" = "414115f25f818d7dfccec8ee535d76949ae78584fc4f79a6f45a904bf8ab4449" +"checksum serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)" = "128f9e303a5a29922045a830221b8f78ec74a5f544944f3d5984f8ec3895ef64" "checksum serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)" = "48c575e0cc52bdd09b47f330f646cf59afc586e9c4e3ccd6fc1f625b8ea1dad7" "checksum serde_urlencoded 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)" = "642dd69105886af2efd227f75a520ec9b44a820d65bc133a9131f7d229fd165a" "checksum sha-1 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)" = "23962131a91661d643c98940b20fcaffe62d776a823247be80a48fcb8b6fce68" diff --git a/Cargo.toml b/Cargo.toml index 6f7837feec..2cef8f55f0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,11 +16,13 @@ clap = "2.33.0" curve25519-dalek = "1.2.3" dirs = "2.0.2" futures = "0.3.1" +hex = "0.4.0" pem = "0.7.0" rand = "0.7.2" rand_distr = "0.2.2" reqwest = "0.9.22" -serde = { version = "1.0", features = ["derive"] } +serde = { version = "1.0.104", features = ["derive"] } +serde_json = "1.0.44" sphinx = { path = "../sphinx" } sfw-provider-requests = { path = "../nym-sfw-provider/sfw-provider-requests" } tokio = { version = "0.2", features = ["full"] } diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 97b833f1b7..c9037bed97 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -1,21 +1,20 @@ -use sphinx::route::{DestinationAddressBytes, NodeAddressBytes, Destination}; -use tokio::runtime::Runtime; -use futures::channel::mpsc; -use std::time::Duration; -use crate::utils; -use futures::{StreamExt, SinkExt}; -use crate::clients::mix::MixClient; -use sphinx::SphinxPacket; -use futures::select; -use crate::utils::topology::get_topology; use crate::clients::directory::presence::Topology; +use crate::clients::mix::MixClient; +use crate::utils; +use crate::utils::topology::get_topology; +use futures::channel::mpsc; +use futures::select; +use futures::{SinkExt, StreamExt}; +use sphinx::route::{Destination, DestinationAddressBytes, NodeAddressBytes}; +use sphinx::SphinxPacket; +use std::time::Duration; +use tokio::runtime::Runtime; pub mod directory; pub mod mix; pub mod provider; pub mod validator; - // TODO: put that in config once it exists const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5; // assume seconds @@ -23,7 +22,6 @@ const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0; // assume seconds; const FETCH_MESSAGES_DELAY: f64 = 10.0; // assume seconds; - // provider-poller sends polls service provider; receives messages // provider-poller sends (TX) to ReceivedBufferController (RX) // ReceivedBufferController sends (TX) to ... ??Client?? @@ -36,13 +34,15 @@ struct MixMessage(NodeAddressBytes, SphinxPacket); struct MixTrafficController; - impl MixTrafficController { // this was way more difficult to implement than what this code may suggest... async fn run(mut rx: mpsc::UnboundedReceiver) { let mix_client = MixClient::new(); while let Some(mix_message) = rx.next().await { - println!("[MIX TRAFFIC CONTROL] - got a mix_message for {:?}", mix_message.0); + println!( + "[MIX TRAFFIC CONTROL] - got a mix_message for {:?}", + mix_message.0 + ); let send_res = mix_client.send(mix_message.1, mix_message.0).await; match send_res { Ok(_) => println!("We successfully sent the message!"), @@ -62,7 +62,6 @@ pub struct NymClient { is_local: bool, } - #[derive(Debug)] pub struct InputMessage(pub Destination, pub Vec); @@ -78,18 +77,29 @@ impl NymClient { } } - async fn start_loop_cover_traffic_stream(mut tx: mpsc::UnboundedSender, our_info: Destination, topology: &Topology) -> Result<(), Box> { + async fn start_loop_cover_traffic_stream( + mut tx: mpsc::UnboundedSender, + our_info: Destination, + topology: &Topology, + ) -> Result<(), Box> { loop { println!("[LOOP COVER TRAFFIC STREAM] - next cover message!"); let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY); let delay_duration = Duration::from_secs_f64(delay); tokio::time::delay_for(delay_duration).await; - let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, topology); - tx.send(MixMessage(cover_message.0, cover_message.1)).await?; + let cover_message = + utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, topology); + tx.send(MixMessage(cover_message.0, cover_message.1)) + .await?; } } - async fn control_out_queue(mut mix_tx: mpsc::UnboundedSender, mut input_rx: mpsc::UnboundedReceiver, our_info: Destination, topology: &Topology) -> Result<(), Box> { + async fn control_out_queue( + mut mix_tx: mpsc::UnboundedSender, + mut input_rx: mpsc::UnboundedReceiver, + our_info: Destination, + topology: &Topology, + ) -> Result<(), Box> { loop { println!("[OUT QUEUE] here I will be sending real traffic (or loop cover if nothing is available)"); select! { @@ -106,15 +116,13 @@ 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?; } - } - ; + }; let delay_duration = Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY); tokio::time::delay_for(delay_duration).await; } } - async fn start_provider_polling() -> Result<(), Box> { loop { println!("[FETCH MSG] - Polling provider..."); @@ -123,7 +131,6 @@ impl NymClient { } } - pub fn start(self) -> Result<(), Box> { println!("starting nym client"); @@ -132,31 +139,59 @@ impl NymClient { let mut rt = Runtime::new()?; rt.spawn(MixTrafficController::run(mix_rx)); -// let foomp = Destination::new(self.address, Default::default()); -// -// let mut input_channel = self.input_tx.clone(); -// rt.spawn(async move { -// let test_message = b"foomp".to_vec(); -// let recipient = foomp; -// let input_message = InputMessage(recipient, test_message); -// input_channel.send(input_message).await.unwrap(); -// }); + // let foomp = Destination::new(self.address, Default::default()); + // + // let mut input_channel = self.input_tx.clone(); + // rt.spawn(async move { + // let test_message = b"foomp".to_vec(); + // let recipient = foomp; + // let input_message = InputMessage(recipient, test_message); + // input_channel.send(input_message).await.unwrap(); + // }); 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; + } + }); rt.block_on(async { let future_results = futures::future::join3( - NymClient::start_loop_cover_traffic_stream(mix_tx.clone(), Destination::new(self.address, Default::default()), &topology.clone()), - NymClient::control_out_queue(mix_tx, self.input_rx, Destination::new(self.address, Default::default()), &topology.clone()), - NymClient::start_provider_polling()).await; + NymClient::start_loop_cover_traffic_stream( + mix_tx.clone(), + Destination::new(self.address, Default::default()), + &topology.clone(), + ), + NymClient::control_out_queue( + mix_tx, + self.input_rx, + Destination::new(self.address, Default::default()), + &topology.clone(), + ), + NymClient::start_provider_polling(), + ) + .await; // start websocket handler here - assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()); + assert!( + future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok() + ); }); // this line in theory should never be reached as the runtime should be permanently blocked on traffic senders eprintln!("The client went kaput..."); Ok(()) } -} \ No newline at end of file +} diff --git a/src/commands/websocket.rs b/src/commands/websocket.rs index 7d9889019c..c5ecfabd81 100644 --- a/src/commands/websocket.rs +++ b/src/commands/websocket.rs @@ -19,5 +19,5 @@ pub fn execute(matches: &ArgMatches) { .next() .expect("Failed to extract the socket address from the iterator"); - ws::start(socket_address); + // ws::start(socket_address); } diff --git a/src/sockets/ws.rs b/src/sockets/ws.rs index f4a08b4ceb..85f82d46e8 100644 --- a/src/sockets/ws.rs +++ b/src/sockets/ws.rs @@ -1,15 +1,39 @@ +use crate::clients::InputMessage; +use futures::channel::mpsc; +use hex::FromHexError; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sphinx::route::Destination; use std::net::SocketAddr; -use ws::{listen, CloseCode, Handler, Message, Result, Sender}; +use ws::{listen, CloseCode, Handler, Message, Sender}; + +#[derive(Debug)] +enum WebSocketError { + InvalidDestinationEncoding, + InvalidDestinationLength, +} + +impl From for WebSocketError { + fn from(_: FromHexError) -> Self { + use WebSocketError::*; + + InvalidDestinationEncoding + } +} struct Server { out: Sender, + input_tx: mpsc::UnboundedSender, } impl Handler for Server { - fn on_message(&mut self, msg: Message) -> Result<()> { - foomp(msg.clone()); + fn on_message(&mut self, msg: Message) -> ws::Result<()> { + let parsed_msg = Server::parse_message(msg.clone()); + + println!("msg: {:?}", parsed_msg); // Echo the message back - self.out.send(msg) + // self.out.send(msg) + Ok(()) } fn on_close(&mut self, code: CloseCode, reason: &str) { @@ -23,12 +47,46 @@ impl Handler for Server { } } -pub fn start(socket_address: SocketAddr) { - listen(socket_address, |out| Server { out: out }).unwrap() +impl Server { + // 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!"), + }; + + 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(), + )) + } } -// Proves we can call Rust methods from the websocket listener. Re-route it to wherever JS puts -// the `send_message` functionality. -fn foomp(msg: Message) { - println!("Foomp!: {:?}", msg); +pub fn start(socket_address: SocketAddr, input_tx: mpsc::UnboundedSender) { + // listen("127.0.0.1:3012", |out| { + // async move |msg| { + // out.send(msg) + // } + // }); + + // listen(socket_address, |out| Server { out, input_tx }).unwrap() +} + +#[derive(Serialize, Deserialize, Debug)] +struct ClientMessageJSON { + message: String, + recipient_address: String, }