From 9114a10880e03f67e43442bddd8e8b559f5d44a6 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Dec 2019 17:23:03 +0000 Subject: [PATCH 01/11] Initial structure for starting all traffic streams --- Cargo.lock | 109 ++++++++++++++++++++++++++++++++++++++++++++ Cargo.toml | 2 +- src/clients/mod.rs | 49 ++++++++++++++++++++ src/commands/run.rs | 34 +++++++------- 4 files changed, 176 insertions(+), 18 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d4d5176172..233600748a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -537,6 +537,29 @@ name = "futures" version = "0.1.29" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "futures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-channel" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "futures-core" version = "0.3.1" @@ -551,6 +574,60 @@ dependencies = [ "num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "futures-executor" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-io" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-macro" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-sink" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-task" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-util" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "memchr 2.2.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)", + "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "generic-array" version = "0.12.3" @@ -936,6 +1013,7 @@ dependencies = [ "clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)", "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)", "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)", "reqwest 0.9.22 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1030,6 +1108,11 @@ name = "pin-project-lite" version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "pin-utils" +version = "0.1.0-alpha.4" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "pkg-config" version = "0.3.17" @@ -1040,6 +1123,21 @@ name = "ppv-lite86" version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "proc-macro-hack" +version = "0.5.11" +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)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "proc-macro-nested" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "proc-macro2" version = "1.0.6" @@ -1952,8 +2050,16 @@ dependencies = [ "checksum fuchsia-zircon 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2e9763c69ebaae630ba35f74888db465e49e259ba1bc0eda7d06f4a067615d82" "checksum fuchsia-zircon-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7" "checksum futures 0.1.29 (registry+https://github.com/rust-lang/crates.io-index)" = "1b980f2816d6ee8673b6517b52cb0e808a180efc92e5c19d02cdda79066703ef" +"checksum futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b6f16056ecbb57525ff698bb955162d0cd03bee84e6241c27ff75c08d8ca5987" +"checksum futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fcae98ca17d102fd8a3603727b9259fcf7fa4239b603d2142926189bc8999b86" "checksum futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "79564c427afefab1dfb3298535b21eda083ef7935b4f0ecbfcb121f0aec10866" "checksum futures-cpupool 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "ab90cde24b3319636588d0c35fe03b1333857621051837ed769faefb4c2162e4" +"checksum futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1e274736563f686a837a0568b478bdabfeaec2dca794b5649b04e2fe1627c231" +"checksum futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e676577d229e70952ab25f3945795ba5b16d63ca794ca9d2c860e5595d20b5ff" +"checksum futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "52e7c56c15537adb4f76d0b7a76ad131cb4d2f4f32d3b0bcabcbe1c7c5e87764" +"checksum futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "171be33efae63c2d59e6dbba34186fe0d6394fb378069a76dfd80fdcffd43c16" +"checksum futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0bae52d6b29cf440e298856fec3965ee6fa71b06aa7495178615953fd669e5f9" +"checksum futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c0d66274fb76985d3c62c886d1da7ac4c0903a8c9f754e8fe0f35a6a6cc39e76" "checksum generic-array 0.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "c68f0274ae0e023facc3c97b2e00f076be70e254bc851d972503b328db79b2ec" "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" @@ -2003,8 +2109,11 @@ dependencies = [ "checksum percent-encoding 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "31010dd2e1ac33d5b46a5b413495239882813e0369f8ed8a5e266f173602f831" "checksum percent-encoding 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e" "checksum pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f0af6cbca0e6e3ce8692ee19fb8d734b641899e07b68eb73e9bbbd32f1703991" +"checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587" "checksum pkg-config 0.3.17 (registry+https://github.com/rust-lang/crates.io-index)" = "05da548ad6865900e60eaba7f589cc0783590a92e940c26953ff81ddbab2d677" "checksum ppv-lite86 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "74490b50b9fbe561ac330df47c08f3f33073d2d00c150f719147d7c54522fa1b" +"checksum proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)" = "ecd45702f76d6d3c75a80564378ae228a85f0b59d2f3ed43c91b4a69eb2ebfc5" +"checksum proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "369a6ed065f249a159e06c45752c780bda2fb53c995718f9e484d08daa9eb42e" "checksum proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "9c9e470a8dc4aeae2dee2f335e8f533e2d4b347e1434e5671afc49b054592f27" "checksum publicsuffix 1.5.4 (registry+https://github.com/rust-lang/crates.io-index)" = "3bbaa49075179162b49acac1c6aa45fb4dafb5f13cf6794276d77bc7fd95757b" "checksum quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "053a8c8bcc71fcce321828dc897a98ab9760bef03a4fc36693c231e5b3216cfe" diff --git a/Cargo.toml b/Cargo.toml index b745efc446..f019dbcb7f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ reqwest = "0.9.22" serde = { version = "1.0", features = ["derive"] } sphinx = { path = "../sphinx" } sfw-provider-requests = { path = "../nym-sfw-provider/sfw-provider-requests" } - +futures = "0.3.1" tokio = { version = "0.2", features = ["full"] } [dev-dependencies] diff --git a/src/clients/mod.rs b/src/clients/mod.rs index e0ee8b42df..bad816a794 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -1,4 +1,53 @@ +use sphinx::route::DestinationAddressBytes; +use tokio::runtime::Runtime; + pub mod directory; pub mod mix; pub mod provider; pub mod validator; + +struct Client { + // to be replaced by something else I guess + address: DestinationAddressBytes +} + +type TripleFutureResult = (Result<(), Box>, Result<(), Box>, Result<(), Box>); + +impl Client { + pub fn new(address: DestinationAddressBytes) -> Self { + Client { + address + } + } + + async fn start_loop_cover_traffic_stream(&self) -> Result<(), Box> { + unimplemented!() + } + + async fn control_out_queue(&self) -> Result<(), Box> { + unimplemented!() + } + + + async fn start_provider_polling(&self) -> Result<(), Box> { + unimplemented!() + } + + + async fn start_traffic(&self) -> TripleFutureResult { + futures::future::join3(self.start_loop_cover_traffic_stream(), self.control_out_queue(), self.start_provider_polling()).await + } + + pub fn start(&self) -> Result<(), Box> { + let mut rt = Runtime::new()?; + + rt.block_on(async { + let future_results = self.start_traffic().await; + 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/run.rs b/src/commands/run.rs index aa6cd27ce4..48474bc1a6 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -34,23 +34,23 @@ pub fn execute(matches: &ArgMatches) { let mut i: usize = 0; loop { interval.tick().await; - let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); - - let route_len = 2; - - // data needed to generate a new Sphinx packet - let route = route_from(&topology, route_len); - let destination = get_destination(); - let delays = sphinx::header::delays::generate(route_len); - - // build the packet - let packet = - sphinx::SphinxPacket::new(message, &route[..], &destination, &delays).unwrap(); - - // send to mixnet - let mix_client = MixClient::new(); - let result = mix_client.send(packet, route.first().unwrap()).await; - println!("packet sent: {:?}", i); +// let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); +// +// let route_len = 2; +// +// // data needed to generate a new Sphinx packet +// let route = route_from(&topology, route_len); +// let destination = get_destination(); +// let delays = sphinx::header::delays::generate(route_len); +// +// // build the packet +// let packet = +// sphinx::SphinxPacket::new(message, &route[..], &destination, &delays).unwrap(); +// +// // send to mixnet +// let mix_client = MixClient::new(); +// let result = mix_client.send(packet, route.first().unwrap()).await; +// println!("packet sent: {:?}", i); i += 1; // retrieve messages every now and then From eee87a4d45541e6a232ee4bb8d3c35efb31375de Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Dec 2019 11:23:25 +0000 Subject: [PATCH 02/11] Sampling poisson distribution --- Cargo.toml | 4 +++- src/utils/mod.rs | 1 + src/utils/poisson.rs | 6 ++++++ 3 files changed, 10 insertions(+), 1 deletion(-) create mode 100644 src/utils/poisson.rs diff --git a/Cargo.toml b/Cargo.toml index f019dbcb7f..17043400e4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,12 +11,14 @@ base64 = "0.11.0" clap = "2.33.0" curve25519-dalek = "1.2.3" dirs = "2.0.2" +futures = "0.3.1" pem = "0.7.0" +rand = "0.7.2" +rand_distr = "0.2.2" reqwest = "0.9.22" serde = { version = "1.0", features = ["derive"] } sphinx = { path = "../sphinx" } sfw-provider-requests = { path = "../nym-sfw-provider/sfw-provider-requests" } -futures = "0.3.1" tokio = { version = "0.2", features = ["full"] } [dev-dependencies] diff --git a/src/utils/mod.rs b/src/utils/mod.rs index ad0049c4c2..2e504a1581 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -1 +1,2 @@ pub mod bytes; +pub mod poisson; diff --git a/src/utils/poisson.rs b/src/utils/poisson.rs new file mode 100644 index 0000000000..d4991f0dd8 --- /dev/null +++ b/src/utils/poisson.rs @@ -0,0 +1,6 @@ +use rand_distr::{Distribution, Exp}; + +pub fn sample(average_delay: f64) -> f64 { + let exp = Exp::new(1.0 / average_delay).unwrap(); + exp.sample(&mut rand::thread_rng()) +} From 59b931b919294cf2c64b0deff89d855fc3cd9e3e Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Dec 2019 11:36:02 +0000 Subject: [PATCH 03/11] Basic delay loop for cover traffic stream --- Cargo.lock | 2 ++ src/clients/mod.rs | 48 +++++++++++++++++++++---- src/commands/run.rs | 86 ++++++++++++++++++++++++--------------------- 3 files changed, 90 insertions(+), 46 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 233600748a..f33fb06525 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1016,6 +1016,8 @@ dependencies = [ "futures 0.3.1 (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_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", diff --git a/src/clients/mod.rs b/src/clients/mod.rs index bad816a794..076ec7d1ae 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -1,36 +1,68 @@ use sphinx::route::DestinationAddressBytes; use tokio::runtime::Runtime; +use futures::channel::mpsc; +use std::time::Duration; +use crate::utils; pub mod directory; pub mod mix; pub mod provider; pub mod validator; -struct Client { + +// TODO: put that in config once it exists +const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; // assume seconds +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?? +// outQueueController sends (TX) to TrafficStreamController (RX) +// TrafficStreamController sends messages to mixnet +// ... ??Client?? sends (TX) to outQueueController (RX) +// Loop cover traffic stream just sends messages to mixnet without any channel communication + +pub struct NymClient { // to be replaced by something else I guess address: DestinationAddressBytes } type TripleFutureResult = (Result<(), Box>, Result<(), Box>, Result<(), Box>); -impl Client { +impl NymClient { pub fn new(address: DestinationAddressBytes) -> Self { - Client { + NymClient { address } } async fn start_loop_cover_traffic_stream(&self) -> Result<(), Box> { - unimplemented!() + loop { + let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY); + let delay_duration = Duration::from_secs_f64(delay); + println!("waiting for {:?}", delay_duration); + tokio::time::delay_for(delay_duration).await; + println!("waited {:?} - time to send cover message!", delay_duration); + } + } async fn control_out_queue(&self) -> Result<(), Box> { - unimplemented!() + loop { + println!("here I will be sending real traffic (or loop cover if nothing is available)"); + let delay_duration = Duration::from_secs_f64(10.0); + tokio::time::delay_for(delay_duration).await; + } } async fn start_provider_polling(&self) -> Result<(), Box> { - unimplemented!() + loop { + println!("here I will be polling provider for messages"); + let delay_duration = Duration::from_secs_f64(10.0); + tokio::time::delay_for(delay_duration).await; + } } @@ -39,8 +71,12 @@ impl Client { } pub fn start(&self) -> Result<(), Box> { + println!("starting nym client"); let mut rt = Runtime::new()?; +// let (out_queue_tx, out_queue_rx) = mpsc::unbounded(); + + rt.block_on(async { let future_results = self.start_traffic().await; assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()); diff --git a/src/commands/run.rs b/src/commands/run.rs index 48474bc1a6..a73d83401b 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -4,6 +4,7 @@ use crate::clients::directory::requests::presence_topology_get::PresenceTopology use crate::clients::directory::DirectoryClient; use crate::clients::mix::MixClient; use crate::clients::provider::ProviderClient; +use crate::clients::NymClient; use crate::utils::bytes; use base64; use clap::ArgMatches; @@ -21,47 +22,52 @@ pub fn execute(matches: &ArgMatches) { custom_cfg ); - // Grab the network topology from the remote directory server - let topology = get_topology(); + // todo: to be taken from config or something + let my_address = [42u8; 32]; + let client = NymClient::new(my_address); + client.start().unwrap(); - // Create the runtime, probably later move it to Client struct itself? - let mut rt = Runtime::new().unwrap(); - - // Spawn the root task - rt.block_on(async { - let start = Instant::now() + Duration::from_nanos(1000); - let mut interval = interval_at(start, Duration::from_millis(1000)); - let mut i: usize = 0; - loop { - interval.tick().await; -// let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); -// -// let route_len = 2; -// -// // data needed to generate a new Sphinx packet -// let route = route_from(&topology, route_len); -// let destination = get_destination(); -// let delays = sphinx::header::delays::generate(route_len); -// -// // build the packet -// let packet = -// sphinx::SphinxPacket::new(message, &route[..], &destination, &delays).unwrap(); -// -// // send to mixnet -// let mix_client = MixClient::new(); -// let result = mix_client.send(packet, route.first().unwrap()).await; -// println!("packet sent: {:?}", i); - i += 1; - - // retrieve messages every now and then - if i % 3 == 0 { - interval.tick().await; - println!("going to retrieve messages!"); - let provider_client = ProviderClient::new(); - provider_client.retrieve_messages().await.unwrap(); - } - } - }) + // // Grab the network topology from the remote directory server + // let topology = get_topology(); + // + // // Create the runtime, probably later move it to Client struct itself? + // let mut rt = Runtime::new().unwrap(); + // + // // Spawn the root task + // rt.block_on(async { + // let start = Instant::now() + Duration::from_nanos(1000); + // let mut interval = interval_at(start, Duration::from_millis(1000)); + // let mut i: usize = 0; + // loop { + // interval.tick().await; + //// let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); + //// + //// let route_len = 2; + //// + //// // data needed to generate a new Sphinx packet + //// let route = route_from(&topology, route_len); + //// let destination = get_destination(); + //// let delays = sphinx::header::delays::generate(route_len); + //// + //// // build the packet + //// let packet = + //// sphinx::SphinxPacket::new(message, &route[..], &destination, &delays).unwrap(); + //// + //// // send to mixnet + //// let mix_client = MixClient::new(); + //// let result = mix_client.send(packet, route.first().unwrap()).await; + //// println!("packet sent: {:?}", i); + // i += 1; + // + // // retrieve messages every now and then + // if i % 3 == 0 { + // interval.tick().await; + // println!("going to retrieve messages!"); + // let provider_client = ProviderClient::new(); + // provider_client.retrieve_messages().await.unwrap(); + // } + // } + // }) } fn get_topology() -> Topology { From 19819729a14f365593dcff8256ed3c42643909b4 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Dec 2019 12:23:53 +0000 Subject: [PATCH 04/11] Sending messages via channel to mix traffic controller --- src/clients/mod.rs | 26 +++++++++++++++++++++----- 1 file changed, 21 insertions(+), 5 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 076ec7d1ae..f57d48ceb8 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -3,6 +3,7 @@ use tokio::runtime::Runtime; use futures::channel::mpsc; use std::time::Duration; use crate::utils; +use futures::{future, Future, Stream, Sink, StreamExt, SinkExt}; pub mod directory; pub mod mix; @@ -23,6 +24,18 @@ const FETCH_MESSAGES_DELAY: f64 = 10.0; // assume seconds; // ... ??Client?? sends (TX) to outQueueController (RX) // Loop cover traffic stream just sends messages to mixnet without any channel communication +struct MixTrafficController; + +impl MixTrafficController { + async fn run(rx: mpsc::Receiver>) { + rx.for_each(move |message| { + println!("here i will be sending {:?} to a mixnode!", message); + + future::ready(()) + }).await + } +} + pub struct NymClient { // to be replaced by something else I guess address: DestinationAddressBytes @@ -37,13 +50,15 @@ impl NymClient { } } - async fn start_loop_cover_traffic_stream(&self) -> Result<(), Box> { + async fn start_loop_cover_traffic_stream(&self, mut tx: mpsc::Sender>) -> Result<(), Box> { loop { let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY); let delay_duration = Duration::from_secs_f64(delay); println!("waiting for {:?}", delay_duration); tokio::time::delay_for(delay_duration).await; println!("waited {:?} - time to send cover message!", delay_duration); + let dummy_message = vec![1,2,3]; + tx.send(dummy_message).await; } } @@ -67,16 +82,17 @@ impl NymClient { async fn start_traffic(&self) -> TripleFutureResult { - futures::future::join3(self.start_loop_cover_traffic_stream(), self.control_out_queue(), self.start_provider_polling()).await + let mix_chan_buf_size = 64; + let (mix_tx, mix_rx) = mpsc::channel(mix_chan_buf_size); + + tokio::spawn(MixTrafficController::run(mix_rx)); + futures::future::join3(self.start_loop_cover_traffic_stream(mix_tx), self.control_out_queue(), self.start_provider_polling()).await } pub fn start(&self) -> Result<(), Box> { println!("starting nym client"); let mut rt = Runtime::new()?; -// let (out_queue_tx, out_queue_rx) = mpsc::unbounded(); - - rt.block_on(async { let future_results = self.start_traffic().await; assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()); From 27bf01eb28ba7f7336124f0195b30593eb3de984 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Dec 2019 12:42:18 +0000 Subject: [PATCH 05/11] Input channel created on new --- src/clients/mod.rs | 45 +++++++++++++++++++++++++++------------------ 1 file changed, 27 insertions(+), 18 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index f57d48ceb8..cb956ba2b6 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -12,8 +12,10 @@ pub mod validator; // TODO: put that in config once it exists -const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; // assume seconds -const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0; // assume seconds; +const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; +// assume seconds +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 @@ -38,32 +40,38 @@ impl MixTrafficController { pub struct NymClient { // to be replaced by something else I guess - address: DestinationAddressBytes + address: DestinationAddressBytes, + pub input_tx: mpsc::UnboundedSender>, + // to be used by "send" function or socket, etc + input_rx: mpsc::UnboundedReceiver>, } type TripleFutureResult = (Result<(), Box>, Result<(), Box>, Result<(), Box>); impl NymClient { pub fn new(address: DestinationAddressBytes) -> Self { + let (input_tx, input_rx) = mpsc::unbounded::>(); + NymClient { - address + address, + input_tx, + input_rx, } } - async fn start_loop_cover_traffic_stream(&self, mut tx: mpsc::Sender>) -> Result<(), Box> { + async fn start_loop_cover_traffic_stream(mut tx: mpsc::Sender>) -> Result<(), Box> { loop { let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY); let delay_duration = Duration::from_secs_f64(delay); println!("waiting for {:?}", delay_duration); tokio::time::delay_for(delay_duration).await; println!("waited {:?} - time to send cover message!", delay_duration); - let dummy_message = vec![1,2,3]; - tx.send(dummy_message).await; + let dummy_message = vec![1, 2, 3]; + tx.send(dummy_message).await?; } - } - async fn control_out_queue(&self) -> Result<(), Box> { + async fn control_out_queue(mut mix_tx: mpsc::Sender>, mut input_rx: mpsc::UnboundedReceiver>) -> Result<(), Box> { loop { println!("here I will be sending real traffic (or loop cover if nothing is available)"); let delay_duration = Duration::from_secs_f64(10.0); @@ -72,7 +80,7 @@ impl NymClient { } - async fn start_provider_polling(&self) -> Result<(), Box> { + async fn start_provider_polling() -> Result<(), Box> { loop { println!("here I will be polling provider for messages"); let delay_duration = Duration::from_secs_f64(10.0); @@ -81,20 +89,21 @@ impl NymClient { } - async fn start_traffic(&self) -> TripleFutureResult { + pub fn start(self) -> Result<(), Box> { + println!("starting nym client"); + let mix_chan_buf_size = 64; let (mix_tx, mix_rx) = mpsc::channel(mix_chan_buf_size); - tokio::spawn(MixTrafficController::run(mix_rx)); - futures::future::join3(self.start_loop_cover_traffic_stream(mix_tx), self.control_out_queue(), self.start_provider_polling()).await - } - - pub fn start(&self) -> Result<(), Box> { - println!("starting nym client"); let mut rt = Runtime::new()?; + rt.spawn(MixTrafficController::run(mix_rx)); + rt.block_on(async { - let future_results = self.start_traffic().await; + let future_results = futures::future::join3( + NymClient::start_loop_cover_traffic_stream(mix_tx.clone()), + NymClient::control_out_queue(mix_tx, self.input_rx), + NymClient::start_provider_polling()).await; assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()); }); From 7dcc81eab48e7a54ff1325ac7aaf561577023f39 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Dec 2019 13:12:03 +0000 Subject: [PATCH 06/11] Creating dummy sphinx loop message --- src/utils/mod.rs | 1 + src/utils/sphinx.rs | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+) create mode 100644 src/utils/sphinx.rs diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 2e504a1581..d213166bdc 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -1,2 +1,3 @@ pub mod bytes; pub mod poisson; +pub mod sphinx; diff --git a/src/utils/sphinx.rs b/src/utils/sphinx.rs new file mode 100644 index 0000000000..6b0fc2e505 --- /dev/null +++ b/src/utils/sphinx.rs @@ -0,0 +1,33 @@ +use crate::utils::bytes; +use sphinx::route::{Destination, DestinationAddressBytes, Node, SURBIdentifier}; + +const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!"; + +pub fn loop_cover_message( + our_address: DestinationAddressBytes, + surb_id: SURBIdentifier, +) -> Vec { + let dummy_route = vec![ + Node::new( + bytes::zero_pad_to_32("127.0.0.1:8080".as_bytes().to_vec()), + Default::default(), + ), + Node::new( + bytes::zero_pad_to_32("127.0.0.1:8081".as_bytes().to_vec()), + Default::default(), + ), + ]; + + let destination = Destination::new(our_address, surb_id); + let delays = sphinx::header::delays::generate(dummy_route.len()); + + // build the packet + let packet = sphinx::SphinxPacket::new( + LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), + &dummy_route[..], + &destination, + &delays, + ) + .unwrap(); + packet.to_bytes() +} From dfece9bda5b485b48ed6aa0ee9736251096b69c8 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Dec 2019 16:03:19 +0000 Subject: [PATCH 07/11] Working loop cover traffic --- src/clients/mix.rs | 5 +++-- src/clients/mod.rs | 46 ++++++++++++++++++++++++++++++--------------- src/utils/sphinx.rs | 14 +++++++------- 3 files changed, 41 insertions(+), 24 deletions(-) diff --git a/src/clients/mix.rs b/src/clients/mix.rs index b5f03a130f..43e7f1199d 100644 --- a/src/clients/mix.rs +++ b/src/clients/mix.rs @@ -2,6 +2,7 @@ use crate::clients::directory::Client; use sphinx::route::Node as MixNode; use sphinx::SphinxPacket; use tokio::prelude::*; +use std::net::SocketAddr; pub struct MixClient {} @@ -14,11 +15,11 @@ impl MixClient { pub async fn send( &self, packet: SphinxPacket, - mix: &MixNode, + mix_addr: SocketAddr, ) -> Result<(), Box> { let bytes = packet.to_bytes(); - let mut stream = tokio::net::TcpStream::connect("127.0.0.1:8080").await?; + let mut stream = tokio::net::TcpStream::connect(mix_addr).await?; stream.write_all(&bytes[..]).await?; Ok(()) } diff --git a/src/clients/mod.rs b/src/clients/mod.rs index cb956ba2b6..0bc5d39f8d 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -1,9 +1,11 @@ -use sphinx::route::DestinationAddressBytes; +use sphinx::route::{DestinationAddressBytes, NodeAddressBytes, Destination}; use tokio::runtime::Runtime; use futures::channel::mpsc; use std::time::Duration; use crate::utils; -use futures::{future, Future, Stream, Sink, StreamExt, SinkExt}; +use futures::{future, StreamExt, SinkExt}; +use crate::clients::mix::MixClient; +use sphinx::SphinxPacket; pub mod directory; pub mod mix; @@ -18,6 +20,7 @@ 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?? @@ -26,15 +29,28 @@ const FETCH_MESSAGES_DELAY: f64 = 10.0; // assume seconds; // ... ??Client?? sends (TX) to outQueueController (RX) // Loop cover traffic stream just sends messages to mixnet without any channel communication +struct MixMessage(NodeAddressBytes, SphinxPacket); + struct MixTrafficController; -impl MixTrafficController { - async fn run(rx: mpsc::Receiver>) { - rx.for_each(move |message| { - println!("here i will be sending {:?} to a mixnode!", message); - future::ready(()) - }).await + +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!("got a mix_message for {:?}", mix_message.0); + + println!("here i will be sending sphinx packet to a mixnode ({:?}!", mix_message.0); + // here NodeAddressBytes would be transformed into a SocketAddr with SOME library call... + let node_net_address = "127.0.0.1:8080"; + let send_res = mix_client.send(mix_message.1, node_net_address.parse().unwrap()).await; + match send_res { + Ok(_) => println!("We successfully sent the message!"), + Err(e) => eprintln!("We failed to send the message :( - {:?}", e), + }; + } } } @@ -59,19 +75,19 @@ impl NymClient { } } - async fn start_loop_cover_traffic_stream(mut tx: mpsc::Sender>) -> Result<(), Box> { + async fn start_loop_cover_traffic_stream(mut tx: mpsc::UnboundedSender, our_info: Destination) -> Result<(), Box> { loop { let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY); let delay_duration = Duration::from_secs_f64(delay); println!("waiting for {:?}", delay_duration); tokio::time::delay_for(delay_duration).await; + let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier); println!("waited {:?} - time to send cover message!", delay_duration); - let dummy_message = vec![1, 2, 3]; - tx.send(dummy_message).await?; + tx.send(MixMessage(cover_message.0, cover_message.1)).await?; } } - async fn control_out_queue(mut mix_tx: mpsc::Sender>, mut input_rx: mpsc::UnboundedReceiver>) -> Result<(), Box> { + async fn control_out_queue(mut mix_tx: mpsc::UnboundedSender, mut input_rx: mpsc::UnboundedReceiver>) -> Result<(), Box> { loop { println!("here I will be sending real traffic (or loop cover if nothing is available)"); let delay_duration = Duration::from_secs_f64(10.0); @@ -92,8 +108,8 @@ impl NymClient { pub fn start(self) -> Result<(), Box> { println!("starting nym client"); - let mix_chan_buf_size = 64; - let (mix_tx, mix_rx) = mpsc::channel(mix_chan_buf_size); + let (mix_tx, mix_rx) = mpsc::unbounded(); + let our_info = Destination::new(self.address, Default::default()); let mut rt = Runtime::new()?; @@ -101,7 +117,7 @@ impl NymClient { rt.block_on(async { let future_results = futures::future::join3( - NymClient::start_loop_cover_traffic_stream(mix_tx.clone()), + NymClient::start_loop_cover_traffic_stream(mix_tx.clone(), our_info), NymClient::control_out_queue(mix_tx, self.input_rx), NymClient::start_provider_polling()).await; assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()); diff --git a/src/utils/sphinx.rs b/src/utils/sphinx.rs index 6b0fc2e505..eafd399fa2 100644 --- a/src/utils/sphinx.rs +++ b/src/utils/sphinx.rs @@ -1,17 +1,16 @@ use crate::utils::bytes; -use sphinx::route::{Destination, DestinationAddressBytes, Node, SURBIdentifier}; +use sphinx::route::{Destination, DestinationAddressBytes, Node, NodeAddressBytes, SURBIdentifier}; +use sphinx::SphinxPacket; const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!"; pub fn loop_cover_message( our_address: DestinationAddressBytes, surb_id: SURBIdentifier, -) -> Vec { +) -> (NodeAddressBytes, SphinxPacket) { + let first_node_address = bytes::zero_pad_to_32("127.0.0.1:8080".as_bytes().to_vec()); let dummy_route = vec![ - Node::new( - bytes::zero_pad_to_32("127.0.0.1:8080".as_bytes().to_vec()), - Default::default(), - ), + Node::new(first_node_address, Default::default()), Node::new( bytes::zero_pad_to_32("127.0.0.1:8081".as_bytes().to_vec()), Default::default(), @@ -29,5 +28,6 @@ pub fn loop_cover_message( &delays, ) .unwrap(); - packet.to_bytes() + + (first_node_address, packet) } From a472c23b48819271f50d568e01a7dea50259a054 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Mon, 16 Dec 2019 10:59:11 +0000 Subject: [PATCH 08/11] Managed to combine cover traffic stream with out queue --- src/clients/mod.rs | 93 ++++++++++++++++++++++++++++++++++----------- src/utils/sphinx.rs | 19 +++++---- 2 files changed, 81 insertions(+), 31 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index 0bc5d39f8d..fe93c65260 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -3,9 +3,14 @@ use tokio::runtime::Runtime; use futures::channel::mpsc; use std::time::Duration; use crate::utils; -use futures::{future, StreamExt, SinkExt}; +use futures::{StreamExt, SinkExt, Stream, TryStreamExt}; use crate::clients::mix::MixClient; use sphinx::SphinxPacket; +use futures::Future; +use futures::future; +use futures::task::{Context, Poll}; +use std::pin::Pin; +use futures::select; pub mod directory; pub mod mix; @@ -34,22 +39,19 @@ 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!("got a mix_message for {:?}", mix_message.0); - - println!("here i will be sending sphinx packet to a mixnode ({:?}!", mix_message.0); + println!("[MIX TRAFFIC CONTROL] - got a mix_message for {:?}", mix_message.0); // here NodeAddressBytes would be transformed into a SocketAddr with SOME library call... let node_net_address = "127.0.0.1:8080"; - let send_res = mix_client.send(mix_message.1, node_net_address.parse().unwrap()).await; - match send_res { - Ok(_) => println!("We successfully sent the message!"), - Err(e) => eprintln!("We failed to send the message :( - {:?}", e), - }; +// let send_res = mix_client.send(mix_message.1, node_net_address.parse().unwrap()).await; +// match send_res { +// Ok(_) => println!("We successfully sent the message!"), +// Err(e) => eprintln!("We failed to send the message :( - {:?}", e), +// }; } } } @@ -57,16 +59,38 @@ impl MixTrafficController { pub struct NymClient { // to be replaced by something else I guess address: DestinationAddressBytes, - pub input_tx: mpsc::UnboundedSender>, + pub input_tx: mpsc::UnboundedSender, // to be used by "send" function or socket, etc - input_rx: mpsc::UnboundedReceiver>, + input_rx: mpsc::UnboundedReceiver, } type TripleFutureResult = (Result<(), Box>, Result<(), Box>, Result<(), Box>); +//struct OutQueueFuture { +// +//} +// +//impl OutQueueFuture { +// fn poll_data(&mut self, cx: &mut Context) -> Poll<()> { +// use Poll::*; +// +// } +//} +// +//impl Future for OutQueueFuture { +// type Output = (); +// +// fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { +// unimplemented!() +// } +//} + +#[derive(Debug)] +pub struct InputMessage(Destination, Vec); + impl NymClient { pub fn new(address: DestinationAddressBytes) -> Self { - let (input_tx, input_rx) = mpsc::unbounded::>(); + let (input_tx, input_rx) = mpsc::unbounded::(); NymClient { address, @@ -77,20 +101,35 @@ impl NymClient { async fn start_loop_cover_traffic_stream(mut tx: mpsc::UnboundedSender, our_info: Destination) -> 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); - println!("waiting for {:?}", delay_duration); tokio::time::delay_for(delay_duration).await; let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier); - println!("waited {:?} - time to send cover message!", delay_duration); 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>) -> Result<(), Box> { + async fn control_out_queue(mut mix_tx: mpsc::UnboundedSender, mut input_rx: mpsc::UnboundedReceiver, our_info: Destination) -> Result<(), Box> { loop { - println!("here I will be sending real traffic (or loop cover if nothing is available)"); - let delay_duration = Duration::from_secs_f64(10.0); + println!("[OUT QUEUE] here I will be sending real traffic (or loop cover if nothing is available)"); + select! { + real_message = input_rx.next() => { + println!("[OUT QUEUE] - we got a real message!"); + let real_message = real_message.expect("The channel must have closed! - if the client hasn't crashed, it should have!"); + println!("real: {:?}", real_message); + let encapsulated_message = utils::sphinx::encapsulate_message(real_message.0, real_message.1); + mix_tx.send(MixMessage(encapsulated_message.0, encapsulated_message.1)).await?; + }, + + default => { + println!("[OUT QUEUE] - no real message - going to send extra loop cover"); + let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier); + 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; } } @@ -98,7 +137,7 @@ impl NymClient { async fn start_provider_polling() -> Result<(), Box> { loop { - println!("here I will be polling provider for messages"); +// println!("here I will be polling provider for messages"); let delay_duration = Duration::from_secs_f64(10.0); tokio::time::delay_for(delay_duration).await; } @@ -109,16 +148,24 @@ impl NymClient { println!("starting nym client"); let (mix_tx, mix_rx) = mpsc::unbounded(); - let our_info = Destination::new(self.address, Default::default()); let mut rt = Runtime::new()?; - rt.spawn(MixTrafficController::run(mix_rx)); + let tmp_dest = Destination::new(self.address, Default::default()); + let mut foo_tx = self.input_tx.clone(); +// just to send something after a delay to the input channel + rt.spawn(async move { + tokio::time::delay_for(Duration::from_secs(8)).await; + let tmp_message = vec![5, 23, 62, 6, 236]; + println!("SENDING TMP MESSAGE!"); + foo_tx.send(InputMessage(tmp_dest, tmp_message)).await.unwrap(); + }); + rt.block_on(async { let future_results = futures::future::join3( - NymClient::start_loop_cover_traffic_stream(mix_tx.clone(), our_info), - NymClient::control_out_queue(mix_tx, self.input_rx), + NymClient::start_loop_cover_traffic_stream(mix_tx.clone(), Destination::new(self.address, Default::default())), + NymClient::control_out_queue(mix_tx, self.input_rx, Destination::new(self.address, Default::default())), NymClient::start_provider_polling()).await; assert!(future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()); }); diff --git a/src/utils/sphinx.rs b/src/utils/sphinx.rs index eafd399fa2..48165259e0 100644 --- a/src/utils/sphinx.rs +++ b/src/utils/sphinx.rs @@ -8,6 +8,16 @@ pub fn loop_cover_message( our_address: DestinationAddressBytes, surb_id: SURBIdentifier, ) -> (NodeAddressBytes, SphinxPacket) { + let destination = Destination::new(our_address, surb_id); + + encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec()) +} + +pub fn encapsulate_message( + recipient: Destination, + message: Vec, +) -> (NodeAddressBytes, SphinxPacket) { + // here we would be getting topology, etc let first_node_address = bytes::zero_pad_to_32("127.0.0.1:8080".as_bytes().to_vec()); let dummy_route = vec![ Node::new(first_node_address, Default::default()), @@ -17,17 +27,10 @@ pub fn loop_cover_message( ), ]; - let destination = Destination::new(our_address, surb_id); let delays = sphinx::header::delays::generate(dummy_route.len()); // build the packet - let packet = sphinx::SphinxPacket::new( - LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), - &dummy_route[..], - &destination, - &delays, - ) - .unwrap(); + let packet = sphinx::SphinxPacket::new(message, &dummy_route[..], &recipient, &delays).unwrap(); (first_node_address, packet) } From f5f2ecb65e0c323a2d9f7259604a56ffabe44571 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Mon, 16 Dec 2019 11:43:15 +0000 Subject: [PATCH 09/11] Some cleanup + sample message sending --- src/clients/mod.rs | 62 ++++++++++++++++------------------------------ 1 file changed, 21 insertions(+), 41 deletions(-) diff --git a/src/clients/mod.rs b/src/clients/mod.rs index fe93c65260..c512a8e11c 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -3,13 +3,9 @@ use tokio::runtime::Runtime; use futures::channel::mpsc; use std::time::Duration; use crate::utils; -use futures::{StreamExt, SinkExt, Stream, TryStreamExt}; +use futures::{StreamExt, SinkExt}; use crate::clients::mix::MixClient; use sphinx::SphinxPacket; -use futures::Future; -use futures::future; -use futures::task::{Context, Poll}; -use std::pin::Pin; use futures::select; pub mod directory; @@ -47,11 +43,11 @@ impl MixTrafficController { println!("[MIX TRAFFIC CONTROL] - got a mix_message for {:?}", mix_message.0); // here NodeAddressBytes would be transformed into a SocketAddr with SOME library call... let node_net_address = "127.0.0.1:8080"; -// let send_res = mix_client.send(mix_message.1, node_net_address.parse().unwrap()).await; -// match send_res { -// Ok(_) => println!("We successfully sent the message!"), -// Err(e) => eprintln!("We failed to send the message :( - {:?}", e), -// }; + let send_res = mix_client.send(mix_message.1, node_net_address.parse().unwrap()).await; + match send_res { + Ok(_) => println!("We successfully sent the message!"), + Err(e) => eprintln!("We failed to send the message :( - {:?}", e), + }; } } } @@ -64,29 +60,9 @@ pub struct NymClient { input_rx: mpsc::UnboundedReceiver, } -type TripleFutureResult = (Result<(), Box>, Result<(), Box>, Result<(), Box>); - -//struct OutQueueFuture { -// -//} -// -//impl OutQueueFuture { -// fn poll_data(&mut self, cx: &mut Context) -> Poll<()> { -// use Poll::*; -// -// } -//} -// -//impl Future for OutQueueFuture { -// type Output = (); -// -// fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { -// unimplemented!() -// } -//} #[derive(Debug)] -pub struct InputMessage(Destination, Vec); +pub struct InputMessage(pub Destination, pub Vec); impl NymClient { pub fn new(address: DestinationAddressBytes) -> Self { @@ -127,7 +103,8 @@ impl NymClient { let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier); 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; @@ -137,8 +114,8 @@ impl NymClient { async fn start_provider_polling() -> Result<(), Box> { loop { -// println!("here I will be polling provider for messages"); - let delay_duration = Duration::from_secs_f64(10.0); + println!("[FETCH MSG] - Polling provider..."); + let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY); tokio::time::delay_for(delay_duration).await; } } @@ -152,14 +129,14 @@ impl NymClient { let mut rt = Runtime::new()?; rt.spawn(MixTrafficController::run(mix_rx)); - let tmp_dest = Destination::new(self.address, Default::default()); - let mut foo_tx = self.input_tx.clone(); -// just to send something after a delay to the input channel + let foomp = Destination::new(self.address, Default::default()); + + let mut input_channel = self.input_tx.clone(); rt.spawn(async move { - tokio::time::delay_for(Duration::from_secs(8)).await; - let tmp_message = vec![5, 23, 62, 6, 236]; - println!("SENDING TMP MESSAGE!"); - foo_tx.send(InputMessage(tmp_dest, tmp_message)).await.unwrap(); + let test_message = b"foomp".to_vec(); + let recipient = foomp; + let input_message = InputMessage(recipient, test_message); + input_channel.send(input_message).await.unwrap(); }); rt.block_on(async { @@ -167,6 +144,9 @@ impl NymClient { NymClient::start_loop_cover_traffic_stream(mix_tx.clone(), Destination::new(self.address, Default::default())), NymClient::control_out_queue(mix_tx, self.input_rx, Destination::new(self.address, Default::default())), 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()); }); From 29106cef6f8b2a6928fe1bea355ae8aa3b728bc1 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Mon, 16 Dec 2019 17:47:11 +0000 Subject: [PATCH 10/11] This should have been 10 separate commits --- src/clients/directory/mod.rs | 2 +- src/clients/mix.rs | 1 + src/clients/mod.rs | 6 ++-- src/commands/run.rs | 47 +-------------------------- src/lib.rs | 1 + src/utils/mod.rs | 1 + src/utils/sphinx.rs | 27 +++++++++------- src/utils/topology.rs | 61 ++++++++++++++++++++++++++++++++++++ 8 files changed, 84 insertions(+), 62 deletions(-) create mode 100644 src/utils/topology.rs diff --git a/src/clients/directory/mod.rs b/src/clients/directory/mod.rs index bc670ac822..a9ff9e4df8 100644 --- a/src/clients/directory/mod.rs +++ b/src/clients/directory/mod.rs @@ -20,7 +20,7 @@ use crate::clients::directory::requests::presence_topology_get::{ PresenceTopologyGetRequester, Request as PresenceTopologyRequest, }; -mod metrics; +pub mod metrics; pub mod presence; pub mod requests; diff --git a/src/clients/mix.rs b/src/clients/mix.rs index f8e07c2787..65480bf539 100644 --- a/src/clients/mix.rs +++ b/src/clients/mix.rs @@ -23,6 +23,7 @@ impl MixClient { let host = Ipv4Addr::new(b[0], b[1], b[2], b[3]); let port: u16 = u16::from_be_bytes([b[4], b[5]]); let socket_address = SocketAddrV4::new(host, port); + println!("socket addr: {:?}", socket_address); let mut stream = tokio::net::TcpStream::connect(socket_address).await?; stream.write_all(&bytes[..]).await?; diff --git a/src/clients/mod.rs b/src/clients/mod.rs index c512a8e11c..b07ca9a7d7 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -15,7 +15,7 @@ pub mod validator; // TODO: put that in config once it exists -const LOOP_COVER_AVERAGE_DELAY: f64 = 10.0; +const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5; // assume seconds const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0; // assume seconds; @@ -41,9 +41,7 @@ impl MixTrafficController { 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); - // here NodeAddressBytes would be transformed into a SocketAddr with SOME library call... - let node_net_address = "127.0.0.1:8080"; - let send_res = mix_client.send(mix_message.1, node_net_address.parse().unwrap()).await; + let send_res = mix_client.send(mix_message.1, mix_message.0).await; match send_res { Ok(_) => println!("We successfully sent the message!"), Err(e) => eprintln!("We failed to send the message :( - {:?}", e), diff --git a/src/commands/run.rs b/src/commands/run.rs index d9f3d6c973..d0b7ea3596 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -25,7 +25,7 @@ pub fn execute(matches: &ArgMatches) { let client = NymClient::new(my_address); client.start().unwrap(); // Grab the network topology from the remote directory server - let topology = get_topology(is_local); + // let topology = get_topology(is_local); // // Grab the network topology from the remote directory server // let topology = get_topology(); @@ -111,51 +111,6 @@ pub fn execute(matches: &ArgMatches) { // }) } -fn get_topology(is_local: bool) -> Topology { - let url = if is_local { - "http://localhost:8080".to_string() - } else { - "https://directory.nymtech.net".to_string() - }; - println!("Using directory server: {:?}", url); - let directory_config = directory::Config { base_url: url }; - let directory = directory::Client::new(directory_config); - - let topology = directory - .presence_topology - .get() - .expect("Failed to retrieve network topology."); - topology -} - -fn route_from(topology: &Topology, route_len: usize) -> Vec { - let mut route = vec![]; - let nodes = topology.mix_nodes.iter(); - for mix in nodes.take(route_len) { - let address_bytes = socket_bytes_from_string(mix.host.clone()); - let decoded_key_bytes = base64::decode_config(&mix.pub_key, base64::URL_SAFE).unwrap(); - let key_bytes = bytes::zero_pad_to_32(decoded_key_bytes); - let key = MontgomeryPoint(key_bytes); - let sphinx_node = SphinxNode { - address: address_bytes, - pub_key: key, - }; - route.push(sphinx_node); - } - route -} - -fn socket_bytes_from_string(address: String) -> [u8; 32] { - let socket: SocketAddrV4 = address.parse().unwrap(); - let host_bytes = socket.ip().octets(); - let port_bytes = socket.port().to_be_bytes(); - let mut address_bytes = [0u8; 32]; - address_bytes.copy_from_slice(&host_bytes); - address_bytes[4] = port_bytes[0]; - address_bytes[5] = port_bytes[1]; - address_bytes -} - // TODO: where do we retrieve this guy from? fn get_destination() -> Destination { Destination { diff --git a/src/lib.rs b/src/lib.rs index ab56f42ee8..0574ba4a4e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,4 @@ pub mod clients; pub mod identity; pub mod persistence; +pub mod utils; diff --git a/src/utils/mod.rs b/src/utils/mod.rs index d213166bdc..ee03437270 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -1,3 +1,4 @@ pub mod bytes; pub mod poisson; pub mod sphinx; +pub mod topology; diff --git a/src/utils/sphinx.rs b/src/utils/sphinx.rs index 48165259e0..6ce8baa32d 100644 --- a/src/utils/sphinx.rs +++ b/src/utils/sphinx.rs @@ -1,4 +1,4 @@ -use crate::utils::bytes; +use crate::utils::{bytes, topology}; use sphinx::route::{Destination, DestinationAddressBytes, Node, NodeAddressBytes, SURBIdentifier}; use sphinx::SphinxPacket; @@ -18,19 +18,24 @@ pub fn encapsulate_message( message: Vec, ) -> (NodeAddressBytes, SphinxPacket) { // here we would be getting topology, etc - let first_node_address = bytes::zero_pad_to_32("127.0.0.1:8080".as_bytes().to_vec()); - let dummy_route = vec![ - Node::new(first_node_address, Default::default()), - Node::new( - bytes::zero_pad_to_32("127.0.0.1:8081".as_bytes().to_vec()), - Default::default(), - ), - ]; - let delays = sphinx::header::delays::generate(dummy_route.len()); + let topology = topology::get_topology(true); + let mixes_route = topology::route_from(&topology, 1); + + let provider = Node::new( + topology::socket_bytes_from_string("127.0.0.1:8081".to_string()), + // bytes::zero_pad_to_32("127.0.0.1:8081".as_bytes().to_vec()), + Default::default(), + ); + + let route = [mixes_route, vec![provider]].concat(); + + let delays = sphinx::header::delays::generate(route.len()); // build the packet - let packet = sphinx::SphinxPacket::new(message, &dummy_route[..], &recipient, &delays).unwrap(); + let packet = sphinx::SphinxPacket::new(message, &route[..], &recipient, &delays).unwrap(); + + let first_node_address = route.first().unwrap().address; (first_node_address, packet) } diff --git a/src/utils/topology.rs b/src/utils/topology.rs new file mode 100644 index 0000000000..35ee9c2f1f --- /dev/null +++ b/src/utils/topology.rs @@ -0,0 +1,61 @@ +use crate::clients::directory; +use crate::clients::directory::presence::Topology; +use crate::clients::directory::requests::presence_topology_get::PresenceTopologyGetRequester; +use crate::clients::directory::DirectoryClient; +use crate::clients::mix::MixClient; +use crate::clients::provider::ProviderClient; +use crate::utils::bytes; +use curve25519_dalek::montgomery::MontgomeryPoint; +use sphinx::route::Node as SphinxNode; +use std::net::SocketAddrV4; +use std::string::ToString; + +pub(crate) fn get_topology(is_local: bool) -> Topology { + let url = if is_local { + "http://localhost:8080".to_string() + } else { + "https://directory.nymtech.net".to_string() + }; + println!("Using directory server: {:?}", url); + let directory_config = directory::Config { base_url: url }; + let directory = directory::Client::new(directory_config); + + let topology = directory + .presence_topology + .get() + .expect("Failed to retrieve network topology."); + topology +} + +pub(crate) fn route_from(topology: &Topology, route_len: usize) -> Vec { + let mut route = vec![]; + let nodes = topology.mix_nodes.iter(); + for mix in nodes.take(route_len) { + let address_bytes = socket_bytes_from_string(mix.host.clone()); + let decoded_key_bytes = base64::decode_config(&mix.pub_key, base64::URL_SAFE).unwrap(); + let key_bytes = bytes::zero_pad_to_32(decoded_key_bytes); + let key = MontgomeryPoint(key_bytes); + let sphinx_node = SphinxNode { + address: address_bytes, + pub_key: key, + }; + route.push(sphinx_node); + } + route +} + +pub(crate) fn socket_bytes_from_string(address: String) -> [u8; 32] { + let socket: SocketAddrV4 = address.parse().unwrap(); + let host_bytes = socket.ip().octets(); + let port_bytes = socket.port().to_be_bytes(); + let mut address_bytes = [0u8; 32]; + + address_bytes[0] = host_bytes[0]; + address_bytes[1] = host_bytes[1]; + address_bytes[2] = host_bytes[2]; + address_bytes[3] = host_bytes[3]; + + address_bytes[4] = port_bytes[0]; + address_bytes[5] = port_bytes[1]; + address_bytes +} From a978e6db9bffeab3fdc966440c1e997e60fa8978 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 17 Dec 2019 10:46:34 +0000 Subject: [PATCH 11/11] Uncommited work before switching websocket implementation --- src/clients/directory/presence.rs | 10 ++++---- src/clients/mod.rs | 39 ++++++++++++++++++------------- src/commands/run.rs | 3 ++- src/utils/sphinx.rs | 6 +++-- 4 files changed, 34 insertions(+), 24 deletions(-) diff --git a/src/clients/directory/presence.rs b/src/clients/directory/presence.rs index 62968eae19..20b666f171 100644 --- a/src/clients/directory/presence.rs +++ b/src/clients/directory/presence.rs @@ -1,6 +1,6 @@ use serde::{Deserialize, Serialize}; -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct CocoPresence { pub host: String, @@ -8,7 +8,7 @@ pub struct CocoPresence { pub last_seen: i64, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct MixNodePresence { pub host: String, @@ -17,7 +17,7 @@ pub struct MixNodePresence { pub last_seen: i64, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct MixProviderPresence { pub host: String, @@ -25,14 +25,14 @@ pub struct MixProviderPresence { pub registered_clients: Vec, } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct MixProviderClient { pub pub_key: String, } // Topology shows us the current state of the overall Nym network -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone)] #[serde(rename_all = "camelCase")] pub struct Topology { pub coco_nodes: Vec, diff --git a/src/clients/mod.rs b/src/clients/mod.rs index b07ca9a7d7..97b833f1b7 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -7,6 +7,8 @@ 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; pub mod directory; pub mod mix; @@ -56,6 +58,8 @@ pub struct NymClient { pub input_tx: mpsc::UnboundedSender, // to be used by "send" function or socket, etc input_rx: mpsc::UnboundedReceiver, + + is_local: bool, } @@ -63,28 +67,29 @@ pub struct NymClient { pub struct InputMessage(pub Destination, pub Vec); impl NymClient { - pub fn new(address: DestinationAddressBytes) -> Self { + pub fn new(address: DestinationAddressBytes, is_local: bool) -> Self { let (input_tx, input_rx) = mpsc::unbounded::(); NymClient { address, input_tx, input_rx, + is_local, } } - async fn start_loop_cover_traffic_stream(mut tx: mpsc::UnboundedSender, our_info: Destination) -> 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); + 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) -> 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! { @@ -92,13 +97,13 @@ impl NymClient { println!("[OUT QUEUE] - we got a real message!"); let real_message = real_message.expect("The channel must have closed! - if the client hasn't crashed, it should have!"); println!("real: {:?}", real_message); - let encapsulated_message = utils::sphinx::encapsulate_message(real_message.0, real_message.1); + let encapsulated_message = utils::sphinx::encapsulate_message(real_message.0, real_message.1, topology); mix_tx.send(MixMessage(encapsulated_message.0, encapsulated_message.1)).await?; }, default => { println!("[OUT QUEUE] - no real message - going to send extra loop cover"); - let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier); + 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?; } } @@ -127,20 +132,22 @@ impl NymClient { let mut rt = Runtime::new()?; rt.spawn(MixTrafficController::run(mix_rx)); - let foomp = Destination::new(self.address, Default::default()); +// 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 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); 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())), - NymClient::control_out_queue(mix_tx, self.input_rx, Destination::new(self.address, Default::default())), + 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 diff --git a/src/commands/run.rs b/src/commands/run.rs index d0b7ea3596..4e30fdad53 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -22,7 +22,8 @@ pub fn execute(matches: &ArgMatches) { // todo: to be taken from config or something let my_address = [42u8; 32]; - let client = NymClient::new(my_address); + 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/utils/sphinx.rs b/src/utils/sphinx.rs index 6ce8baa32d..2d0219b54b 100644 --- a/src/utils/sphinx.rs +++ b/src/utils/sphinx.rs @@ -1,3 +1,4 @@ +use crate::clients::directory::presence::Topology; use crate::utils::{bytes, topology}; use sphinx::route::{Destination, DestinationAddressBytes, Node, NodeAddressBytes, SURBIdentifier}; use sphinx::SphinxPacket; @@ -7,19 +8,20 @@ const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!"; pub fn loop_cover_message( our_address: DestinationAddressBytes, surb_id: SURBIdentifier, + topology: &Topology, ) -> (NodeAddressBytes, SphinxPacket) { let destination = Destination::new(our_address, surb_id); - encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec()) + encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), topology) } pub fn encapsulate_message( recipient: Destination, message: Vec, + topology: &Topology, ) -> (NodeAddressBytes, SphinxPacket) { // here we would be getting topology, etc - let topology = topology::get_topology(true); let mixes_route = topology::route_from(&topology, 1); let provider = Node::new(