diff --git a/Cargo.lock b/Cargo.lock index ec7632c9b2..2b061f193c 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" @@ -952,8 +1029,11 @@ 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)", + "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)", @@ -1047,6 +1127,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" @@ -1057,6 +1142,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" @@ -1997,8 +2097,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" @@ -2050,8 +2158,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 13776b879a..6f7837feec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,10 @@ 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" } diff --git a/src/clients/directory/presence.rs b/src/clients/directory/presence.rs index 1d2e56f7e1..d693ff7d7e 100644 --- a/src/clients/directory/presence.rs +++ b/src/clients/directory/presence.rs @@ -1,6 +1,6 @@ use serde::{Deserialize, Serialize}; -#[derive(Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct CocoPresence { pub host: String, @@ -8,7 +8,7 @@ pub struct CocoPresence { pub last_seen: i64, } -#[derive(Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct MixNodePresence { pub host: String, @@ -17,7 +17,7 @@ pub struct MixNodePresence { pub last_seen: u64, } -#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct MixProviderPresence { pub host: String, @@ -25,14 +25,14 @@ pub struct MixProviderPresence { pub registered_clients: Vec, } -#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct MixProviderClient { pub pub_key: String, } // Topology shows us the current state of the overall Nym network -#[derive(Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct Topology { pub coco_nodes: Vec, diff --git a/src/clients/mix.rs b/src/clients/mix.rs index 0150bc5c45..65480bf539 100644 --- a/src/clients/mix.rs +++ b/src/clients/mix.rs @@ -1,5 +1,6 @@ -use sphinx::route::Node as MixNode; +use sphinx::route::{Node as MixNode, NodeAddressBytes}; use sphinx::SphinxPacket; +use std::net::SocketAddr; use std::net::{Ipv4Addr, SocketAddrV4}; use tokio::prelude::*; @@ -14,14 +15,15 @@ impl MixClient { pub async fn send( &self, packet: SphinxPacket, - mix: &MixNode, + mix_addr: NodeAddressBytes, ) -> Result<(), Box> { let bytes = packet.to_bytes(); - let b = mix.address; + let b = mix_addr; 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 e0ee8b42df..9e5064a2d7 100644 --- a/src/clients/mod.rs +++ b/src/clients/mod.rs @@ -1,4 +1,187 @@ +use crate::clients::directory::presence::Topology; +use crate::clients::mix::MixClient; +use crate::clients::provider::ProviderClient; +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::net::SocketAddrV4; +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 = 1.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 + +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 + ); + 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), + }; + } + } +} + +pub struct NymClient { + // to be replaced by something else I guess + address: DestinationAddressBytes, + pub input_tx: mpsc::UnboundedSender, + // to be used by "send" function or socket, etc + input_rx: mpsc::UnboundedReceiver, + + is_local: bool, +} + +#[derive(Debug)] +pub struct InputMessage(pub Destination, pub Vec); + +impl NymClient { + 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, + 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?; + } + } + + 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! { + 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, 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, 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( + provider_address: SocketAddrV4, + ) -> Result<(), Box> { + let provider_client = ProviderClient::new(); + + loop { + println!("[FETCH MSG] - Polling provider..."); + let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY); + tokio::time::delay_for(delay_duration).await; + provider_client + .retrieve_messages(provider_address) + .await + .unwrap(); + } + } + + pub fn start(self) -> Result<(), Box> { + println!("starting nym client"); + + let (mix_tx, mix_rx) = mpsc::unbounded(); + + let mut rt = Runtime::new()?; + rt.spawn(MixTrafficController::run(mix_rx)); + let topology = get_topology(self.is_local); + let provider_address: SocketAddrV4 = topology + .mix_provider_nodes + .first() + .unwrap() + .host + .parse() + .unwrap(); + + 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(provider_address), + ) + .await; + + // start websocket handler here + 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(()) + } +} diff --git a/src/clients/provider.rs b/src/clients/provider.rs index e0b03dbaec..ff2fdb6bac 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -14,13 +14,19 @@ impl ProviderClient { pub async fn retrieve_messages( &self, - provider: &SocketAddrV4, + provider: SocketAddrV4, ) -> Result<(), Box> { let address = [42; 32]; let pull_request = PullRequest::new(address); let bytes = pull_request.to_bytes(); - let mut socket = tokio::net::TcpStream::connect(provider).await?; + // DH temporary: the provider's client port is not in the topology, but we can't change that + // right now without messing up the existing Go mixnet. So I'm going to hardcode this + // for the moment until the Go mixnet goes away. + let provider_socket = SocketAddrV4::new(*provider.ip(), 9000); + println!("Provider: {:?}", provider_socket); + + let mut socket = tokio::net::TcpStream::connect(provider_socket).await?; println!("keep alive: {:?}", socket.keepalive()); socket.set_keepalive(Some(Duration::from_secs(2))).unwrap(); socket.write_all(&bytes[..]).await?; diff --git a/src/commands/run.rs b/src/commands/run.rs index f1e7218fdf..5f3e5af77f 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; @@ -19,110 +20,12 @@ pub fn execute(matches: &ArgMatches) { 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); - let provider_address: SocketAddrV4 = topology - .mix_provider_nodes - .first() - .unwrap() - .host - .parse() - .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 = 1; // TODO: kill magic number - - // 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(); - mix_client - .send(packet, route.first().unwrap()) - .await - .unwrap(); - 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(&provider_address) - .await - .unwrap(); - } - } - }) -} - -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 -} - -// Parses a String socket address, then returns it as a -// 4+2 byte array, zero-padded to a constant size of 32 bytes. -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 } // TODO: where do we retrieve this guy from? 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 ad0049c4c2..ee03437270 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -1 +1,4 @@ pub mod bytes; +pub mod poisson; +pub mod sphinx; +pub mod topology; 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()) +} diff --git a/src/utils/sphinx.rs b/src/utils/sphinx.rs new file mode 100644 index 0000000000..0aabbd3194 --- /dev/null +++ b/src/utils/sphinx.rs @@ -0,0 +1,48 @@ +use crate::clients::directory::presence::Topology; +use crate::utils::{bytes, topology}; +use curve25519_dalek::montgomery::MontgomeryPoint; +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, + topology: &Topology, +) -> (NodeAddressBytes, SphinxPacket) { + let destination = Destination::new(our_address, surb_id); + + 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 mixes_route = topology::route_from(&topology, 1); + let first_provider = topology.mix_provider_nodes.first().unwrap(); + let decoded_key_bytes = + base64::decode_config(&first_provider.pub_key, base64::URL_SAFE).unwrap(); + let key_bytes = bytes::zero_pad_to_32(decoded_key_bytes); + let key = MontgomeryPoint(key_bytes); + + let provider = Node::new( + topology::socket_bytes_from_string(first_provider.host.clone()), + key, + ); + + let route = [mixes_route, vec![provider]].concat(); + + let delays = sphinx::header::delays::generate(route.len()); + + // build the packet + 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 +}