From 566e17719c8e8691a0da7e7bb76488ccec0fe5ee Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 22 Jan 2020 12:22:05 +0000 Subject: [PATCH 01/13] added PartialEq restriction on NymTopology for convenience sake --- common/topology/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index dd29508ef2..0cd511de3f 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -83,7 +83,7 @@ pub enum NymTopologyError { MissingLayerError(Vec), } -pub trait NymTopology: Sized { +pub trait NymTopology: Sized + PartialEq { fn new(directory_server: String) -> Self; fn new_from_nodes( mix_nodes: Vec, From 91eb03fbb6b9959946f3c1242c325083741a2205 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Wed, 22 Jan 2020 12:22:15 +0000 Subject: [PATCH 02/13] Derived PartialEq on Topology --- common/clients/directory-client/src/presence.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/common/clients/directory-client/src/presence.rs b/common/clients/directory-client/src/presence.rs index f20ed9f914..b333e9fce9 100644 --- a/common/clients/directory-client/src/presence.rs +++ b/common/clients/directory-client/src/presence.rs @@ -6,7 +6,7 @@ use std::io; use std::net::ToSocketAddrs; use topology::{CocoNode, MixNode, MixProviderNode, NymTopology}; -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct CocoPresence { pub host: String, @@ -37,7 +37,7 @@ impl From for CocoPresence { } } -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct MixNodePresence { pub host: String, @@ -81,7 +81,7 @@ impl From for MixNodePresence { } } -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct MixProviderPresence { pub client_listener: String, @@ -126,7 +126,7 @@ impl From for MixProviderPresence { } } -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct MixProviderClient { pub pub_key: String, @@ -149,7 +149,7 @@ impl From for MixProviderClient { } // Topology shows us the current state of the overall Nym network -#[derive(Clone, Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct Topology { pub coco_nodes: Vec, From e5b3a29c5f1b29bc0d9a7d172edf5f56465f7cae Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 16:53:02 +0000 Subject: [PATCH 03/13] Defined ability to compare two 'Topology' instances The comparison ignores 'last-seen' field or ordering of nodes and clients --- .../clients/directory-client/src/presence.rs | 155 +++++++++++++++++- 1 file changed, 149 insertions(+), 6 deletions(-) diff --git a/common/clients/directory-client/src/presence.rs b/common/clients/directory-client/src/presence.rs index b333e9fce9..f691bc9035 100644 --- a/common/clients/directory-client/src/presence.rs +++ b/common/clients/directory-client/src/presence.rs @@ -1,12 +1,19 @@ use crate::requests::presence_topology_get::PresenceTopologyGetRequester; use crate::{Client, Config, DirectoryClient}; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; use std::convert::TryInto; use std::io; use std::net::ToSocketAddrs; use topology::{CocoNode, MixNode, MixProviderNode, NymTopology}; -#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +// special version of 'PartialEq' that does not care about last_seen field (where applicable) +// or order of elements in vectors +trait PresenceEq { + fn presence_eq(&self, other: &Self) -> bool; +} + +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct CocoPresence { pub host: String, @@ -37,7 +44,7 @@ impl From for CocoPresence { } } -#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct MixNodePresence { pub host: String, @@ -81,7 +88,7 @@ impl From for MixNodePresence { } } -#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct MixProviderPresence { pub client_listener: String, @@ -92,6 +99,32 @@ pub struct MixProviderPresence { pub version: String, } +impl PresenceEq for MixProviderPresence { + fn presence_eq(&self, other: &Self) -> bool { + if self.registered_clients.len() != other.registered_clients.len() { + return false; + } + + if self.client_listener != other.client_listener + || self.mixnet_listener != other.mixnet_listener + || self.pub_key != other.pub_key + || self.version != other.version + { + return false; + } + + let clients_self_set: HashSet<_> = + self.registered_clients.iter().map(|c| &c.pub_key).collect(); + let clients_other_set: HashSet<_> = other + .registered_clients + .iter() + .map(|c| &c.pub_key) + .collect(); + + clients_self_set == clients_other_set + } +} + impl Into for MixProviderPresence { fn into(self) -> topology::MixProviderNode { topology::MixProviderNode { @@ -126,7 +159,7 @@ impl From for MixProviderPresence { } } -#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct MixProviderClient { pub pub_key: String, @@ -148,8 +181,102 @@ impl From for MixProviderClient { } } +impl PresenceEq for Vec { + fn presence_eq(&self, other: &Self) -> bool { + if self.len() != other.len() { + return false; + } + + // we can't take the whole thing into set as it does not implement 'Eq' and we can't + // derive it as we don't want to take 'last_seen' into consideration + let self_set: HashSet<_> = self + .iter() + .map(|c| (&c.host, &c.pub_key, &c.version)) + .collect(); + let other_set: HashSet<_> = other + .iter() + .map(|c| (&c.host, &c.pub_key, &c.version)) + .collect(); + + self_set == other_set + } +} + +impl PresenceEq for Vec { + fn presence_eq(&self, other: &Self) -> bool { + if self.len() != other.len() { + return false; + } + + // we can't take the whole thing into set as it does not implement 'Eq' and we can't + // derive it as we don't want to take 'last_seen' into consideration + let self_set: HashSet<_> = self + .iter() + .map(|m| (&m.host, &m.pub_key, &m.version, &m.layer)) + .collect(); + let other_set: HashSet<_> = other + .iter() + .map(|m| (&m.host, &m.pub_key, &m.version, &m.layer)) + .collect(); + + self_set == other_set + } +} + +impl PresenceEq for Vec { + fn presence_eq(&self, other: &Self) -> bool { + if self.len() != other.len() { + return false; + } + + // we can't take the whole thing into set as it does not implement 'Eq' and we can't + // derive it as we don't want to take 'last_seen' into consideration. + // We also don't care about order of registered_clients + + // since we're going to be getting rid of this very soon anyway, just clone registered + // clients vector and sort it + + let self_set: HashSet<_> = self + .iter() + .map(|p| { + ( + &p.client_listener, + &p.mixnet_listener, + &p.pub_key, + &p.version, + p.registered_clients + .iter() + .cloned() + .map(|c| c.pub_key) + .collect::>() + .sort(), + ) + }) + .collect(); + let other_set: HashSet<_> = other + .iter() + .map(|p| { + ( + &p.client_listener, + &p.mixnet_listener, + &p.pub_key, + &p.version, + p.registered_clients + .iter() + .cloned() + .map(|c| c.pub_key) + .collect::>() + .sort(), + ) + }) + .collect(); + + self_set == other_set + } +} + // Topology shows us the current state of the overall Nym network -#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +#[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct Topology { pub coco_nodes: Vec, @@ -157,9 +284,25 @@ pub struct Topology { pub mix_provider_nodes: Vec, } +impl PartialEq for Topology { + // we need a custom implementation as the order of nodes does not matter + // also we do not care about 'last_seen' when comparing topologies + fn eq(&self, other: &Self) -> bool { + if !self.coco_nodes.presence_eq(&other.coco_nodes) + || !self.mix_nodes.presence_eq(&other.mix_nodes) + || !self + .mix_provider_nodes + .presence_eq(&other.mix_provider_nodes) + { + return false; + } + + true + } +} + impl NymTopology for Topology { fn new(directory_server: String) -> Self { - println!("Using directory server: {:?}", directory_server); let directory_config = Config { base_url: directory_server, }; From 52272d6119cbad53dac099f70113ca7a39ffc943 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 16:53:30 +0000 Subject: [PATCH 04/13] Added restriction on NymTopology to require Debug and Send + Sync --- common/topology/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 0cd511de3f..2833c0dd08 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -83,7 +83,7 @@ pub enum NymTopologyError { MissingLayerError(Vec), } -pub trait NymTopology: Sized + PartialEq { +pub trait NymTopology: Sized + PartialEq + std::fmt::Debug + Send + Sync { fn new(directory_server: String) -> Self; fn new_from_nodes( mix_nodes: Vec, From 8c26ef848a8b9d4e169777a657727e452d318274 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 16:54:16 +0000 Subject: [PATCH 05/13] Created a module responsible for periodically refreshing topology --- nym-client/src/client/topology_control.rs | 140 ++++++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 nym-client/src/client/topology_control.rs diff --git a/nym-client/src/client/topology_control.rs b/nym-client/src/client/topology_control.rs new file mode 100644 index 0000000000..8adf953dfb --- /dev/null +++ b/nym-client/src/client/topology_control.rs @@ -0,0 +1,140 @@ +use crate::built_info; +use log::{error, info, trace, warn}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::RwLock as FRwLock; +use topology::NymTopology; + +const NODE_HEALTH_THRESHOLD: f64 = 0.0; + +// auxiliary type for ease of use +pub type TopologyInnerRef = Arc>>; + +pub(crate) struct TopologyControl { + directory_server: String, + refresh_rate: f64, + inner: Arc>>, +} + +#[derive(Debug)] +enum TopologyError { + HealthCheckError, + NoValidPathsError, +} + +impl TopologyControl { + pub(crate) async fn new(directory_server: String, refresh_rate: f64) -> Self { + let initial_topology = match Self::get_current_compatible_topology(directory_server.clone()) + .await + { + Ok(topology) => Some(topology), + Err(err) => { + error!("Initial topology is invalid - {:?}. Right now it will be impossible to send any packets through the mixnet!", err); + None + } + }; + + TopologyControl { + directory_server, + refresh_rate, + inner: Arc::new(FRwLock::new(Inner::new(initial_topology))), + } + } + + async fn get_current_compatible_topology(directory_server: String) -> Result { + let full_topology = T::new(directory_server.clone()); + + // run a healthcheck to determine healthy-ish nodes: + // this is a temporary solution as the healthcheck will eventually be moved to validators + let healthcheck_config = healthcheck::config::HealthCheck { + directory_server, + // those are literally irrelevant when running single check + interval: 100000.0, + resolution_timeout: 5.0, + num_test_packets: 2, + }; + let healthcheck = healthcheck::HealthChecker::new(healthcheck_config); + let healthcheck_result = healthcheck.do_check().await; + + let healthcheck_scores = match healthcheck_result { + Err(err) => { + error!("Error while performing the healtcheck: {:?}", err); + return Err(TopologyError::HealthCheckError); + } + Ok(scores) => scores, + }; + + let healthy_topology = + healthcheck_scores.filter_topology_by_score(&full_topology, NODE_HEALTH_THRESHOLD); + + // for time being assume same versioning, i.e. if client is running X.Y.Z, + // we're expecting mixes, providers and coconodes to also be running X.Y.Z + let versioned_healthy_topology = + healthy_topology.filter_node_versions("0.3.2", "0.3.2", built_info::PKG_VERSION); + + // make sure you can still send a packet through the network: + if !versioned_healthy_topology.can_construct_path_through() { + return Err(TopologyError::NoValidPathsError); + } + + Ok(versioned_healthy_topology) + } + + pub(crate) fn get_inner_ref(&self) -> Arc>> { + self.inner.clone() + } + + async fn update_global_topology(&mut self, new_topology: Option) { + // acquire write lock + let mut write_lock = self.inner.write().await; + write_lock.topology = new_topology; + } + + async fn should_update_topology(&mut self, new_topology: &Option) -> bool { + let read_lock = self.inner.read().await; + match new_topology { + // if new topology is invalid, we MUST update to it as it is impossible to send packets through + None => true, + Some(new_topology) => match &read_lock.topology { + None => true, + Some(old_topology) => new_topology != old_topology, + }, + } + } + + pub(crate) async fn run_refresher(mut self) { + info!("Starting topology refresher"); + let delay_duration = Duration::from_secs_f64(self.refresh_rate); + loop { + trace!("Refreshing the topology"); + let new_topology_res = + Self::get_current_compatible_topology(self.directory_server.clone()).await; + + let new_topology = match new_topology_res { + Ok(topology) => Some(topology), + Err(err) => { + warn!("the obtained topology seems to be invalid - {:?}, it will be impossible to send packets through", err); + None + } + }; + + if self.should_update_topology(&new_topology).await { + info!("Detected changes in topology - updating global view!"); + + self.update_global_topology(new_topology).await; + } + + tokio::time::delay_for(delay_duration).await; + } + } +} + +pub struct Inner { + pub topology: Option, +} + +impl Inner { + fn new(topology: Option) -> Self { + Inner { topology } + } +} From 2079b8f9262b4e3d3dbd9481acbceeb31b4b5066 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 16:54:43 +0000 Subject: [PATCH 06/13] Said module used by cover traffic stream --- nym-client/src/client/cover_traffic_stream.rs | 28 +++++++++++-------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/nym-client/src/client/cover_traffic_stream.rs b/nym-client/src/client/cover_traffic_stream.rs index 82f904eec7..f1caedc740 100644 --- a/nym-client/src/client/cover_traffic_stream.rs +++ b/nym-client/src/client/cover_traffic_stream.rs @@ -1,29 +1,35 @@ use crate::client::mix_traffic::MixMessage; +use crate::client::topology_control::TopologyInnerRef; use crate::client::LOOP_COVER_AVERAGE_DELAY; use futures::channel::mpsc; -use log::{info, trace}; +use log::{info, trace, warn}; use sphinx::route::Destination; use std::time::Duration; use topology::NymTopology; -pub(crate) async fn start_loop_cover_traffic_stream( +pub(crate) async fn start_loop_cover_traffic_stream( tx: mpsc::UnboundedSender, our_info: Destination, - topology: T, -) where - T: NymTopology, -{ + topology_ctrl_ref: TopologyInnerRef, +) { info!("Starting loop cover traffic stream"); loop { trace!("next cover message!"); let delay = mix_client::poisson::sample(LOOP_COVER_AVERAGE_DELAY); let delay_duration = Duration::from_secs_f64(delay); tokio::time::delay_for(delay_duration).await; - let cover_message = mix_client::packet::loop_cover_message( - our_info.address, - our_info.identifier, - &topology, - ); + + let read_lock = topology_ctrl_ref.read().await; + let topology = read_lock.topology.as_ref(); + if topology.is_none() { + warn!("No valid topology detected - won't send any loop cover message this time"); + continue; + } + + let topology = topology.unwrap(); + + let cover_message = + mix_client::packet::loop_cover_message(our_info.address, our_info.identifier, topology); // if this one fails, there's no retrying because it means that either: // - we run out of memory From 35451a05de2760c6d2e125dbdc8bb57f37baa65d Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 16:55:11 +0000 Subject: [PATCH 07/13] And by real traffic stream + necessary adjustments --- nym-client/src/client/real_traffic_stream.rs | 81 +++++++++++--------- 1 file changed, 45 insertions(+), 36 deletions(-) diff --git a/nym-client/src/client/real_traffic_stream.rs b/nym-client/src/client/real_traffic_stream.rs index f89257747b..66a80a2ebe 100644 --- a/nym-client/src/client/real_traffic_stream.rs +++ b/nym-client/src/client/real_traffic_stream.rs @@ -1,38 +1,38 @@ use crate::client::mix_traffic::MixMessage; +use crate::client::topology_control::TopologyInnerRef; use crate::client::{InputMessage, MESSAGE_SENDING_AVERAGE_DELAY}; -use directory_client::presence::Topology; use futures::channel::mpsc; use futures::task::{Context, Poll}; use futures::{Future, Stream, StreamExt}; -use log::{debug, info, trace}; +use log::{info, trace, warn}; use sphinx::route::Destination; -use sphinx::SphinxPacket; -use std::net::SocketAddr; use std::pin::Pin; use std::time::Duration; use tokio::time; +use topology::NymTopology; // have a rather low value for test sake const AVERAGE_PACKET_DELAY: f64 = 0.1; -pub(crate) struct OutQueueControl { +pub(crate) struct OutQueueControl { delay: time::Delay, mix_tx: mpsc::UnboundedSender, input_rx: mpsc::UnboundedReceiver, our_info: Destination, - - // due to pinning, DerefMut trait, futures, etc its way easier to - // just have concrete implementation here rather than generic NymTopology - // considering that it will be replaced with refreshing topology within few days anyway - topology: Topology, + topology_ctrl_ref: TopologyInnerRef, } -impl Stream for OutQueueControl { - type Item = (SocketAddr, SphinxPacket); +pub(crate) enum StreamMessage { + Cover, + Real(InputMessage), +} + +impl Stream for OutQueueControl { + type Item = StreamMessage; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { // it is not yet time to return a message - if Future::poll(Pin::new(&mut self.delay), cx).is_pending() { + if Pin::new(&mut self.delay).poll(cx).is_pending() { return Poll::Pending; }; @@ -49,41 +49,26 @@ impl Stream for OutQueueControl { self.delay.reset(next); // decide what kind of message to send - match Stream::poll_next(Pin::new(&mut self.input_rx), cx) { + match Pin::new(&mut self.input_rx).poll_next(cx) { // in the case our real message channel stream was closed, we should also indicate we are closed // (and whoever is using the stream should panic) Poll::Ready(None) => Poll::Ready(None), // if there's an actual message - return it - Poll::Ready(Some(real_message)) => { - trace!("real message"); - Poll::Ready(Some(mix_client::packet::encapsulate_message( - real_message.0, - real_message.1, - &self.topology, - AVERAGE_PACKET_DELAY, - ))) - } + Poll::Ready(Some(real_message)) => Poll::Ready(Some(StreamMessage::Real(real_message))), // otherwise construct a dummy one - _ => { - trace!("loop cover message"); - Poll::Ready(Some(mix_client::packet::loop_cover_message( - self.our_info.address, - self.our_info.identifier, - &self.topology, - ))) - } + Poll::Pending => Poll::Ready(Some(StreamMessage::Cover)), } } } -impl OutQueueControl { +impl OutQueueControl { pub(crate) fn new( mix_tx: mpsc::UnboundedSender, input_rx: mpsc::UnboundedReceiver, our_info: Destination, - topology: Topology, + topology: TopologyInnerRef, ) -> Self { let initial_delay = time::delay_for(Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY)); OutQueueControl { @@ -91,20 +76,44 @@ impl OutQueueControl { mix_tx, input_rx, our_info, - topology, + topology_ctrl_ref: topology, } } pub(crate) async fn run_out_queue_control(mut self) { info!("starting out queue controller"); while let Some(next_message) = self.next().await { - debug!("created new message"); + trace!("created new message"); + let read_lock = self.topology_ctrl_ref.read().await; + let topology = read_lock.topology.as_ref(); + + if topology.is_none() { + warn!("No valid topology detected - won't send any loop cover message this time"); + continue; + } + + let topology = topology.unwrap(); + + let next_packet = match next_message { + StreamMessage::Cover => mix_client::packet::loop_cover_message( + self.our_info.address, + self.our_info.identifier, + topology, + ), + StreamMessage::Real(real_message) => mix_client::packet::encapsulate_message( + real_message.0, + real_message.1, + topology, + AVERAGE_PACKET_DELAY, + ), + }; + // if this one fails, there's no retrying because it means that either: // - we run out of memory // - the receiver channel is closed // in either case there's no recovery and we can only panic self.mix_tx - .unbounded_send(MixMessage::new(next_message.0, next_message.1)) + .unbounded_send(MixMessage::new(next_packet.0, next_packet.1)) .unwrap(); } } From 6f69641443cb83af6f7d29919f761cdf020d404c Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 16:55:39 +0000 Subject: [PATCH 08/13] Made sockets more generic to work on any NymTopology object --- nym-client/src/sockets/tcp.rs | 52 +++++++++++++++++++---------------- nym-client/src/sockets/ws.rs | 50 +++++++++++++++++++-------------- 2 files changed, 58 insertions(+), 44 deletions(-) diff --git a/nym-client/src/sockets/tcp.rs b/nym-client/src/sockets/tcp.rs index c063c7e4b8..7270453a9e 100644 --- a/nym-client/src/sockets/tcp.rs +++ b/nym-client/src/sockets/tcp.rs @@ -1,17 +1,16 @@ use crate::client::received_buffer::BufferResponse; +use crate::client::topology_control::TopologyInnerRef; use crate::client::InputMessage; -use directory_client::presence::Topology; use futures::channel::{mpsc, oneshot}; use futures::future::FutureExt; use futures::io::Error; use futures::SinkExt; use sphinx::route::{Destination, DestinationAddressBytes}; -use std::borrow::Borrow; use std::convert::TryFrom; use std::io; use std::net::SocketAddr; -use std::sync::Arc; use tokio::prelude::*; +use topology::NymTopology; const SEND_REQUEST_PREFIX: u8 = 1; const FETCH_REQUEST_PREFIX: u8 = 2; @@ -129,15 +128,24 @@ impl ClientRequest { ServerResponse::Fetch { messages } } - async fn handle_get_clients(topology: &Topology) -> ServerResponse { - println!("get client handle"); - let clients = topology - .mix_provider_nodes - .iter() - .flat_map(|provider| provider.registered_clients.iter()) - .map(|client| base64::decode_config(&client.pub_key, base64::URL_SAFE).unwrap()) // TODO: this can potentially throw an error - .collect(); - ServerResponse::GetClients { clients } + async fn handle_get_clients(topology: &TopologyInnerRef) -> ServerResponse { + let topology_data = &topology.read().await.topology; + match topology_data { + Some(topology) => { + let clients = topology + .get_mix_provider_nodes() + .iter() + .flat_map(|provider| provider.registered_clients.iter()) + .filter_map(|client| { + base64::decode_config(&client.pub_key, base64::URL_SAFE).ok() + }) + .collect(); + ServerResponse::GetClients { clients } + } + None => ServerResponse::Error { + message: "Invalid network topology".to_string(), + }, + } } async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse { @@ -203,9 +211,9 @@ impl ServerResponse { } } -async fn handle_connection( +async fn handle_connection( data: &[u8], - request_handling_data: RequestHandlingData, + request_handling_data: RequestHandlingData, ) -> Result { let request = ClientRequest::try_from(data)?; let response = match request { @@ -218,7 +226,7 @@ async fn handle_connection( } ClientRequest::Fetch => ClientRequest::handle_fetch(request_handling_data.msg_query).await, ClientRequest::GetClients => { - ClientRequest::handle_get_clients(request_handling_data.topology.borrow()).await + ClientRequest::handle_get_clients(&request_handling_data.topology).await } ClientRequest::OwnDetails => { ClientRequest::handle_own_details(request_handling_data.self_address).await @@ -228,27 +236,25 @@ async fn handle_connection( Ok(response) } -struct RequestHandlingData { +struct RequestHandlingData { msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Arc, + topology: TopologyInnerRef, } -async fn accept_connection( +async fn accept_connection( mut socket: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, ) { let address = socket .peer_addr() .expect("connected streams should have a peer address"); println!("Peer address: {}", address); - let topology = Arc::new(topology); - let mut buf = [0u8; 2048]; // In a loop, read data from the socket and write the data back. @@ -287,12 +293,12 @@ async fn accept_connection( } } -pub async fn start_tcpsocket( +pub async fn start_tcpsocket( address: SocketAddr, message_tx: mpsc::UnboundedSender, received_messages_query_tx: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, ) -> Result<(), TCPSocketError> { let mut listener = tokio::net::TcpListener::bind(address).await?; diff --git a/nym-client/src/sockets/ws.rs b/nym-client/src/sockets/ws.rs index 14aacdcb5d..10f9eb4020 100644 --- a/nym-client/src/sockets/ws.rs +++ b/nym-client/src/sockets/ws.rs @@ -1,6 +1,6 @@ use crate::client::received_buffer::BufferResponse; +use crate::client::topology_control::TopologyInnerRef; use crate::client::InputMessage; -use directory_client::presence::Topology; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; use futures::channel::{mpsc, oneshot}; use futures::future::FutureExt; @@ -12,20 +12,21 @@ use sphinx::route::{Destination, DestinationAddressBytes}; use std::convert::TryFrom; use std::io; use std::net::SocketAddr; +use topology::NymTopology; use tungstenite::protocol::frame::coding::CloseCode; use tungstenite::protocol::{CloseFrame, Message}; -struct Connection { +struct Connection { address: SocketAddr, msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, rx: UnboundedReceiver, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, tx: UnboundedSender, } -impl Connection { +impl Connection { async fn handle_text_message(&self, msg: String) -> ServerResponse { debug!("Handling text message request"); trace!("Content: {:?}", msg.clone()); @@ -48,9 +49,7 @@ impl Connection { ClientRequest::handle_send(message, recipient_address, self.msg_input.clone()).await } ClientRequest::Fetch => ClientRequest::handle_fetch(self.msg_query.clone()).await, - ClientRequest::GetClients => { - ClientRequest::handle_get_clients(self.topology.clone()).await - } + ClientRequest::GetClients => ClientRequest::handle_get_clients(&self.topology).await, ClientRequest::OwnDetails => ClientRequest::handle_own_details(self.self_address).await, } } @@ -212,7 +211,7 @@ impl ClientRequest { Err(e) => { return ServerResponse::Error { message: e.to_string(), - } + }; } Ok(hex) => hex, }; @@ -269,14 +268,22 @@ impl ClientRequest { ServerResponse::Fetch { messages } } - async fn handle_get_clients(topology: Topology) -> ServerResponse { - let clients = topology - .mix_provider_nodes - .into_iter() - .flat_map(|provider| provider.registered_clients.into_iter()) - .map(|client| client.pub_key) - .collect(); - ServerResponse::GetClients { clients } + async fn handle_get_clients(topology: &TopologyInnerRef) -> ServerResponse { + let topology_data = &topology.read().await.topology; + match topology_data { + Some(topology) => { + let clients = topology + .get_mix_provider_nodes() + .iter() + .flat_map(|provider| provider.registered_clients.iter()) + .map(|client| client.pub_key.clone()) + .collect(); + ServerResponse::GetClients { clients } + } + None => ServerResponse::Error { + message: "Invalid network topology".to_string(), + }, + } } async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse { @@ -306,14 +313,13 @@ impl Into for ServerResponse { } } -async fn accept_connection( +async fn accept_connection( stream: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, ) { - warn!("accept_connection"); let address = stream .peer_addr() .expect("connected streams should have a peer address"); @@ -337,6 +343,8 @@ async fn accept_connection( msg_query, self_address, }; + + // TODO: make sure this actually doesn't leak memory... tokio::spawn(conn.handle()); while let Some(message) = ws_stream.next().await { @@ -385,12 +393,12 @@ async fn accept_connection( } } -pub async fn start_websocket( +pub async fn start_websocket( address: SocketAddr, message_tx: mpsc::UnboundedSender, received_messages_query_tx: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, ) -> Result<(), WebSocketError> { let mut listener = tokio::net::TcpListener::bind(address).await?; From 54c244e4eacab88602c1ed1b1925e08643ff3400 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 16:55:59 +0000 Subject: [PATCH 09/13] Everything put together in mod.rs --- nym-client/src/client/mod.rs | 102 +++++++++++------------------------ 1 file changed, 32 insertions(+), 70 deletions(-) diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index c9685af0b9..01c12470da 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -1,6 +1,6 @@ -use crate::built_info; use crate::client::mix_traffic::MixTrafficController; use crate::client::received_buffer::ReceivedMessagesBuffer; +use crate::client::topology_control::TopologyInnerRef; use crate::sockets::tcp; use crate::sockets::ws; use directory_client::presence::Topology; @@ -18,6 +18,7 @@ mod mix_traffic; mod provider_poller; mod real_traffic_stream; pub mod received_buffer; +pub mod topology_control; // TODO: all of those constants should probably be moved to config file const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5; @@ -26,6 +27,8 @@ const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 0.5; // seconds; const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds; +const TOPOLOGY_REFRESH_RATE: f64 = 10.0; // seconds + pub enum SocketType { TCP, WebSocket, @@ -46,13 +49,6 @@ pub struct NymClient { socket_type: SocketType, } -// TODO: this will be moved into module responsible for refreshing topology -#[derive(Debug)] -enum TopologyError { - HealthCheckError, - NoValidPathsError, -} - #[derive(Debug)] pub struct InputMessage(pub Destination, pub Vec); @@ -77,50 +73,16 @@ impl NymClient { } } - // TODO: this will be moved into module responsible for refreshing topology - async fn get_compatible_topology(&self) -> Result { - let score_threshold = 0.0; - info!("Trying to obtain valid, healthy, topology"); - - let full_topology = Topology::new(self.directory.clone()); - - // run a healthcheck to determine healthy-ish nodes: - // this is a temporary solution as the healthcheck will eventually be moved to validators - let healthcheck_config = healthcheck::config::HealthCheck { - directory_server: self.directory.clone(), - // those are literally unrelevant when running single check - interval: 100000.0, - resolution_timeout: 5.0, - num_test_packets: 2, - }; - let healthcheck = healthcheck::HealthChecker::new(healthcheck_config); - let healthcheck_result = healthcheck.do_check().await; - - let healthcheck_scores = match healthcheck_result { - Err(err) => { - error!("Error while performing the healtcheck: {:?}", err); - return Err(TopologyError::HealthCheckError); - } - Ok(scores) => scores, - }; - - let healthy_topology = - healthcheck_scores.filter_topology_by_score(&full_topology, score_threshold); - - // for time being assume same versioning, i.e. if client is running X.Y.Z, - // we're expecting mixes, providers and coconodes to also be running X.Y.Z - let versioned_healthy_topology = healthy_topology.filter_node_versions( - built_info::PKG_VERSION, - built_info::PKG_VERSION, - built_info::PKG_VERSION, - ); - - // make sure you can still send a packet through the network: - if !versioned_healthy_topology.can_construct_path_through() { - return Err(TopologyError::NoValidPathsError); - } - - Ok(versioned_healthy_topology) + async fn get_provider_socket_address( + &self, + topology_ctrl_ref: TopologyInnerRef, + ) -> SocketAddr { + // this is temporary and assumes there exists only a single provider. + topology_ctrl_ref.read().await.topology.as_ref().unwrap() + .get_mix_provider_nodes() + .first() + .expect("Could not get a provider from the initial network topology, are you using the right directory server?") + .client_listener } pub fn start(self) -> Result<(), Box> { @@ -144,20 +106,14 @@ impl NymClient { let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) = mpsc::unbounded(); - // get initial topology; already filtered by health and version - let initial_topology = match rt.block_on(self.get_compatible_topology()) { - Ok(topology) => topology, - Err(err) => { - panic!("Failed to obtain initial network topology: {:?}", err); - } - }; + // TODO: when we switch to our graph topology, we need to remember to change 'Topology' type + let topology_controller = rt.block_on(topology_control::TopologyControl::::new( + self.directory.clone(), + TOPOLOGY_REFRESH_RATE, + )); - // this is temporary and assumes there exists only a single provider. - let provider_client_listener_address: SocketAddr = initial_topology - .get_mix_provider_nodes() - .first() - .expect("Could not get a provider from the supplied network topology, are you using the right directory server?") - .client_listener; + let provider_client_listener_address = + rt.block_on(self.get_provider_socket_address(topology_controller.get_inner_ref())); let mut provider_poller = provider_poller::ProviderPoller::new( poller_input_tx, @@ -189,13 +145,13 @@ impl NymClient { rt.spawn(cover_traffic_stream::start_loop_cover_traffic_stream( mix_tx.clone(), Destination::new(self.address, Default::default()), - initial_topology.clone(), + topology_controller.get_inner_ref(), )); // cloning arguments required by OutQueueControl; required due to move - let topology_clone = initial_topology.clone(); let self_address = self.address; let input_rx = self.input_rx; + let topology_ref = topology_controller.get_inner_ref(); // future constantly pumping traffic at some specified average rate // if a real message is available on 'input_rx' that might have been received from say @@ -206,7 +162,7 @@ impl NymClient { mix_tx, input_rx, Destination::new(self_address, Default::default()), - topology_clone, + topology_ref, ) .run_out_queue_control() .await @@ -225,7 +181,7 @@ impl NymClient { self.input_tx, received_messages_buffer_output_tx, self.address, - initial_topology, + topology_controller.get_inner_ref(), )); } SocketType::TCP => { @@ -234,12 +190,16 @@ impl NymClient { self.input_tx, received_messages_buffer_output_tx, self.address, - initial_topology, + topology_controller.get_inner_ref(), )); } SocketType::None => (), } + // future responsible for periodically polling directory server and updating + // the current global view of topology + let topology_refresher_future = rt.spawn(topology_controller.run_refresher()); + rt.block_on(async { let future_results = join!( received_messages_buffer_controllers_future, @@ -247,6 +207,7 @@ impl NymClient { loop_cover_traffic_future, out_queue_control_future, provider_polling_future, + topology_refresher_future, ); assert!( @@ -255,6 +216,7 @@ impl NymClient { && future_results.2.is_ok() && future_results.3.is_ok() && future_results.4.is_ok() + && future_results.5.is_ok() ); }); From c61b49c353a6c94831ac274f5f4503a51112f9b3 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 17:08:26 +0000 Subject: [PATCH 10/13] Corrected warnin message for out queue control --- nym-client/src/client/real_traffic_stream.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nym-client/src/client/real_traffic_stream.rs b/nym-client/src/client/real_traffic_stream.rs index 66a80a2ebe..9d1ee055ab 100644 --- a/nym-client/src/client/real_traffic_stream.rs +++ b/nym-client/src/client/real_traffic_stream.rs @@ -88,7 +88,7 @@ impl OutQueueControl { let topology = read_lock.topology.as_ref(); if topology.is_none() { - warn!("No valid topology detected - won't send any loop cover message this time"); + warn!("No valid topology detected - won't send any loop cover or real message this time"); continue; } From d9ac85abb2449a89f8b9ff1ee3ac507ca40bb704 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 17:17:28 +0000 Subject: [PATCH 11/13] Removed hardcoded version numbers --- nym-client/src/client/topology_control.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/nym-client/src/client/topology_control.rs b/nym-client/src/client/topology_control.rs index 8adf953dfb..6f749682dd 100644 --- a/nym-client/src/client/topology_control.rs +++ b/nym-client/src/client/topology_control.rs @@ -67,10 +67,11 @@ impl TopologyControl { let healthy_topology = healthcheck_scores.filter_topology_by_score(&full_topology, NODE_HEALTH_THRESHOLD); - // for time being assume same versioning, i.e. if client is running X.Y.Z, - // we're expecting mixes, providers and coconodes to also be running X.Y.Z - let versioned_healthy_topology = - healthy_topology.filter_node_versions("0.3.2", "0.3.2", built_info::PKG_VERSION); + let versioned_healthy_topology = healthy_topology.filter_node_versions( + built_info::PKG_VERSION, + built_info::PKG_VERSION, + built_info::PKG_VERSION, + ); // make sure you can still send a packet through the network: if !versioned_healthy_topology.can_construct_path_through() { From 4a2eab76e70746d60af40ef82cbf6e5ecd2a92b1 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 17:19:08 +0000 Subject: [PATCH 12/13] Decreased log severity for failing to send test packet during healthcheck --- common/healthcheck/src/path_check.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/healthcheck/src/path_check.rs b/common/healthcheck/src/path_check.rs index 62b94a86b1..42705b181d 100644 --- a/common/healthcheck/src/path_check.rs +++ b/common/healthcheck/src/path_check.rs @@ -223,7 +223,7 @@ impl PathChecker { debug!("sending test packet to {}", first_node_address); match first_node_client.send(packet, first_node_address).await { Err(err) => { - warn!("failed to send packet to {} - {}", first_node_address, err); + info!("failed to send packet to {} - {}", first_node_address, err); if self .paths_status .insert(path_identifier, PathStatus::Unhealthy) From 143d45904c872f655ff42435442c4499077cb340 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 23 Jan 2020 17:19:40 +0000 Subject: [PATCH 13/13] missing import --- common/healthcheck/src/path_check.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/healthcheck/src/path_check.rs b/common/healthcheck/src/path_check.rs index 42705b181d..89fe995880 100644 --- a/common/healthcheck/src/path_check.rs +++ b/common/healthcheck/src/path_check.rs @@ -1,6 +1,6 @@ use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey}; use itertools::Itertools; -use log::{debug, error, trace, warn}; +use log::{debug, error, info, trace, warn}; use mix_client::MixClient; use provider_client::ProviderClient; use sphinx::header::delays::Delay;