diff --git a/Cargo.lock b/Cargo.lock index 075d755cd5..a288322d19 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6747,6 +6747,7 @@ dependencies = [ "log", "nym-bin-common", "nym-crypto", + "nym-mixnet-contract-common", "nym-network-defaults", "nym-sphinx-acknowledgements", "nym-sphinx-addressing", diff --git a/common/client-libs/mixnet-client/src/client.rs b/common/client-libs/mixnet-client/src/client.rs index 4da308144f..15639508a2 100644 --- a/common/client-libs/mixnet-client/src/client.rs +++ b/common/client-libs/mixnet-client/src/client.rs @@ -123,13 +123,10 @@ impl Client { // We could have as well used conn.send_all(receiver.map(Ok)), but considering we don't care // about neither receiver nor the connection, it doesn't matter which one gets consumed if let Err(err) = receiver.map(Ok).forward(conn).await { - warn!("Failed to forward packets to {} - {err}", address); + warn!("Failed to forward packets to {address} - {err}"); } - debug!( - "connection manager to {} is finished. Either the connection failed or mixnet client got dropped", - address - ); + debug!("connection manager to {address} is finished. Either the connection failed or mixnet client got dropped"); } /// If we're trying to reconnect, determine how long we should wait. @@ -207,7 +204,7 @@ impl SendWithoutResponse for Client { if let Some(sender) = self.conn_new.get_mut(&address) { if let Err(err) = sender.channel.try_send(framed_packet) { if err.is_full() { - debug!("Connection to {} seems to not be able to handle all the traffic - dropping the current packet", address); + debug!("Connection to {address} seems to not be able to handle all the traffic - dropping the current packet"); // it's not a 'big' error, but we did not manage to send the packet // if the queue is full, we can't really do anything but to drop the packet Err(io::Error::new( @@ -215,10 +212,8 @@ impl SendWithoutResponse for Client { "connection queue is full", )) } else if err.is_disconnected() { - debug!( - "Connection to {} seems to be dead. attempting to re-establish it...", - address - ); + debug!("Connection to {address} seems to be dead. attempting to re-establish it..."); + // it's not a 'big' error, but we did not manage to send the packet, but queue // it up to send it as soon as the connection is re-established self.make_connection(address, err.into_inner()); @@ -238,7 +233,7 @@ impl SendWithoutResponse for Client { } } else { // there was never a connection to begin with - debug!("establishing initial connection to {}", address); + debug!("establishing initial connection to {address}"); // it's not a 'big' error, but we did not manage to send the packet, but queue the packet // for sending for as soon as the connection is created self.make_connection(address, framed_packet); diff --git a/common/client-libs/validator-client/src/nyxd/mod.rs b/common/client-libs/validator-client/src/nyxd/mod.rs index c9c27370e2..f1733e095e 100644 --- a/common/client-libs/validator-client/src/nyxd/mod.rs +++ b/common/client-libs/validator-client/src/nyxd/mod.rs @@ -55,6 +55,9 @@ pub use tendermint_rpc::{ }; pub use tendermint_rpc::{Request, Response, SimpleRequest}; +#[cfg(feature = "http-client")] +pub use tendermint_rpc::Url as RpcUrl; + #[cfg(feature = "http-client")] use crate::http_client; #[cfg(feature = "http-client")] diff --git a/common/cosmwasm-smart-contracts/mixnet-contract/src/mixnode.rs b/common/cosmwasm-smart-contracts/mixnet-contract/src/mixnode.rs index a09d21dd2d..5d4d96e34e 100644 --- a/common/cosmwasm-smart-contracts/mixnet-contract/src/mixnode.rs +++ b/common/cosmwasm-smart-contracts/mixnet-contract/src/mixnode.rs @@ -623,6 +623,24 @@ pub enum Layer { Three = 3, } +impl Layer { + pub fn try_next(&self) -> Option { + match self { + Layer::One => Some(Layer::Two), + Layer::Two => Some(Layer::Three), + Layer::Three => None, + } + } + + pub fn try_previous(&self) -> Option { + match self { + Layer::One => None, + Layer::Two => Some(Layer::One), + Layer::Three => Some(Layer::Two), + } + } +} + impl From for String { fn from(layer: Layer) -> Self { (layer as u8).to_string() diff --git a/common/mixnode-common/Cargo.toml b/common/mixnode-common/Cargo.toml index aec5b7849c..894069f0bb 100644 --- a/common/mixnode-common/Cargo.toml +++ b/common/mixnode-common/Cargo.toml @@ -19,6 +19,7 @@ tokio = { workspace = true, features = [ "rt", "net", "io-util", + "sync" ] } tokio-util = { workspace = true, features = ["codec"] } url = { workspace = true } @@ -37,6 +38,7 @@ nym-sphinx-params = { path = "../nymsphinx/params" } nym-sphinx-types = { path = "../nymsphinx/types" } nym-task = { path = "../task" } nym-validator-client = { path = "../client-libs/validator-client" } +nym-mixnet-contract-common = { path = "../cosmwasm-smart-contracts/mixnet-contract" } nym-bin-common = { path = "../bin-common" } cfg-if = "1.0.0" diff --git a/common/mixnode-common/src/forward_travel/error.rs b/common/mixnode-common/src/forward_travel/error.rs new file mode 100644 index 0000000000..d2aadc1068 --- /dev/null +++ b/common/mixnode-common/src/forward_travel/error.rs @@ -0,0 +1,27 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_validator_client::nyxd::error::NyxdError; +use std::net::IpAddr; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum ForwardTravelError { + #[error("received a connection request from a forbidden address: '{address}'")] + DisallowedIngressAddress { address: IpAddr }, + + #[error("received a request to open connection to a forbidden address: '{address}'")] + DisallowedEgressAddress { address: IpAddr }, + + #[error("no valid nyxd urls are available for topology queries")] + NoNyxdUrlsAvailable, + + #[error("nyxd interaction failure: {source}")] + NyxdFailure { + #[from] + source: NyxdError, + }, + + #[error("the current epoch appears to be stuck")] + StuckEpoch, +} diff --git a/common/mixnode-common/src/forward_travel/mod.rs b/common/mixnode-common/src/forward_travel/mod.rs new file mode 100644 index 0000000000..17ee42396f --- /dev/null +++ b/common/mixnode-common/src/forward_travel/mod.rs @@ -0,0 +1,312 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::forward_travel::error::ForwardTravelError; +use log::{debug, error, trace, warn}; +use nym_mixnet_contract_common::{EpochId, GatewayBond, Layer, MixNodeBond}; +use nym_network_defaults::NymNetworkDetails; +use nym_task::TaskClient; +use nym_validator_client::client::IdentityKey; +use nym_validator_client::nyxd::contract_traits::{MixnetQueryClient, PagedMixnetQueryClient}; +use nym_validator_client::{nyxd, QueryHttpRpcNyxdClient}; +use rand::seq::SliceRandom; +use rand::{thread_rng, Rng}; +use std::collections::HashSet; +use std::mem; +use std::net::{IpAddr, ToSocketAddrs}; +use std::str::FromStr; +use std::sync::Arc; +use std::time::Duration; +use tokio::{sync::RwLock, time::sleep}; +use url::Url; + +pub mod error; + +pub type Ingress = AllowedPaths; +pub type Egress = AllowedPaths; + +pub struct AllowedAddressesProvider { + current_epoch: EpochId, + + identity: IdentityKey, + + client_config: nyxd::Config, + + /// URLs to the nyxd validators for obtaining unfiltered network topology. + nyxd_endpoints: Vec, + + ingress: Ingress, + egress: Egress, +} + +impl AllowedAddressesProvider { + fn new(&self, network_details: Option) -> Result { + todo!() + } + + fn ephemeral_nyxd_client(&self) -> Result { + let mut possible_nyxd_endpoints = self.nyxd_endpoints.clone(); + possible_nyxd_endpoints.shuffle(&mut thread_rng()); + + let mut last_error = match QueryHttpRpcNyxdClient::connect( + self.client_config.clone(), + possible_nyxd_endpoints + .pop() + .ok_or(ForwardTravelError::NoNyxdUrlsAvailable)? + .as_str(), + ) { + Ok(client) => return Ok(client), + Err(err) => err, + }; + + for url in possible_nyxd_endpoints { + match QueryHttpRpcNyxdClient::connect(self.client_config.clone(), url.as_str()) { + Ok(client) => return Ok(client), + Err(err) => last_error = err, + }; + } + + Err(last_error.into()) + } + + pub fn ingress(&self) -> Ingress { + Ingress { + inner: Arc::clone(&self.ingress.inner), + } + } + + pub fn egress(&self) -> Egress { + Egress { + inner: Arc::clone(&self.egress.inner), + } + } + + /// Gets ip addresses of all mixnodes on given layer + fn get_addresses_on_layer(layer: Layer, nodes: &[MixNodeBond]) -> HashSet { + let mut allowed = HashSet::new(); + + for node in nodes.iter().filter(|m| m.layer == layer) { + match IpAddr::from_str(&node.mix_node.host) { + Ok(ip) => { + allowed.insert(ip); + } + Err(_) => { + // this might still be a valid hostname + + // annoyingly there exists a method of looking up a socket address but not an ip address, + // so append any port and perform the lookup + let Ok(sockets) = format!("{}:{}", node.mix_node.host, node.mix_node.mix_port) + .to_socket_addrs() + else { + warn!( + "failed to resolve ip address of mixnode '{}' (hostname: {})", + node.identity(), + node.mix_node.host + ); + continue; + }; + + for socket in sockets { + allowed.insert(socket.ip()); + } + } + } + } + + allowed + } + + fn gateway_addresses(gateways: &[GatewayBond]) -> HashSet { + let mut allowed = HashSet::new(); + + for gateway in gateways.iter() { + match IpAddr::from_str(&gateway.gateway.host) { + Ok(ip) => { + allowed.insert(ip); + } + Err(_) => { + // this might still be a valid hostname + + // annoyingly there exists a method of looking up a socket address but not an ip address, + // so append any port and perform the lookup + let Ok(sockets) = + format!("{}:{}", gateway.gateway.host, gateway.gateway.mix_port) + .to_socket_addrs() + else { + warn!( + "failed to resolve ip address of gateway '{}' (hostname: {})", + gateway.identity(), + gateway.gateway.host + ); + continue; + }; + + for socket in sockets { + allowed.insert(socket.ip()); + } + } + } + } + + allowed + } + + fn locate_layer(&self, nodes: &[MixNodeBond]) -> Option { + nodes + .iter() + .find(|m| m.identity() == self.identity) + .map(|m| m.layer) + } + + fn is_gateway(&self, gateways: &[GatewayBond]) -> bool { + gateways + .iter() + .find(|g| g.gateway.identity_key == self.identity) + .is_some() + } + + async fn update_state( + &mut self, + client: QueryHttpRpcNyxdClient, + ) -> Result<(), ForwardTravelError> { + let current_interval = client.get_current_interval_details().await?; + let current_epoch = current_interval.interval.current_epoch_absolute_id(); + + if current_epoch == self.current_epoch { + error!("can't update the allowed ips list as the epoch appears to be stuck"); + return Err(ForwardTravelError::StuckEpoch); + } + + let has_epoch_deviated = current_epoch > self.current_epoch + 1; + + let mixnodes = client.get_all_mixnode_bonds().await?; + let our_mix_layer = self.locate_layer(&mixnodes); + + let previous_mix_layer = our_mix_layer.map(|l| l.try_previous()).flatten(); + let next_mix_layer = our_mix_layer.map(|l| l.try_next()).flatten(); + + let (allowed_ingress, allowed_egress) = match (previous_mix_layer, next_mix_layer) { + (Some(previous), Some(next)) => ( + Self::get_addresses_on_layer(previous, &mixnodes), + Self::get_addresses_on_layer(next, &mixnodes), + ), + (Some(previous), None) => { + let gateways = client.get_all_gateways().await?; + ( + Self::get_addresses_on_layer(previous, &mixnodes), + Self::gateway_addresses(&gateways), + ) + } + (None, Some(next)) => { + let gateways = client.get_all_gateways().await?; + + ( + Self::gateway_addresses(&gateways), + Self::get_addresses_on_layer(next, &mixnodes), + ) + } + + _ => unreachable!("both previous and next layers are set to be gateways!"), + }; + + self.current_epoch = current_epoch; + self.ingress + .advance_epoch(allowed_ingress, has_epoch_deviated) + .await; + self.egress + .advance_epoch(allowed_egress, has_epoch_deviated) + .await; + + Ok(()) + } + + async fn update_allowed_addresses(&mut self) -> Result<(), ForwardTravelError> { + // create new client every epoch because it results in different nyxd endpoint being used + // what may help in distributing the load + let client = self.ephemeral_nyxd_client()?; + self.wait_for_epoch_rollover(&client).await?; + self.update_state(client).await + } + + async fn wait_for_epoch_rollover( + &self, + client: &QueryHttpRpcNyxdClient, + ) -> Result<(), ForwardTravelError> { + let current_interval = client.get_current_interval_details().await?; + let current_epoch = current_interval.interval.current_epoch_absolute_id(); + + if current_epoch <= self.current_epoch { + let remaining = current_interval.time_until_current_epoch_end(); + // add few more seconds to account for block time drift and to spread queries of all + // other nodes + let adjustment_secs = rand::thread_rng().gen_range(5..90); + sleep(remaining + Duration::from_secs(adjustment_secs)).await; + } + + Ok(()) + } + + async fn run(&mut self, mut task_client: TaskClient) { + debug!("Started ValidAddressesProvider with graceful shutdown support"); + while !task_client.is_shutdown() { + tokio::select! { + biased; + _ = task_client.recv() => { + trace!("ValidAddressesProvider: Received shutdown"); + } + res = self.update_allowed_addresses() => { + if let Err(err) = res { + warn!("failed to update the allowed addresses: {err}"); + // don't retry immediately in case it was a network failure, wait a bit instead. + tokio::select! { + biased; + _ = task_client.recv() => { + trace!("ValidAddressesProvider: Received shutdown"); + } + _ = sleep(Duration::from_secs(5 * 60)) => {} + } + } + } + } + } + task_client.recv_timeout().await; + log::debug!("ValidAddressesProvider: Exiting"); + } +} + +pub struct AllowedPaths { + inner: Arc>, +} + +impl AllowedPaths { + fn new() -> Self { + AllowedPaths { + inner: Arc::new(RwLock::new(AllowedPathsInner { + previous_epoch: HashSet::new(), + current_epoch: HashSet::new(), + })), + } + } + + pub async fn is_allowed(&self, address: IpAddr) -> bool { + let guard = self.inner.read().await; + guard.current_epoch.contains(&address) || guard.previous_epoch.contains(&address) + } + + async fn advance_epoch(&self, current_epoch: HashSet, reset_previous: bool) { + let mut guard = self.inner.write().await; + + let old_current = mem::replace(&mut guard.current_epoch, current_epoch); + + if reset_previous { + guard.previous_epoch = HashSet::new() + } else { + guard.previous_epoch = old_current; + } + } +} + +struct AllowedPathsInner { + previous_epoch: HashSet, + current_epoch: HashSet, +} diff --git a/common/mixnode-common/src/lib.rs b/common/mixnode-common/src/lib.rs index 9689b157a5..f781400b35 100644 --- a/common/mixnode-common/src/lib.rs +++ b/common/mixnode-common/src/lib.rs @@ -1,5 +1,7 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 + +pub mod forward_travel; pub mod packet_processor; pub mod verloc; diff --git a/nym-connect/desktop/Cargo.lock b/nym-connect/desktop/Cargo.lock index c7a91547c7..bfc7e6f5d0 100644 --- a/nym-connect/desktop/Cargo.lock +++ b/nym-connect/desktop/Cargo.lock @@ -7448,9 +7448,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.31.0" +version = "1.33.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "40de3a2ba249dcb097e01be5e67a5ff53cf250397715a071a81543e8a832a920" +checksum = "4f38200e3ef7995e5ef13baec2f432a6da0aa9ac495b2c0e8f3b7eec2c92d653" dependencies = [ "backtrace", "bytes",