diff --git a/Cargo.lock b/Cargo.lock index 9f2d2f7a14..210211462d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3776,6 +3776,7 @@ dependencies = [ "nym-network-defaults", "nym-sphinx", "nym-task", + "nym-topology", "pretty_env_logger", "rand 0.7.3", "tap", @@ -3983,6 +3984,7 @@ dependencies = [ name = "nym-topology" version = "0.1.0" dependencies = [ + "async-trait", "bs58", "log", "nym-bin-common", diff --git a/Cargo.toml b/Cargo.toml index 96ef6b17cd..19a77963c4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -105,7 +105,7 @@ edition = "2021" license = "Apache-2.0" [workspace.dependencies] -async-trait = "0.1.63" +async-trait = "0.1.64" cfg-if = "1.0.0" dotenv = "0.15.0" lazy_static = "1.4.0" diff --git a/clients/client-core/Cargo.toml b/clients/client-core/Cargo.toml index 4334cc66e4..1dfc530a59 100644 --- a/clients/client-core/Cargo.toml +++ b/clients/client-core/Cargo.toml @@ -8,7 +8,7 @@ rust-version = "1.66" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -async-trait = { version = "0.1.58" } +async-trait = { workspace = true } dirs = "4.0" dashmap = "5.4.0" futures = "0.3" diff --git a/clients/client-core/src/client/base_client/mod.rs b/clients/client-core/src/client/base_client/mod.rs index 64a292d2b8..e8a3d215f9 100644 --- a/clients/client-core/src/client/base_client/mod.rs +++ b/clients/client-core/src/client/base_client/mod.rs @@ -15,6 +15,7 @@ use crate::client::replies::reply_controller::{ReplyControllerReceiver, ReplyCon use crate::client::replies::reply_storage::{ CombinedReplyStorage, PersistentReplyStorage, ReplyStorageBackend, SentReplyKeys, }; +use crate::client::topology_control::nym_api_provider::NymApiTopologyProvider; use crate::client::topology_control::{ TopologyAccessor, TopologyRefresher, TopologyRefresherConfig, }; @@ -37,10 +38,12 @@ use nym_sphinx::addressing::nodes::NodeIdentity; use nym_sphinx::receiver::ReconstructedMessage; use nym_task::connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths}; use nym_task::{TaskClient, TaskManager}; +use nym_topology::provider_trait::TopologyProvider; use std::sync::Arc; use std::time::Duration; use tap::TapFallible; use url::Url; + #[cfg(not(target_arch = "wasm32"))] use validator_client::nyxd::CosmWasmClient; @@ -90,6 +93,7 @@ impl ClientOutput { pub struct ClientState { pub shared_lane_queue_lengths: LaneQueueLengths, pub reply_controller_sender: ReplyControllerSender, + pub topology_accessor: TopologyAccessor, } pub enum ClientInputStatus { @@ -154,6 +158,7 @@ pub struct BaseClientBuilder<'a, B, C: Clone> { nym_api_endpoints: Vec, reply_storage_backend: B, + custom_topology_provider: Option>, bandwidth_controller: Option>, key_manager: KeyManager, } @@ -177,6 +182,7 @@ where bandwidth_controller, reply_storage_backend, key_manager, + custom_topology_provider: None, } } @@ -195,11 +201,17 @@ where disabled_credentials: credentials_toggle.is_disabled(), nym_api_endpoints, reply_storage_backend, + custom_topology_provider: None, bandwidth_controller, key_manager, } } + pub fn with_topology_provider(mut self, provider: Box) -> Self { + self.custom_topology_provider = Some(provider); + self + } + pub fn as_mix_recipient(&self) -> Recipient { Recipient::new( *self.key_manager.identity_keypair().public_key(), @@ -341,25 +353,38 @@ where Ok(gateway_client) } + fn setup_topology_provider( + custom_provider: Option>, + nym_api_urls: Vec, + ) -> Box { + // if no custom provider was ... provided ..., create one using nym-api + custom_provider.unwrap_or_else(|| { + Box::new(NymApiTopologyProvider::new( + nym_api_urls, + env!("CARGO_PKG_VERSION").to_string(), + )) + }) + } + // future responsible for periodically polling directory server and updating // the current global view of topology async fn start_topology_refresher( - nym_api_urls: Vec, + topology_provider: Box, refresh_rate: Duration, topology_accessor: TopologyAccessor, shutdown: TaskClient, ) -> Result<(), ClientCoreError> { - let topology_refresher_config = TopologyRefresherConfig::new( - nym_api_urls, - refresh_rate, - env!("CARGO_PKG_VERSION").to_string(), + let topology_refresher_config = TopologyRefresherConfig::new(refresh_rate); + + let mut topology_refresher = TopologyRefresher::new( + topology_refresher_config, + topology_accessor, + topology_provider, ); - let mut topology_refresher = - TopologyRefresher::new(topology_refresher_config, topology_accessor); // before returning, block entire runtime to refresh the current network view so that any // components depending on topology would see a non-empty view info!("Obtaining initial network topology"); - topology_refresher.refresh().await; + topology_refresher.try_refresh().await; if let Err(err) = topology_refresher.ensure_topology_is_routable().await { log::error!( @@ -468,8 +493,12 @@ where ) .await?; + let topology_provider = Self::setup_topology_provider( + self.custom_topology_provider.take(), + self.nym_api_endpoints, + ); Self::start_topology_refresher( - self.nym_api_endpoints.clone(), + topology_provider, self.debug_config.topology_refresh_rate, shared_topology_accessor.clone(), task_manager.subscribe(), @@ -530,7 +559,7 @@ where self.debug_config, self.key_manager.ack_key(), self_address, - shared_topology_accessor, + shared_topology_accessor.clone(), sphinx_message_sender, task_manager.subscribe(), ); @@ -554,6 +583,7 @@ where client_state: ClientState { shared_lane_queue_lengths, reply_controller_sender, + topology_accessor: shared_topology_accessor, }, task_manager, }) diff --git a/clients/client-core/src/client/topology_control.rs b/clients/client-core/src/client/topology_control.rs deleted file mode 100644 index e94340f074..0000000000 --- a/clients/client-core/src/client/topology_control.rs +++ /dev/null @@ -1,336 +0,0 @@ -// Copyright 2021-2022 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::spawn_future; -use futures::StreamExt; -use log::*; -use nym_sphinx::addressing::clients::Recipient; -use nym_sphinx::params::DEFAULT_NUM_MIX_HOPS; -use nym_topology::{nym_topology_from_detailed, NymTopology, NymTopologyError}; -use rand::seq::SliceRandom; -use rand::thread_rng; -use std::ops::Deref; -use std::sync::Arc; -use std::time::Duration; -use tokio::sync::{RwLock, RwLockReadGuard}; -use url::Url; - -// I'm extremely curious why compiler NEVER complained about lack of Debug here before -#[derive(Debug)] -pub struct TopologyAccessorInner(Option); - -impl AsRef> for TopologyAccessorInner { - fn as_ref(&self) -> &Option { - &self.0 - } -} - -impl TopologyAccessorInner { - fn new() -> Self { - TopologyAccessorInner(None) - } - - fn update(&mut self, new: Option) { - self.0 = new; - } -} - -pub struct TopologyReadPermit<'a> { - permit: RwLockReadGuard<'a, TopologyAccessorInner>, -} - -impl<'a> Deref for TopologyReadPermit<'a> { - type Target = TopologyAccessorInner; - - fn deref(&self) -> &Self::Target { - &self.permit - } -} - -impl<'a> TopologyReadPermit<'a> { - /// Using provided topology read permit, tries to get an immutable reference to the underlying - /// topology. For obvious reasons the lifetime of the topology reference is bound to the permit. - pub(super) fn try_get_valid_topology_ref( - &'a self, - ack_recipient: &Recipient, - packet_recipient: Option<&Recipient>, - ) -> Result<&'a NymTopology, NymTopologyError> { - // 1. Have we managed to get anything from the refresher, i.e. have the nym-api queries gone through? - let topology = self - .permit - .as_ref() - .as_ref() - .ok_or(NymTopologyError::EmptyNetworkTopology)?; - - // 2. does it have any mixnode at all? - // 3. does it have any gateways at all? - // 4. does it have a mixnode on each layer? - topology.ensure_can_construct_path_through(DEFAULT_NUM_MIX_HOPS)?; - - // 5. does it contain OUR gateway (so that we could create an ack packet)? - if !topology.gateway_exists(ack_recipient.gateway()) { - return Err(NymTopologyError::NonExistentGatewayError { - identity_key: ack_recipient.gateway().to_base58_string(), - }); - } - - // 6. for our target recipient, does it contain THEIR gateway (so that we could create - if let Some(recipient) = packet_recipient { - if !topology.gateway_exists(recipient.gateway()) { - return Err(NymTopologyError::NonExistentGatewayError { - identity_key: recipient.gateway().to_base58_string(), - }); - } - } - - Ok(topology) - } -} - -impl<'a> From> for TopologyReadPermit<'a> { - fn from(read_permit: RwLockReadGuard<'a, TopologyAccessorInner>) -> Self { - TopologyReadPermit { - permit: read_permit, - } - } -} - -#[derive(Clone, Debug)] -pub struct TopologyAccessor { - // `RwLock` *seems to* be the better approach for this as write access is only requested every - // few seconds, while reads are needed every single packet generated. - // However, proper benchmarks will be needed to determine if `RwLock` is indeed a better - // approach than a `Mutex` - inner: Arc>, -} - -impl TopologyAccessor { - pub fn new() -> Self { - TopologyAccessor { - inner: Arc::new(RwLock::new(TopologyAccessorInner::new())), - } - } - - pub async fn get_read_permit(&self) -> TopologyReadPermit<'_> { - self.inner.read().await.into() - } - - async fn update_global_topology(&self, new_topology: Option) { - self.inner.write().await.update(new_topology); - } - - // only used by the client at startup to get a slightly more reasonable error message - // (currently displays as unused because health checker is disabled due to required changes) - pub async fn ensure_is_routable(&self) -> Result<(), NymTopologyError> { - match &self.inner.read().await.0 { - None => Err(NymTopologyError::EmptyNetworkTopology), - Some(ref topology) => topology.ensure_can_construct_path_through(DEFAULT_NUM_MIX_HOPS), - } - } -} - -impl Default for TopologyAccessor { - fn default() -> Self { - TopologyAccessor::new() - } -} - -pub struct TopologyRefresherConfig { - nym_api_urls: Vec, - refresh_rate: Duration, - client_version: String, -} - -impl TopologyRefresherConfig { - pub fn new(nym_api_urls: Vec, refresh_rate: Duration, client_version: String) -> Self { - TopologyRefresherConfig { - nym_api_urls, - refresh_rate, - client_version, - } - } -} - -pub struct TopologyRefresher { - validator_client: validator_client::client::NymApiClient, - client_version: String, - - nym_api_urls: Vec, - topology_accessor: TopologyAccessor, - refresh_rate: Duration, - - currently_used_api: usize, - was_latest_valid: bool, -} - -impl TopologyRefresher { - pub fn new(mut cfg: TopologyRefresherConfig, topology_accessor: TopologyAccessor) -> Self { - cfg.nym_api_urls.shuffle(&mut thread_rng()); - - TopologyRefresher { - validator_client: validator_client::client::NymApiClient::new( - cfg.nym_api_urls[0].clone(), - ), - client_version: cfg.client_version, - nym_api_urls: cfg.nym_api_urls, - topology_accessor, - refresh_rate: cfg.refresh_rate, - currently_used_api: 0, - was_latest_valid: true, - } - } - - fn use_next_nym_api(&mut self) { - if self.nym_api_urls.len() == 1 { - warn!("There's only a single nym API available - it won't be possible to use a different one"); - return; - } - - self.currently_used_api = (self.currently_used_api + 1) % self.nym_api_urls.len(); - self.validator_client - .change_nym_api(self.nym_api_urls[self.currently_used_api].clone()) - } - - /// Verifies whether nodes a reasonably distributed among all mix layers. - /// - /// In ideal world we would have 33% nodes on layer 1, 33% on layer 2 and 33% on layer 3. - /// However, this is a rather unrealistic expectation, instead we check whether there exists - /// a layer with more than 66% of nodes or with fewer than 15% and if so, we trigger a failure. - /// - /// # Arguments - /// - /// * `topology`: active topology constructed from validator api data - fn check_layer_distribution(&self, active_topology: &NymTopology) -> bool { - let mixes = active_topology.mixes(); - let mixnodes_count = active_topology.num_mixnodes(); - - if active_topology.gateways().is_empty() { - return false; - } - - // trivial check to see if have at least a single node on each layer (regardless of active set size) - if mixes.get(&1).is_none() || mixes.get(&2).is_none() || mixes.get(&3).is_none() { - return false; - } - - let upper_bound = (mixnodes_count as f32 * 0.66) as usize; - let lower_bound = (mixnodes_count as f32 * 0.15) as usize; - - let layer1 = mixes.get(&1).unwrap().len(); - let layer2 = mixes.get(&2).unwrap().len(); - let layer3 = mixes.get(&3).unwrap().len(); - - if layer1 < lower_bound || layer1 > upper_bound { - warn!( - "nodes: {}, layer1: {}, layer2: {}, layer3: {}", - mixnodes_count, layer1, layer2, layer3 - ); - return false; - } - - if layer2 < lower_bound || layer2 > upper_bound { - warn!( - "nodes: {}, layer1: {}, layer2: {}, layer3: {}", - mixnodes_count, layer1, layer2, layer3 - ); - return false; - } - - if layer3 < lower_bound || layer3 > upper_bound { - warn!( - "nodes: {}, layer1: {}, layer2: {}, layer3: {}", - mixnodes_count, layer1, layer2, layer3 - ); - return false; - } - - true - } - - async fn get_current_compatible_topology(&self) -> Option { - // TODO: optimization for the future: - // only refresh mixnodes on timer and refresh gateways only when - // we have to send to a new, unknown, gateway - - let mixnodes = match self.validator_client.get_cached_active_mixnodes().await { - Err(err) => { - error!("failed to get network mixnodes - {err}"); - return None; - } - Ok(mixes) => mixes, - }; - - let gateways = match self.validator_client.get_cached_gateways().await { - Err(err) => { - error!("failed to get network gateways - {err}"); - return None; - } - Ok(gateways) => gateways, - }; - - let topology = nym_topology_from_detailed(mixnodes, gateways) - .filter_system_version(&self.client_version); - - if !self.check_layer_distribution(&topology) { - warn!("The current filtered active topology has extremely skewed layer distribution. It cannot be used."); - None - } else { - Some(topology) - } - } - - pub async fn refresh(&mut self) { - trace!("Refreshing the topology"); - let new_topology = self.get_current_compatible_topology().await; - - if new_topology.is_none() { - self.use_next_nym_api(); - } - - if new_topology.is_none() && self.was_latest_valid { - // if we failed to grab this topology, but the one before it was alright, let's assume - // validator had a tiny hiccup and use the old data - warn!("we're going to keep on using the old topology for this iteration"); - self.was_latest_valid = false; - return; - } else if new_topology.is_some() { - self.was_latest_valid = true; - } - - self.topology_accessor - .update_global_topology(new_topology) - .await; - } - - pub async fn ensure_topology_is_routable(&self) -> Result<(), NymTopologyError> { - self.topology_accessor.ensure_is_routable().await - } - - pub fn start_with_shutdown(mut self, mut shutdown: nym_task::TaskClient) { - spawn_future(async move { - debug!("Started TopologyRefresher with graceful shutdown support"); - - #[cfg(not(target_arch = "wasm32"))] - let mut interval = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval( - self.refresh_rate, - )); - - #[cfg(target_arch = "wasm32")] - let mut interval = - gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32); - - while !shutdown.is_shutdown() { - tokio::select! { - _ = interval.next() => { - self.refresh().await; - }, - _ = shutdown.recv() => { - log::trace!("TopologyRefresher: Received shutdown"); - }, - } - } - shutdown.recv_timeout().await; - log::debug!("TopologyRefresher: Exiting"); - }) - } -} diff --git a/clients/client-core/src/client/topology_control/accessor.rs b/clients/client-core/src/client/topology_control/accessor.rs new file mode 100644 index 0000000000..fcb272c934 --- /dev/null +++ b/clients/client-core/src/client/topology_control/accessor.rs @@ -0,0 +1,154 @@ +// Copyright 2021-2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_sphinx::addressing::clients::Recipient; +use nym_sphinx::params::DEFAULT_NUM_MIX_HOPS; +use nym_topology::{NymTopology, NymTopologyError}; +use std::ops::Deref; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use tokio::sync::{Notify, RwLock, RwLockReadGuard}; + +#[derive(Debug)] +pub struct TopologyAccessorInner { + controlled_manually: AtomicBool, + released_manual_control: Notify, + // `RwLock` *seems to* be the better approach for this as write access is only requested every + // few seconds, while reads are needed every single packet generated. + // However, proper benchmarks will be needed to determine if `RwLock` is indeed a better + // approach than a `Mutex` + topology: RwLock>, +} + +impl TopologyAccessorInner { + fn new() -> Self { + TopologyAccessorInner { + controlled_manually: AtomicBool::new(false), + released_manual_control: Notify::new(), + topology: RwLock::new(None), + } + } + + async fn update(&self, new: Option) { + *self.topology.write().await = new; + } +} + +pub struct TopologyReadPermit<'a> { + permit: RwLockReadGuard<'a, Option>, +} + +impl<'a> Deref for TopologyReadPermit<'a> { + type Target = Option; + + fn deref(&self) -> &Self::Target { + &self.permit + } +} + +impl<'a> TopologyReadPermit<'a> { + /// Using provided topology read permit, tries to get an immutable reference to the underlying + /// topology. For obvious reasons the lifetime of the topology reference is bound to the permit. + pub(crate) fn try_get_valid_topology_ref( + &'a self, + ack_recipient: &Recipient, + packet_recipient: Option<&Recipient>, + ) -> Result<&'a NymTopology, NymTopologyError> { + // 1. Have we managed to get anything from the refresher, i.e. have the nym-api queries gone through? + let topology = self + .permit + .as_ref() + .ok_or(NymTopologyError::EmptyNetworkTopology)?; + + // 2. does it have any mixnode at all? + // 3. does it have any gateways at all? + // 4. does it have a mixnode on each layer? + topology.ensure_can_construct_path_through(DEFAULT_NUM_MIX_HOPS)?; + + // 5. does it contain OUR gateway (so that we could create an ack packet)? + if !topology.gateway_exists(ack_recipient.gateway()) { + return Err(NymTopologyError::NonExistentGatewayError { + identity_key: ack_recipient.gateway().to_base58_string(), + }); + } + + // 6. for our target recipient, does it contain THEIR gateway (so that we could create + if let Some(recipient) = packet_recipient { + if !topology.gateway_exists(recipient.gateway()) { + return Err(NymTopologyError::NonExistentGatewayError { + identity_key: recipient.gateway().to_base58_string(), + }); + } + } + + Ok(topology) + } +} + +impl<'a> From>> for TopologyReadPermit<'a> { + fn from(read_permit: RwLockReadGuard<'a, Option>) -> Self { + TopologyReadPermit { + permit: read_permit, + } + } +} + +#[derive(Clone, Debug)] +pub struct TopologyAccessor { + inner: Arc, +} + +impl TopologyAccessor { + pub fn new() -> Self { + TopologyAccessor { + inner: Arc::new(TopologyAccessorInner::new()), + } + } + + pub fn controlled_manually(&self) -> bool { + self.inner.controlled_manually.load(Ordering::SeqCst) + } + + pub async fn get_read_permit(&self) -> TopologyReadPermit<'_> { + self.inner.topology.read().await.into() + } + + pub(crate) async fn update_global_topology(&self, new_topology: Option) { + self.inner.update(new_topology).await; + } + + pub(crate) async fn wait_for_released_manual_control(&self) { + self.inner.released_manual_control.notified().await + } + + pub async fn current_topology(&self) -> Option { + self.inner.topology.read().await.clone() + } + + pub async fn manually_change_topology(&self, new_topology: NymTopology) { + self.inner.controlled_manually.store(true, Ordering::SeqCst); + self.inner.update(Some(new_topology)).await; + } + + pub fn release_manual_control(&self) { + self.inner + .controlled_manually + .store(false, Ordering::SeqCst); + self.inner.released_manual_control.notify_waiters(); + } + + // only used by the client at startup to get a slightly more reasonable error message + // (currently displays as unused because health checker is disabled due to required changes) + pub async fn ensure_is_routable(&self) -> Result<(), NymTopologyError> { + match self.inner.topology.read().await.deref() { + None => Err(NymTopologyError::EmptyNetworkTopology), + Some(ref topology) => topology.ensure_can_construct_path_through(DEFAULT_NUM_MIX_HOPS), + } + } +} + +impl Default for TopologyAccessor { + fn default() -> Self { + TopologyAccessor::new() + } +} diff --git a/clients/client-core/src/client/topology_control/mod.rs b/clients/client-core/src/client/topology_control/mod.rs new file mode 100644 index 0000000000..168767cd3e --- /dev/null +++ b/clients/client-core/src/client/topology_control/mod.rs @@ -0,0 +1,115 @@ +// Copyright 2021-2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::spawn_future; +pub(crate) use accessor::{TopologyAccessor, TopologyReadPermit}; +use futures::StreamExt; +use log::*; +use nym_topology::provider_trait::TopologyProvider; +use nym_topology::NymTopologyError; +use std::time::Duration; + +mod accessor; +pub(crate) mod nym_api_provider; + +// TODO: move it to config later +const MAX_FAILURE_COUNT: usize = 10; + +pub struct TopologyRefresherConfig { + refresh_rate: Duration, +} + +impl TopologyRefresherConfig { + pub fn new(refresh_rate: Duration) -> Self { + TopologyRefresherConfig { refresh_rate } + } +} + +pub struct TopologyRefresher { + topology_provider: Box, + topology_accessor: TopologyAccessor, + + refresh_rate: Duration, + consecutive_failure_count: usize, +} + +impl TopologyRefresher { + pub fn new( + cfg: TopologyRefresherConfig, + topology_accessor: TopologyAccessor, + topology_provider: Box, + ) -> Self { + TopologyRefresher { + topology_provider, + topology_accessor, + refresh_rate: cfg.refresh_rate, + consecutive_failure_count: 0, + } + } + + pub fn change_topology_provider(&mut self, provider: Box) { + self.topology_provider = provider; + } + + pub async fn try_refresh(&mut self) { + trace!("Refreshing the topology"); + + if self.topology_accessor.controlled_manually() { + info!("topology is being controlled manually - we're going to wait until the control is released..."); + self.topology_accessor + .wait_for_released_manual_control() + .await; + } + + let new_topology = self.topology_provider.get_new_topology().await; + if new_topology.is_none() { + warn!("failed to obtain new network topology"); + } + + if new_topology.is_none() && self.consecutive_failure_count < MAX_FAILURE_COUNT { + // if we failed to grab this topology, but the one before it was alright, let's assume + // validator had a tiny hiccup and use the old data + warn!("we're going to keep on using the old topology for this iteration"); + self.consecutive_failure_count += 1; + return; + } else if new_topology.is_some() { + self.consecutive_failure_count = 0; + } + + self.topology_accessor + .update_global_topology(new_topology) + .await; + } + + pub async fn ensure_topology_is_routable(&self) -> Result<(), NymTopologyError> { + self.topology_accessor.ensure_is_routable().await + } + + pub fn start_with_shutdown(mut self, mut shutdown: nym_task::TaskClient) { + spawn_future(async move { + debug!("Started TopologyRefresher with graceful shutdown support"); + + #[cfg(not(target_arch = "wasm32"))] + let mut interval = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval( + self.refresh_rate, + )); + + #[cfg(target_arch = "wasm32")] + let mut interval = + gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32); + + while !shutdown.is_shutdown() { + tokio::select! { + _ = interval.next() => { + self.try_refresh().await; + }, + _ = shutdown.recv() => { + log::trace!("TopologyRefresher: Received shutdown"); + }, + } + } + shutdown.recv_timeout().await; + log::debug!("TopologyRefresher: Exiting"); + }) + } +} diff --git a/clients/client-core/src/client/topology_control/nym_api_provider.rs b/clients/client-core/src/client/topology_control/nym_api_provider.rs new file mode 100644 index 0000000000..159387d921 --- /dev/null +++ b/clients/client-core/src/client/topology_control/nym_api_provider.rs @@ -0,0 +1,106 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use async_trait::async_trait; +use log::{error, warn}; +use nym_topology::provider_trait::TopologyProvider; +use nym_topology::{nym_topology_from_detailed, NymTopology, NymTopologyError}; +use rand::prelude::SliceRandom; +use rand::thread_rng; +use url::Url; + +pub(crate) struct NymApiTopologyProvider { + validator_client: validator_client::client::NymApiClient, + nym_api_urls: Vec, + + client_version: String, + currently_used_api: usize, +} + +impl NymApiTopologyProvider { + pub(crate) fn new(mut nym_api_urls: Vec, client_version: String) -> Self { + nym_api_urls.shuffle(&mut thread_rng()); + + NymApiTopologyProvider { + validator_client: validator_client::client::NymApiClient::new(nym_api_urls[0].clone()), + nym_api_urls, + client_version, + currently_used_api: 0, + } + } + + fn use_next_nym_api(&mut self) { + if self.nym_api_urls.len() == 1 { + warn!("There's only a single nym API available - it won't be possible to use a different one"); + return; + } + + self.currently_used_api = (self.currently_used_api + 1) % self.nym_api_urls.len(); + self.validator_client + .change_nym_api(self.nym_api_urls[self.currently_used_api].clone()) + } + + /// Verifies whether nodes a reasonably distributed among all mix layers. + /// + /// In ideal world we would have 33% nodes on layer 1, 33% on layer 2 and 33% on layer 3. + /// However, this is a rather unrealistic expectation, instead we check whether there exists + /// a layer with more than 66% of nodes or with fewer than 15% and if so, we trigger a failure. + /// + /// # Arguments + /// + /// * `topology`: active topology constructed from validator api data + fn check_layer_distribution( + &self, + active_topology: &NymTopology, + ) -> Result<(), NymTopologyError> { + let lower_threshold = 0.15; + let upper_threshold = 0.66; + active_topology.ensure_even_layer_distribution(lower_threshold, upper_threshold) + } + + async fn get_current_compatible_topology(&mut self) -> Option { + let mixnodes = match self.validator_client.get_cached_active_mixnodes().await { + Err(err) => { + error!("failed to get network mixnodes - {err}"); + return None; + } + Ok(mixes) => mixes, + }; + + let gateways = match self.validator_client.get_cached_gateways().await { + Err(err) => { + error!("failed to get network gateways - {err}"); + return None; + } + Ok(gateways) => gateways, + }; + + let topology = nym_topology_from_detailed(mixnodes, gateways) + .filter_system_version(&self.client_version); + + if let Err(err) = self.check_layer_distribution(&topology) { + warn!("The current filtered active topology has extremely skewed layer distribution. It cannot be used: {err}"); + self.use_next_nym_api(); + None + } else { + Some(topology) + } + } +} + +// hehe, wasm +#[cfg(not(target_arch = "wasm32"))] +#[async_trait] +impl TopologyProvider for NymApiTopologyProvider { + async fn get_new_topology(&mut self) -> Option { + self.get_current_compatible_topology().await + } +} + +#[cfg(target_arch = "wasm32")] +#[async_trait(?Send)] +impl TopologyProvider for NymApiTopologyProvider { + async fn get_new_topology(&mut self) -> Option { + self.get_current_compatible_topology().await + } +} diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index 1e4e17d94c..796435fb5f 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -101,6 +101,7 @@ impl SocketClient { let ClientState { shared_lane_queue_lengths, reply_controller_sender, + .. } = client_state; let websocket_handler = websocket::HandlerBuilder::new( diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index 99a78c86b8..5a6454d361 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -123,7 +123,7 @@ impl NymClient { let ClientState { shared_lane_queue_lengths, - reply_controller_sender: _, + .. } = client_status; let authenticator = Authenticator::new(auth_methods, allowed_users); diff --git a/common/client-libs/gateway-client/Cargo.toml b/common/client-libs/gateway-client/Cargo.toml index 195d7df616..a9867755d8 100644 --- a/common/client-libs/gateway-client/Cargo.toml +++ b/common/client-libs/gateway-client/Cargo.toml @@ -14,7 +14,7 @@ log = { workspace = true } thiserror = "1.0" url = "2.2" rand = { version = "0.7.3", features = ["wasm-bindgen"] } -async-trait = { version = "0.1.51" } +async-trait = { workspace = true } tokio = { version = "1.24.1", features = ["macros"] } # internal diff --git a/common/client-libs/validator-client/Cargo.toml b/common/client-libs/validator-client/Cargo.toml index 379cf546b0..a5f95b9508 100644 --- a/common/client-libs/validator-client/Cargo.toml +++ b/common/client-libs/validator-client/Cargo.toml @@ -35,7 +35,7 @@ nym-api-requests = { path = "../../../nym-api/nym-api-requests" } # required for nyxd-client # at some point it might be possible to make it wasm-compatible # perhaps after https://github.com/cosmos/cosmos-rust/pull/97 is resolved (and tendermint-rs is updated) -async-trait = { version = "0.1.51", optional = true } +async-trait = { workspace = true, optional = true } bip39 = { version = "1", features = ["rand"], optional = true } nym-config = { path = "../../config", optional = true } cosmrs = { git = "https://github.com/neacsu/cosmos-rust", branch = "neacsu/feegrant_support", features = ["rpc", "bip32", "cosmwasm"], optional = true} diff --git a/common/credential-storage/Cargo.toml b/common/credential-storage/Cargo.toml index 0c10ef7535..ec828bce97 100644 --- a/common/credential-storage/Cargo.toml +++ b/common/credential-storage/Cargo.toml @@ -6,7 +6,7 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -async-trait = { version = "0.1.51" } +async-trait = { workspace = true } log = { workspace = true } sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"]} diff --git a/common/crypto/src/asymmetric/encryption/mod.rs b/common/crypto/src/asymmetric/encryption/mod.rs index e81bea6e85..4d079d4378 100644 --- a/common/crypto/src/asymmetric/encryption/mod.rs +++ b/common/crypto/src/asymmetric/encryption/mod.rs @@ -3,6 +3,7 @@ use nym_pemstore::traits::{PemStorableKey, PemStorableKeyPair}; use std::fmt::{self, Display, Formatter}; +use std::str::FromStr; use thiserror::Error; #[cfg(feature = "rand")] @@ -131,6 +132,14 @@ impl PublicKey { } } +impl FromStr for PublicKey { + type Err = KeyRecoveryError; + + fn from_str(s: &str) -> Result { + PublicKey::from_base58_string(s) + } +} + #[cfg(feature = "serde")] impl Serialize for PublicKey { fn serialize(&self, serializer: S) -> Result diff --git a/common/mobile-storage/Cargo.toml b/common/mobile-storage/Cargo.toml index 3c38de8d2d..98b9d9b42f 100644 --- a/common/mobile-storage/Cargo.toml +++ b/common/mobile-storage/Cargo.toml @@ -6,6 +6,6 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -async-trait = { version = "0.1.51" } +async-trait = { workspace = true } thiserror = "1.0" diff --git a/common/statistics/Cargo.toml b/common/statistics/Cargo.toml index 5649e22414..29573f1e06 100644 --- a/common/statistics/Cargo.toml +++ b/common/statistics/Cargo.toml @@ -9,7 +9,7 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -async-trait = { version = "0.1.51" } +async-trait = { workspace = true } log = { workspace = true } reqwest = { version = "0.11", features = ["json"] } serde = { version = "1.0", features = ["derive"] } diff --git a/common/topology/Cargo.toml b/common/topology/Cargo.toml index 70c8e2db9a..fbb5985ad5 100644 --- a/common/topology/Cargo.toml +++ b/common/topology/Cargo.toml @@ -16,6 +16,7 @@ bs58 = "0.4" log = { workspace = true } rand = { version = "0.7.3", features = ["wasm-bindgen"] } thiserror = "1.0.37" +async-trait = { workspace = true, optional = true } ## internal nym-crypto = { path = "../crypto" } @@ -23,3 +24,7 @@ nym-mixnet-contract-common = { path = "../cosmwasm-smart-contracts/mixnet-contra nym-sphinx-addressing = { path = "../nymsphinx/addressing" } nym-sphinx-types = { path = "../nymsphinx/types" } nym-bin-common = { path = "../bin-common" } + +[features] +default = ["provider-trait"] +provider-trait = ["async-trait"] \ No newline at end of file diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 54e41199e0..6824f6c0d7 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -20,6 +20,9 @@ pub mod filter; pub mod gateway; pub mod mix; +#[cfg(feature = "provider-trait")] +pub mod provider_trait; + #[derive(Debug, Clone, Error)] pub enum NymTopologyError { #[error("The provided network topology is empty - there are no mixnodes and no gateways on it - the network request(s) probably failed")] @@ -39,6 +42,16 @@ pub enum NymTopologyError { #[error("No mixnodes available on layer {layer}")] EmptyMixLayer { layer: MixLayer }, + + #[error("Uneven layer distribution. Layer {layer} has {nodes} on it, while we expected a value between {lower_bound} and {upper_bound} as we have {total_nodes} nodes in total. Full breakdown: {layer_distribution:?}")] + UnevenLayerDistribution { + layer: MixLayer, + nodes: usize, + lower_bound: usize, + upper_bound: usize, + total_nodes: usize, + layer_distribution: Vec<(MixLayer, usize)>, + }, } #[derive(Debug, Clone)] @@ -97,7 +110,7 @@ impl NymTopology { } pub fn num_mixnodes(&self) -> usize { - self.mixes.values().flat_map(|m| m.iter()).count() + self.mixes.values().map(|m| m.len()).sum() } pub fn mixes_as_vec(&self) -> Vec { @@ -238,6 +251,46 @@ impl NymTopology { Ok(()) } + pub fn ensure_even_layer_distribution( + &self, + lower_threshold: f32, + upper_threshold: f32, + ) -> Result<(), NymTopologyError> { + let mixnodes_count = self.num_mixnodes(); + + let layers = self + .mixes + .iter() + .map(|(k, v)| (*k, v.len())) + .collect::>(); + + if self.gateways.is_empty() { + return Err(NymTopologyError::NoGatewaysAvailable); + } + + if layers.is_empty() { + return Err(NymTopologyError::NoMixnodesAvailable); + } + + let upper_bound = (mixnodes_count as f32 * upper_threshold) as usize; + let lower_bound = (mixnodes_count as f32 * lower_threshold) as usize; + + for (layer, nodes) in &layers { + if nodes < &lower_bound || nodes > &upper_bound { + return Err(NymTopologyError::UnevenLayerDistribution { + layer: *layer, + nodes: *nodes, + lower_bound, + upper_bound, + total_nodes: mixnodes_count, + layer_distribution: layers, + }); + } + } + + Ok(()) + } + #[must_use] pub fn filter_system_version(&self, expected_version: &str) -> Self { self.filter_node_versions(expected_version) diff --git a/common/topology/src/mix.rs b/common/topology/src/mix.rs index c03e9d9978..14dcc02857 100644 --- a/common/topology/src/mix.rs +++ b/common/topology/src/mix.rs @@ -3,7 +3,8 @@ use crate::{filter, NetworkAddress}; use nym_crypto::asymmetric::{encryption, identity}; -use nym_mixnet_contract_common::{Layer, MixId, MixNodeBond}; +pub use nym_mixnet_contract_common::Layer; +use nym_mixnet_contract_common::{MixId, MixNodeBond}; use nym_sphinx_addressing::nodes::NymNodeRoutingAddress; use nym_sphinx_types::Node as SphinxNode; use std::convert::{TryFrom, TryInto}; diff --git a/common/topology/src/provider_trait.rs b/common/topology/src/provider_trait.rs new file mode 100644 index 0000000000..562cfaa5f7 --- /dev/null +++ b/common/topology/src/provider_trait.rs @@ -0,0 +1,44 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::NymTopology; +pub use async_trait::async_trait; + +// hehe, wasm +#[cfg(not(target_arch = "wasm32"))] +#[async_trait] +pub trait TopologyProvider: Send { + async fn get_new_topology(&mut self) -> Option; +} + +#[cfg(target_arch = "wasm32")] +#[async_trait(?Send)] +pub trait TopologyProvider { + async fn get_new_topology(&mut self) -> Option; +} + +pub struct HardcodedTopologyProvider { + topology: NymTopology, +} + +impl HardcodedTopologyProvider { + pub fn new(topology: NymTopology) -> Self { + HardcodedTopologyProvider { topology } + } +} + +#[cfg(not(target_arch = "wasm32"))] +#[async_trait] +impl TopologyProvider for HardcodedTopologyProvider { + async fn get_new_topology(&mut self) -> Option { + Some(self.topology.clone()) + } +} + +#[cfg(target_arch = "wasm32")] +#[async_trait(?Send)] +impl TopologyProvider for HardcodedTopologyProvider { + async fn get_new_topology(&mut self) -> Option { + Some(self.topology.clone()) + } +} diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index b3d975adbc..d7c72244c2 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -16,7 +16,7 @@ rust-version = "1.56" [dependencies] anyhow = "1.0.53" -async-trait = { version = "0.1.51" } +async-trait = { workspace = true } bip39 = "1.0.1" bs58 = "0.4.0" clap = { version = "4.0", features = ["cargo", "derive"] } diff --git a/nym-api/Cargo.toml b/nym-api/Cargo.toml index 5378923d39..f785280d3e 100644 --- a/nym-api/Cargo.toml +++ b/nym-api/Cargo.toml @@ -15,7 +15,7 @@ rust-version = "1.56" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -async-trait = "0.1.52" +async-trait = { workspace = true } bs58 = {version = "0.4.0" } bip39 = "1" cfg-if = "1.0" diff --git a/nym-api/src/network_monitor/monitor/mod.rs b/nym-api/src/network_monitor/monitor/mod.rs index 5fa7ef0793..3dc98ccb9d 100644 --- a/nym-api/src/network_monitor/monitor/mod.rs +++ b/nym-api/src/network_monitor/monitor/mod.rs @@ -88,10 +88,7 @@ impl Monitor { ) .await { - error!( - "Failed to submit monitor run information to the database - {}", - err - ); + error!("Failed to submit monitor run information to the database - {err}",); // TODO: slightly more graceful shutdown here process::exit(1); @@ -124,12 +121,8 @@ impl Monitor { for route in routes { let mut packet_preparer = self.packet_preparer.clone(); let route = route.clone(); - let route_test_packets = self.route_test_packets; - let gateway_packets = tokio::spawn(async move { - packet_preparer.prepare_test_route_viability_packets(&route, route_test_packets) - }) - .await - .unwrap(); + let gateway_packets = packet_preparer + .prepare_test_route_viability_packets(&route, self.route_test_packets); packets.push(gateway_packets); } diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index 6c9027a70d..32f98097ca 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/sdk/rust/nym-sdk/Cargo.toml @@ -13,6 +13,7 @@ gateway-requests = { path = "../../../gateway/gateway-requests" } nym-network-defaults = { path = "../../../common/network-defaults" } nym-sphinx = { path = "../../../common/nymsphinx" } nym-task = { path = "../../../common/task" } +nym-topology = { path = "../../../common/topology" } validator-client = { path = "../../../common/client-libs/validator-client", features = ["nyxd-client"] } futures = "0.3" diff --git a/sdk/rust/nym-sdk/examples/custom_topology_provider.rs b/sdk/rust/nym-sdk/examples/custom_topology_provider.rs new file mode 100644 index 0000000000..59da44e711 --- /dev/null +++ b/sdk/rust/nym-sdk/examples/custom_topology_provider.rs @@ -0,0 +1,78 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_sdk::mixnet; +use nym_topology::provider_trait::{async_trait, TopologyProvider}; +use nym_topology::{nym_topology_from_detailed, NymTopology}; +use url::Url; + +struct MyTopologyProvider { + validator_client: validator_client::client::NymApiClient, +} + +impl MyTopologyProvider { + fn new(nym_api_url: Url) -> MyTopologyProvider { + MyTopologyProvider { + validator_client: validator_client::client::NymApiClient::new(nym_api_url), + } + } + + async fn get_topology(&self) -> NymTopology { + let mixnodes = self + .validator_client + .get_cached_active_mixnodes() + .await + .unwrap(); + + // in our topology provider only use mixnodes that have mix_id divisible by 3 + // and have more than 100k nym (i.e. 100'000'000'000 unym) in stake + // why? because this is just an example to showcase arbitrary uses and capabilities of this trait + let filtered_mixnodes = mixnodes + .into_iter() + .filter(|mix| { + mix.mix_id() % 3 == 0 && mix.total_stake() > "100000000000".parse().unwrap() + }) + .collect::>(); + + let gateways = self.validator_client.get_cached_gateways().await.unwrap(); + + nym_topology_from_detailed(filtered_mixnodes, gateways) + } +} + +#[async_trait] +impl TopologyProvider for MyTopologyProvider { + // this will be manually refreshed on a timer specified inside mixnet client config + async fn get_new_topology(&mut self) -> Option { + Some(self.get_topology().await) + } +} + +#[tokio::main] +async fn main() { + nym_bin_common::logging::setup_logging(); + + let nym_api = "https://validator.nymtech.net/api/".parse().unwrap(); + let my_topology_provider = MyTopologyProvider::new(nym_api); + + // Passing no config makes the client fire up an ephemeral session and figure shit out on its own + let mut client = mixnet::MixnetClientBuilder::new() + .custom_topology_provider(Box::new(my_topology_provider)) + .build::() + .await + .unwrap() + .connect_to_mixnet() + .await + .unwrap(); + + let our_address = client.nym_address(); + println!("Our client nym address is: {our_address}"); + + // Send a message through the mixnet to ourselves + client.send_str(*our_address, "hello there").await; + + println!("Waiting for message (ctrl-c to exit)"); + client + .on_messages(|msg| println!("Received: {}", String::from_utf8_lossy(&msg.message))) + .await; +} diff --git a/sdk/rust/nym-sdk/examples/manually_overwrite_topology.rs b/sdk/rust/nym-sdk/examples/manually_overwrite_topology.rs new file mode 100644 index 0000000000..0c951c736b --- /dev/null +++ b/sdk/rust/nym-sdk/examples/manually_overwrite_topology.rs @@ -0,0 +1,90 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_sdk::mixnet; +use nym_topology::mix::Layer; +use nym_topology::{mix, NymTopology}; +use std::collections::HashMap; + +#[tokio::main] +async fn main() { + nym_bin_common::logging::setup_logging(); + + // Passing no config makes the client fire up an ephemeral session and figure shit out on its own + let mut client = mixnet::MixnetClient::connect_new().await.unwrap(); + let starting_topology = client.read_current_topology().await.unwrap(); + + // but we don't like our default topology, we want to use only those very specific, hardcoded, nodes: + let mut mixnodes = HashMap::new(); + mixnodes.insert( + 1, + vec![mix::Node { + mix_id: 63, + owner: "n1k52k5n45cqt5qpjh8tcwmgqm0wkt355yy0g5vu".to_string(), + host: "172.105.92.48".parse().unwrap(), + mix_host: "172.105.92.48:1789".parse().unwrap(), + identity_key: "GLdR2NRVZBiCoCbv4fNqt9wUJZAnNjGXHkx3TjVAUzrK" + .parse() + .unwrap(), + sphinx_key: "CBmYewWf43iarBq349KhbfYMc9ys2ebXWd4Vp4CLQ5Rq" + .parse() + .unwrap(), + layer: Layer::One, + version: "1.1.0".to_string(), + }], + ); + mixnodes.insert( + 2, + vec![mix::Node { + mix_id: 23, + owner: "n1fzv4jc7fanl9s0qj02ge2ezk3kts545kjtek47".to_string(), + host: "178.79.143.65".parse().unwrap(), + mix_host: "178.79.143.65:1789".parse().unwrap(), + identity_key: "4Yr4qmEHd9sgsuQ83191FR2hD88RfsbMmB4tzhhZWriz" + .parse() + .unwrap(), + sphinx_key: "8ndjk5oZ6HxUZNScLJJ7hk39XtUqGexdKgW7hSX6kpWG" + .parse() + .unwrap(), + layer: Layer::Two, + version: "1.1.0".to_string(), + }], + ); + mixnodes.insert( + 3, + vec![mix::Node { + mix_id: 66, + owner: "n1ae2pjd7q9p0dea65pqkvcm4x9s264v4fktpyru".to_string(), + host: "139.162.247.97".parse().unwrap(), + mix_host: "139.162.247.97:1789".parse().unwrap(), + identity_key: "66UngapebhJRni3Nj52EW1qcNsWYiuonjkWJzHFsmyYY" + .parse() + .unwrap(), + sphinx_key: "7KyZh8Z8KxuVunqytAJ2eXFuZkCS7BLTZSzujHJZsGa2" + .parse() + .unwrap(), + layer: Layer::Three, + version: "1.1.0".to_string(), + }], + ); + + // but we like the available gateways, so keep using them! + // (we like them because the author of this example is too lazy to use the same hardcoded gateway + // during client initialisation to make sure we are able to send to ourselves : ) ) + let custom_topology = NymTopology::new(mixnodes, starting_topology.gateways().to_vec()); + + client.manually_overwrite_topology(custom_topology).await; + + // and everything we send now should only ever go via those nodes + + let our_address = client.nym_address(); + println!("Our client nym address is: {our_address}"); + + // Send a message through the mixnet to ourselves + client.send_str(*our_address, "hello there").await; + + println!("Waiting for message (ctrl-c to exit)"); + client + .on_messages(|msg| println!("Received: {}", String::from_utf8_lossy(&msg.message))) + .await; +} diff --git a/sdk/rust/nym-sdk/src/mixnet.rs b/sdk/rust/nym-sdk/src/mixnet.rs index 5dafb6e94d..c2a6c42108 100644 --- a/sdk/rust/nym-sdk/src/mixnet.rs +++ b/sdk/rust/nym-sdk/src/mixnet.rs @@ -52,4 +52,5 @@ pub use nym_sphinx::{ addressing::clients::{ClientIdentity, Recipient}, receiver::ReconstructedMessage, }; +pub use nym_topology::{provider_trait::TopologyProvider, NymTopology}; pub use paths::{GatewayKeyMode, KeyMode, StoragePaths}; diff --git a/sdk/rust/nym-sdk/src/mixnet/client.rs b/sdk/rust/nym-sdk/src/mixnet/client.rs index 36b51a0428..7b754d8af2 100644 --- a/sdk/rust/nym-sdk/src/mixnet/client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/client.rs @@ -23,6 +23,8 @@ use nym_task::{ }; use futures::StreamExt; +use nym_topology::provider_trait::TopologyProvider; +use nym_topology::NymTopology; use validator_client::nyxd::SigningNyxdClient; use crate::{Error, Result}; @@ -38,6 +40,7 @@ pub struct MixnetClientBuilder { storage_paths: Option, keys: Option, gateway_config: Option, + custom_topology_provider: Option>, } impl MixnetClientBuilder { @@ -70,6 +73,15 @@ impl MixnetClientBuilder { self } + #[must_use] + pub fn custom_topology_provider( + mut self, + topology_provider: Box, + ) -> Self { + self.custom_topology_provider = Some(topology_provider); + self + } + /// Construct a [`DisconnectedMixnetClient`] from the setup specified. pub async fn build(self) -> Result> where @@ -79,7 +91,12 @@ impl MixnetClientBuilder { let config = self.config.unwrap_or_default(); let storage_paths = self.storage_paths; - let mut client = DisconnectedMixnetClient::new(Some(config), storage_paths).await?; + let mut client = DisconnectedMixnetClient::new( + Some(config), + storage_paths, + self.custom_topology_provider, + ) + .await?; if let Some(keys) = self.keys { client.set_keys(keys); @@ -118,6 +135,9 @@ where /// The storage backend for reply-SURBs reply_storage_backend: B, + + /// Alternative provider of network topology used for constructing sphinx packets. + custom_topology_provider: Option>, } impl DisconnectedMixnetClient @@ -133,6 +153,7 @@ where async fn new( config: Option, paths: Option, + custom_topology_provider: Option>, ) -> Result> where ::StorageError: Send + Sync, @@ -189,6 +210,7 @@ where storage_paths: paths, state: BuilderState::New, reply_storage_backend, + custom_topology_provider, }) } @@ -377,7 +399,7 @@ where // TODO: we currently don't support having a bandwidth controller let bandwidth_controller = None; - let base_builder: BaseClientBuilder<'_, _, SigningNyxdClient> = BaseClientBuilder::new( + let mut base_builder: BaseClientBuilder<'_, _, SigningNyxdClient> = BaseClientBuilder::new( &gateway_endpoint_config, &self.config.debug_config, self.key_manager.clone(), @@ -387,6 +409,10 @@ where self.config.nym_api_endpoints.clone(), ); + if let Some(topology_provider) = self.custom_topology_provider { + base_builder = base_builder.with_topology_provider(topology_provider); + } + let mut started_client = base_builder.start_base().await?; let client_input = started_client.client_input.register_producer(); let mut client_output = started_client.client_output.register_consumer(); @@ -502,7 +528,7 @@ impl MixnetClient { } /// Get a shallow clone of [`ConnectionCommandSender`]. This is useful if you want to e.g - /// explictly close a transmission lane that is still sending data even though it should + /// explicitly close a transmission lane that is still sending data even though it should /// cancel. pub fn connection_command_sender(&self) -> ConnectionCommandSender { self.client_input.connection_command_sender.clone() @@ -514,6 +540,25 @@ impl MixnetClient { self.client_state.shared_lane_queue_lengths.clone() } + /// Change the network topology used by this client for constructing sphinx packets into the + /// provided one. + pub async fn manually_overwrite_topology(&self, new_topology: NymTopology) { + self.client_state + .topology_accessor + .manually_change_topology(new_topology) + .await + } + + /// Gets the value of the currently used network topology. + pub async fn read_current_topology(&self) -> Option { + self.client_state.topology_accessor.current_topology().await + } + + /// Restore default topology refreshing behaviour of this client. + pub fn restore_automatic_topology_refreshing(&self) { + self.client_state.topology_accessor.release_manual_control() + } + /// Sends stringy data to the supplied Nym address /// /// # Example diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index 615d17b9c4..1d135c18cb 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/service-providers/network-requester/Cargo.toml @@ -11,7 +11,7 @@ rust-version = "1.65" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -async-trait = { version = "0.1.51" } +async-trait = { workspace = true } clap = {version = "4.0", features = ["cargo", "derive"]} dirs = "4.0" futures = "0.3.24"