From dfece9bda5b485b48ed6aa0ee9736251096b69c8 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Fri, 13 Dec 2019 16:03:19 +0000 Subject: [PATCH] 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) }