From 456494f05f4b4564fcc22bd6e6fe1907832bf05a Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 11 Dec 2019 15:17:47 +0000 Subject: [PATCH 1/6] Requests for messages --- src/clients/provider.rs | 160 ++++++++++++++++++++++++++++++++++++++++ src/commands/run.rs | 48 ++++++------ 2 files changed, 186 insertions(+), 22 deletions(-) diff --git a/src/clients/provider.rs b/src/clients/provider.rs index e69de29bb2..967d01d088 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -0,0 +1,160 @@ +use sphinx::route::Node as MixNode; +use sphinx::SphinxPacket; +use tokio::prelude::*; + +const PULL_REQUEST_MESSAGE_PREFIX: [u8; 2] = [1, 0]; +const REGISTER_MESSAGE_PREFIX: [u8; 2] = [0, 1]; + +// TODO: how to do it more nicely, considering all requests implement same trait that is exercised here? +pub enum ProviderRequests { + PullMessages(PullRequest), + Register(RegisterRequest), +} + +impl ProviderRequests { + fn to_bytes(&self) -> Vec { + use ProviderRequests::*; + match self { + PullMessages(pr) => pr.to_bytes(), + Register(pr) => pr.to_bytes(), + } + } + + fn from_bytes(bytes: &[u8]) -> Result { + use ProviderRequests::*; + if bytes.len() < 2 { + return Err(ProviderRequestError::UnmarshalError); + } + let mut received_prefix = [0; 2]; + received_prefix.copy_from_slice(&bytes[..2]); + match received_prefix { + PULL_REQUEST_MESSAGE_PREFIX => Ok(PullMessages(PullRequest::from_bytes(bytes)?)), + REGISTER_MESSAGE_PREFIX => Ok(Register(RegisterRequest::from_bytes(bytes)?)), + _ => Err(ProviderRequestError::UnmarshalErrorIncorrectPrefix), + } + } +} + +#[derive(Debug)] +pub enum ProviderRequestError { + MarshalError, + UnmarshalError, + UnmarshalErrorIncorrectPrefix, +} + +pub trait ProviderRequest where Self: Sized { + fn get_prefix() -> [u8; 2]; + fn to_bytes(&self) -> Vec; + fn from_bytes(bytes: &[u8]) -> Result; +} + +pub struct PullRequest { + // TODO: public keys, signatures, tokens, etc. basically some kind of authentication bs + destination_address: sphinx::route::DestinationAddressBytes, +} + +impl PullRequest { + pub(crate) fn new(destination_address: sphinx::route::DestinationAddressBytes) -> Self { + PullRequest { + destination_address + } + } +} + +impl ProviderRequest for PullRequest { + fn get_prefix() -> [u8; 2] { + PULL_REQUEST_MESSAGE_PREFIX + } + + fn to_bytes(&self) -> Vec { + Self::get_prefix().to_vec().into_iter().chain(self.destination_address.iter().cloned()).collect() + } + + fn from_bytes(bytes: &[u8]) -> Result { + if bytes.len() != 2 + 32 { + return Err(ProviderRequestError::UnmarshalError); + } + + let mut received_prefix = [0u8; 2]; + received_prefix.copy_from_slice(&bytes[..2]); + if received_prefix != Self::get_prefix() { + return Err(ProviderRequestError::UnmarshalErrorIncorrectPrefix); + } + + let mut destination_address = [0u8; 32]; + destination_address.copy_from_slice(&bytes[2..]); + + Ok(PullRequest { + destination_address, + }) + } +} + + +pub struct RegisterRequest {} + +impl ProviderRequest for RegisterRequest { + fn get_prefix() -> [u8; 2] { + unimplemented!() + } + + fn to_bytes(&self) -> Vec { + unimplemented!() + } + + fn from_bytes(bytes: &[u8]) -> Result { + unimplemented!() + } +} + +pub struct ProviderClient {} + + +impl ProviderClient { + pub fn new() -> Self { + ProviderClient {} + } + + pub async fn send( + &self, +// provider: &MixNode, + ) -> Result<(), Box> { + let address = [0; 32]; + let pull_request = PullRequest::new(address); + let bytes = pull_request.to_bytes(); + + let mut stream = tokio::net::TcpStream::connect("127.0.0.1:9000").await?; + stream.write_all(&bytes[..]).await?; + Ok(()) + } +} + +#[cfg(test)] +mod creating_pull_request { + use super::*; + + #[test] + fn it_is_possible_to_recover_it_from_bytes() { + let address = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2]; + let pull_request = PullRequest::new(address); + let bytes = pull_request.to_bytes(); + + let recovered = PullRequest::from_bytes(&bytes).unwrap(); + assert_eq!(address, recovered.destination_address); + } + + #[test] + fn it_is_possible_to_recover_it_from_bytes_with_enum_wrapper() { + let address = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2]; + let pull_request = PullRequest::new(address); + let bytes = pull_request.to_bytes(); + + let recovered = ProviderRequests::from_bytes(&bytes).unwrap(); + match recovered { + ProviderRequests::PullMessages(req) => assert_eq!(address, req.destination_address), + _ => panic!("expected to recover pull request!"), + } + } +} + + diff --git a/src/commands/run.rs b/src/commands/run.rs index 59e35c504d..fe715f7e11 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -8,6 +8,7 @@ use sphinx::route::Destination; use std::time::Duration; use tokio::runtime::Runtime; use tokio::time::{interval_at, Instant}; +use crate::clients::provider::ProviderClient; pub fn execute(matches: &ArgMatches) { let custom_cfg = matches.value_of("customCfg"); @@ -26,30 +27,33 @@ pub fn execute(matches: &ArgMatches) { let mut i: usize = 0; loop { interval.tick().await; - let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); - - // set up the route - let directory_config = directory::Config { - base_url: "https://directory.nymtech.net".to_string(), - }; - let directory = directory::Client::new(directory_config); - - let topology = directory - .presence_topology - .get() - .expect("Failed to retrieve network topology."); - let route = topology.mix_nodes; - let destination = get_destination(); - let delays = sphinx::header::delays::generate(2); - - // 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; +// let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); +// +// // set up the route +// let directory_config = directory::Config { +// base_url: "https://directory.nymtech.net".to_string(), +// }; +// let directory = directory::Client::new(directory_config); +// +// let topology = directory +// .presence_topology +// .get() +// .expect("Failed to retrieve network topology."); +// let route = topology.mix_nodes; +// let destination = get_destination(); +// let delays = sphinx::header::delays::generate(2); +// +// // 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; + + let provider_client = ProviderClient::new(); + provider_client.send().await.unwrap(); } }) } From 639187776ee6d87d8139294986e466257075d625 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 11 Dec 2019 15:50:29 +0000 Subject: [PATCH 2/6] Moved provider requests to subcrate in sfw provider --- Cargo.lock | 8 +++ Cargo.toml | 2 + src/clients/provider.rs | 136 +--------------------------------------- 3 files changed, 11 insertions(+), 135 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e92b1b450d..2ce803ba88 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -892,6 +892,7 @@ dependencies = [ "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)", + "sfw-provider-requests 0.1.0", "sphinx 0.1.0", "tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -1322,6 +1323,13 @@ dependencies = [ "url 1.7.2 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "sfw-provider-requests" +version = "0.1.0" +dependencies = [ + "sphinx 0.1.0", +] + [[package]] name = "sha2" version = "0.8.0" diff --git a/Cargo.toml b/Cargo.toml index 8e3a3a0d65..6485f5366c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,8 @@ clap = "2.33.0" reqwest = "0.9.22" serde = { version = "1.0", features = ["derive"] } sphinx = { path = "../sphinx" } +sfw-provider-requests = { path = "../nym-sfw-provider/sfw-provider-requests" } + tokio = { version = "0.2", features = ["full"] } [dev-dependencies] diff --git a/src/clients/provider.rs b/src/clients/provider.rs index 967d01d088..9bfe304a7b 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -1,111 +1,7 @@ use sphinx::route::Node as MixNode; use sphinx::SphinxPacket; use tokio::prelude::*; - -const PULL_REQUEST_MESSAGE_PREFIX: [u8; 2] = [1, 0]; -const REGISTER_MESSAGE_PREFIX: [u8; 2] = [0, 1]; - -// TODO: how to do it more nicely, considering all requests implement same trait that is exercised here? -pub enum ProviderRequests { - PullMessages(PullRequest), - Register(RegisterRequest), -} - -impl ProviderRequests { - fn to_bytes(&self) -> Vec { - use ProviderRequests::*; - match self { - PullMessages(pr) => pr.to_bytes(), - Register(pr) => pr.to_bytes(), - } - } - - fn from_bytes(bytes: &[u8]) -> Result { - use ProviderRequests::*; - if bytes.len() < 2 { - return Err(ProviderRequestError::UnmarshalError); - } - let mut received_prefix = [0; 2]; - received_prefix.copy_from_slice(&bytes[..2]); - match received_prefix { - PULL_REQUEST_MESSAGE_PREFIX => Ok(PullMessages(PullRequest::from_bytes(bytes)?)), - REGISTER_MESSAGE_PREFIX => Ok(Register(RegisterRequest::from_bytes(bytes)?)), - _ => Err(ProviderRequestError::UnmarshalErrorIncorrectPrefix), - } - } -} - -#[derive(Debug)] -pub enum ProviderRequestError { - MarshalError, - UnmarshalError, - UnmarshalErrorIncorrectPrefix, -} - -pub trait ProviderRequest where Self: Sized { - fn get_prefix() -> [u8; 2]; - fn to_bytes(&self) -> Vec; - fn from_bytes(bytes: &[u8]) -> Result; -} - -pub struct PullRequest { - // TODO: public keys, signatures, tokens, etc. basically some kind of authentication bs - destination_address: sphinx::route::DestinationAddressBytes, -} - -impl PullRequest { - pub(crate) fn new(destination_address: sphinx::route::DestinationAddressBytes) -> Self { - PullRequest { - destination_address - } - } -} - -impl ProviderRequest for PullRequest { - fn get_prefix() -> [u8; 2] { - PULL_REQUEST_MESSAGE_PREFIX - } - - fn to_bytes(&self) -> Vec { - Self::get_prefix().to_vec().into_iter().chain(self.destination_address.iter().cloned()).collect() - } - - fn from_bytes(bytes: &[u8]) -> Result { - if bytes.len() != 2 + 32 { - return Err(ProviderRequestError::UnmarshalError); - } - - let mut received_prefix = [0u8; 2]; - received_prefix.copy_from_slice(&bytes[..2]); - if received_prefix != Self::get_prefix() { - return Err(ProviderRequestError::UnmarshalErrorIncorrectPrefix); - } - - let mut destination_address = [0u8; 32]; - destination_address.copy_from_slice(&bytes[2..]); - - Ok(PullRequest { - destination_address, - }) - } -} - - -pub struct RegisterRequest {} - -impl ProviderRequest for RegisterRequest { - fn get_prefix() -> [u8; 2] { - unimplemented!() - } - - fn to_bytes(&self) -> Vec { - unimplemented!() - } - - fn from_bytes(bytes: &[u8]) -> Result { - unimplemented!() - } -} +use sfw_provider_requests::*; pub struct ProviderClient {} @@ -128,33 +24,3 @@ impl ProviderClient { Ok(()) } } - -#[cfg(test)] -mod creating_pull_request { - use super::*; - - #[test] - fn it_is_possible_to_recover_it_from_bytes() { - let address = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2]; - let pull_request = PullRequest::new(address); - let bytes = pull_request.to_bytes(); - - let recovered = PullRequest::from_bytes(&bytes).unwrap(); - assert_eq!(address, recovered.destination_address); - } - - #[test] - fn it_is_possible_to_recover_it_from_bytes_with_enum_wrapper() { - let address = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2]; - let pull_request = PullRequest::new(address); - let bytes = pull_request.to_bytes(); - - let recovered = ProviderRequests::from_bytes(&bytes).unwrap(); - match recovered { - ProviderRequests::PullMessages(req) => assert_eq!(address, req.destination_address), - _ => panic!("expected to recover pull request!"), - } - } -} - - From 39bbecb55d4187d7053f30a495a0eff9e0d01eaa Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 11 Dec 2019 16:42:25 +0000 Subject: [PATCH 3/6] Closing socket to server on being done --- src/clients/provider.rs | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/src/clients/provider.rs b/src/clients/provider.rs index 9bfe304a7b..eaba6db9b8 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -2,6 +2,7 @@ use sphinx::route::Node as MixNode; use sphinx::SphinxPacket; use tokio::prelude::*; use sfw_provider_requests::*; +use std::net::Shutdown; pub struct ProviderClient {} @@ -19,8 +20,20 @@ impl ProviderClient { let pull_request = PullRequest::new(address); let bytes = pull_request.to_bytes(); - let mut stream = tokio::net::TcpStream::connect("127.0.0.1:9000").await?; - stream.write_all(&bytes[..]).await?; + let mut socket = tokio::net::TcpStream::connect("127.0.0.1:9000").await?; + socket.write_all(&bytes[..]).await?; + if let Err(e) = socket.shutdown(Shutdown::Write) { + eprintln!("failed to close write part of the socket; err = {:?}", e) + } + + let mut response = Vec::new(); + socket.read_to_end(&mut response).await?; + if let Err(e) = socket.shutdown(Shutdown::Read) { + eprintln!("failed to close read part of the socket; err = {:?}", e) + } + + println!("Received the following response: {:?}", response); + Ok(()) } } From 536d86d6e3780c9309c92a5f7b9297c4d8668f70 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Dec 2019 13:56:31 +0000 Subject: [PATCH 4/6] Retrieving stored messages --- src/clients/provider.rs | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/src/clients/provider.rs b/src/clients/provider.rs index eaba6db9b8..fbf644b58c 100644 --- a/src/clients/provider.rs +++ b/src/clients/provider.rs @@ -3,6 +3,9 @@ use sphinx::SphinxPacket; use tokio::prelude::*; use sfw_provider_requests::*; use std::net::Shutdown; +use std::time::Duration; +use sfw_provider_requests::requests::{PullRequest, ProviderRequest}; +use sfw_provider_requests::responses::{PullResponse, ProviderResponse}; pub struct ProviderClient {} @@ -12,15 +15,17 @@ impl ProviderClient { ProviderClient {} } - pub async fn send( + pub async fn retrieve_messages( &self, // provider: &MixNode, ) -> Result<(), Box> { - let address = [0; 32]; + let address = [42; 32]; let pull_request = PullRequest::new(address); let bytes = pull_request.to_bytes(); let mut socket = tokio::net::TcpStream::connect("127.0.0.1:9000").await?; + println!("keep alive: {:?}", socket.keepalive()); + socket.set_keepalive(Some(tokio::time::Duration::from_secs(2))); socket.write_all(&bytes[..]).await?; if let Err(e) = socket.shutdown(Shutdown::Write) { eprintln!("failed to close write part of the socket; err = {:?}", e) @@ -33,7 +38,10 @@ impl ProviderClient { } println!("Received the following response: {:?}", response); - + let parsed_response = PullResponse::from_bytes(&response).unwrap(); + for message in parsed_response.messages { + println!("Received: {:?}", String::from_utf8(message).unwrap()) + } Ok(()) } } From f28eac7f94601fbc8663975c1bf33b5dddae87a0 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Dec 2019 14:01:02 +0000 Subject: [PATCH 5/6] changed default destination address to be non-zero --- src/commands/run.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/commands/run.rs b/src/commands/run.rs index 67926ccc00..c892fe6a08 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -92,7 +92,7 @@ fn route_from(topology: &Topology, route_len: usize) -> Vec { // TODO: where do we retrieve this guy from? fn get_destination() -> Destination { Destination { - address: [0u8; 32], - identifier: [0u8; 16], + address: [42u8; 32], + identifier: [1u8; 16], } } From ccca67acd5bc0c1036192803abe95b67a165303d Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 12 Dec 2019 14:15:08 +0000 Subject: [PATCH 6/6] Zeroed public keys to match local deployment + provider 'polling' --- src/commands/run.rs | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/src/commands/run.rs b/src/commands/run.rs index c892fe6a08..aa6cd27ce4 100644 --- a/src/commands/run.rs +++ b/src/commands/run.rs @@ -36,7 +36,7 @@ pub fn execute(matches: &ArgMatches) { interval.tick().await; let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec(); - let route_len = 3; + let route_len = 2; // data needed to generate a new Sphinx packet let route = route_from(&topology, route_len); @@ -53,8 +53,13 @@ pub fn execute(matches: &ArgMatches) { println!("packet sent: {:?}", i); i += 1; - let provider_client = ProviderClient::new(); - provider_client.send().await.unwrap(); + // 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(); + } } }) } @@ -80,10 +85,13 @@ fn route_from(topology: &Topology, route_len: usize) -> Vec { 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 { + let mut sphinx_node = SphinxNode { address: address_bytes, pub_key: key, }; + + // temporary to make it work locally: + sphinx_node.pub_key = Default::default(); route.push(sphinx_node); } route