module responsible for updating paths allowed for forward travel

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",

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);

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")]

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<Self> {
+        match self {
+            Layer::One => Some(Layer::Two),
+            Layer::Two => Some(Layer::Three),
+            Layer::Three => None,
+        }
+    }
+
+    pub fn try_previous(&self) -> Option<Self> {
+        match self {
+            Layer::One => None,
+            Layer::Two => Some(Layer::One),
+            Layer::Three => Some(Layer::Two),
+        }
+    }
+}
+
 impl From<Layer> for String {
     fn from(layer: Layer) -> Self {
         (layer as u8).to_string()

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"

common/mixnode-common/src/forward_travel/error.rs

@@ -0,0 +1,27 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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,
+}

common/mixnode-common/src/forward_travel/mod.rs

@@ -0,0 +1,312 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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<Url>,
+
+    ingress: Ingress,
+    egress: Egress,
+}
+
+impl AllowedAddressesProvider {
+    fn new(&self, network_details: Option<NymNetworkDetails>) -> Result<Self, ForwardTravelError> {
+        todo!()
+    }
+
+    fn ephemeral_nyxd_client(&self) -> Result<QueryHttpRpcNyxdClient, ForwardTravelError> {
+        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<IpAddr> {
+        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<IpAddr> {
+        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<Layer> {
+        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<RwLock<AllowedPathsInner>>,
+}
+
+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<IpAddr>, 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<IpAddr>,
+    current_epoch: HashSet<IpAddr>,
+}

common/mixnode-common/src/lib.rs

@@ -1,5 +1,7 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
+
+pub mod forward_travel;
 pub mod packet_processor;
 pub mod verloc;
 

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",