diff --git a/common/clients/directory-client/src/presence.rs b/common/clients/directory-client/src/presence.rs index 8be3c73b9e..8db97f673a 100644 --- a/common/clients/directory-client/src/presence.rs +++ b/common/clients/directory-client/src/presence.rs @@ -2,11 +2,18 @@ use crate::requests::presence_topology_get::PresenceTopologyGetRequester; use crate::{Client, Config, DirectoryClient}; use log::*; 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}; +// 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 { @@ -93,6 +100,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 { @@ -149,6 +182,100 @@ 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)] #[serde(rename_all = "camelCase")] @@ -158,6 +285,23 @@ 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 { debug!("Using directory server: {:?}", directory_server); diff --git a/common/healthcheck/src/path_check.rs b/common/healthcheck/src/path_check.rs index 62b94a86b1..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; @@ -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) diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 8fadc5c2ab..116920e7f0 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -84,7 +84,7 @@ pub enum NymTopologyError { MissingLayerError(Vec), } -pub trait NymTopology: Sized { +pub trait NymTopology: Sized + PartialEq + std::fmt::Debug + Send + Sync { fn new(directory_server: String) -> Self; fn new_from_nodes( mix_nodes: Vec, 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 diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index 8359716f71..cf0b609661 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() ); }); diff --git a/nym-client/src/client/real_traffic_stream.rs b/nym-client/src/client/real_traffic_stream.rs index f89257747b..9d1ee055ab 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 or real 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(); } } diff --git a/nym-client/src/client/topology_control.rs b/nym-client/src/client/topology_control.rs new file mode 100644 index 0000000000..6f749682dd --- /dev/null +++ b/nym-client/src/client/topology_control.rs @@ -0,0 +1,141 @@ +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); + + 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) + } + + 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 } + } +} diff --git a/nym-client/src/sockets/tcp.rs b/nym-client/src/sockets/tcp.rs index 709181a684..df02c5074a 100644 --- a/nym-client/src/sockets/tcp.rs +++ b/nym-client/src/sockets/tcp.rs @@ -1,18 +1,17 @@ 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 log::*; 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; @@ -124,14 +123,24 @@ impl ClientRequest { ServerResponse::Fetch { messages } } - async fn handle_get_clients(topology: &Topology) -> ServerResponse { - 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 { @@ -196,9 +205,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 { @@ -211,7 +220,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 @@ -221,27 +230,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"); debug!("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. @@ -280,12 +287,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 621c9ffe80..0a2807038e 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?;