feature: disallow routing mix packets to nodes not present in the topology (#5526)

* new NymNodeTopologyProvider to also keep track of ips of all nodes

* added nym-api endpoint for nodes existence by ip

* change behaviour of updating allowed nodes alongside the topology

* clippy

* license fix

* fix default filtering limit

common/client-libs/validator-client/src/client.rs

@@ -23,11 +23,12 @@ use nym_api_requests::models::{
     NymNodeDescription, RewardEstimationResponse, StakeSaturationResponse,
 };
 use nym_api_requests::models::{LegacyDescribedGateway, MixNodeBondAnnotated};
-use nym_api_requests::nym_nodes::SkimmedNode;
+use nym_api_requests::nym_nodes::{NodesByAddressesResponse, SkimmedNode};
 use nym_coconut_dkg_common::types::EpochId;
 use nym_ecash_contract_common::deposit::DepositId;
 use nym_http_api_client::UserAgent;
 use nym_network_defaults::NymNetworkDetails;
+use std::net::IpAddr;
 use time::Date;
 use url::Url;
 
@@ -710,4 +711,11 @@ impl NymApiClient {
             .issued_ticketbooks_challenge(expiration_date, deposits)
             .await?)
     }
+
+    pub async fn nodes_by_addresses(
+        &self,
+        addresses: Vec<IpAddr>,
+    ) -> Result<NodesByAddressesResponse, ValidatorClientError> {
+        Ok(self.nym_api.nodes_by_addresses(addresses).await?)
+    }
 }

common/client-libs/validator-client/src/nym_api/mod.rs

@@ -15,7 +15,9 @@ use nym_api_requests::models::{
     AnnotationResponse, ApiHealthResponse, LegacyDescribedMixNode, NodePerformanceResponse,
     NodeRefreshBody, NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse,
 };
-use nym_api_requests::nym_nodes::PaginatedCachedNodesResponse;
+use nym_api_requests::nym_nodes::{
+    NodesByAddressesRequestBody, NodesByAddressesResponse, PaginatedCachedNodesResponse,
+};
 use nym_api_requests::pagination::PaginatedResponse;
 pub use nym_api_requests::{
     ecash::{
@@ -40,6 +42,7 @@ pub use nym_http_api_client::Client;
 use nym_http_api_client::{ApiClient, NO_PARAMS};
 use nym_mixnet_contract_common::mixnode::MixNodeDetails;
 use nym_mixnet_contract_common::{GatewayBond, IdentityKeyRef, NodeId, NymNodeDetails};
+use std::net::IpAddr;
 use time::format_description::BorrowedFormatItem;
 use time::Date;
 use tracing::instrument;
@@ -1015,6 +1018,23 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
+    async fn nodes_by_addresses(
+        &self,
+        addresses: Vec<IpAddr>,
+    ) -> Result<NodesByAddressesResponse, NymAPIError> {
+        self.post_json(
+            &[
+                routes::API_VERSION,
+                "unstable",
+                routes::NYM_NODES_ROUTES,
+                routes::nym_nodes::BY_ADDRESSES,
+            ],
+            NO_PARAMS,
+            &NodesByAddressesRequestBody { addresses },
+        )
+        .await
+    }
+
     #[instrument(level = "debug", skip(self))]
     async fn get_network_details(&self) -> Result<NymNetworkDetailsResponse, NymAPIError> {
         self.get_json(

common/client-libs/validator-client/src/nym_api/routes.rs

@@ -43,6 +43,7 @@ pub mod nym_nodes {
     pub const NYM_NODES_BONDED: &str = "bonded";
     pub const NYM_NODES_REWARDED_SET: &str = "rewarded-set";
     pub const NYM_NODES_REFRESH_DESCRIBED: &str = "refresh-described";
+    pub const BY_ADDRESSES: &str = "by-addresses";
 }
 
 pub const STATUS_ROUTES: &str = "status";

common/config/src/helpers.rs

@@ -25,15 +25,6 @@ pub fn in6addr_any_init() -> IpAddr {
     IpAddr::V6(Ipv6Addr::UNSPECIFIED)
 }
 
-/// Helper for providing binding warnings if node tries to bind to any of those
-pub const SPECIAL_ADDRESSES: &[IpAddr] = &[
-    IpAddr::V4(Ipv4Addr::LOCALHOST),
-    IpAddr::V4(Ipv4Addr::UNSPECIFIED),
-    IpAddr::V4(Ipv4Addr::BROADCAST),
-    IpAddr::V6(Ipv6Addr::LOCALHOST),
-    IpAddr::V6(Ipv6Addr::UNSPECIFIED),
-];
-
 // TODO: is it really part of 'Config'?
 pub trait OptionalSet {
     /// If the value is available (i.e. `Some`), the provided closure is applied.

common/nymsphinx/addressing/src/nodes.rs

@@ -161,6 +161,12 @@ impl From<NymNodeRoutingAddress> for SocketAddr {
     }
 }
 
+impl AsRef<SocketAddr> for NymNodeRoutingAddress {
+    fn as_ref(&self) -> &SocketAddr {
+        &self.0
+    }
+}
+
 impl TryInto<NodeAddressBytes> for NymNodeRoutingAddress {
     type Error = NymNodeRoutingAddressError;
 

common/topology/src/lib.rs

@@ -254,6 +254,15 @@ impl NymTopology {
         }
     }
 
+    pub fn with_additional_nodes<N>(mut self, nodes: impl Iterator<Item = N>) -> Self
+    where
+        N: TryInto<RoutingNode>,
+        <N as TryInto<RoutingNode>>::Error: Display,
+    {
+        self.add_additional_nodes(nodes);
+        self
+    }
+
     pub fn has_node_details(&self, node_id: NodeId) -> bool {
         self.node_details.contains_key(&node_id)
     }

nym-api/nym-api-requests/src/nym_nodes.rs

@@ -10,6 +10,7 @@ use nym_mixnet_contract_common::nym_node::Role;
 use nym_mixnet_contract_common::reward_params::Performance;
 use nym_mixnet_contract_common::{Interval, NodeId};
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 use std::net::IpAddr;
 use time::OffsetDateTime;
 use utoipa::ToSchema;
@@ -212,3 +213,15 @@ pub struct FullFatNode {
     // kinda temporary for now to make as few changes as possible for now
     pub self_described: Option<NymNodeData>,
 }
+
+#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, ToSchema)]
+pub struct NodesByAddressesRequestBody {
+    #[schema(value_type = Vec<String>)]
+    pub addresses: Vec<IpAddr>,
+}
+
+#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, ToSchema)]
+pub struct NodesByAddressesResponse {
+    #[schema(value_type = HashMap<String, Option<u32>>)]
+    pub existence: HashMap<IpAddr, Option<NodeId>>,
+}

nym-api/src/node_describe_cache/mod.rs

@@ -17,6 +17,7 @@ use nym_mixnet_contract_common::{NodeId, NymNodeDetails};
 use nym_node_requests::api::client::{NymNodeApiClientError, NymNodeApiClientExt};
 use nym_topology::node::{RoutingNode, RoutingNodeError};
 use std::collections::HashMap;
+use std::net::IpAddr;
 use std::time::Duration;
 use thiserror::Error;
 use tracing::{debug, error, info};
@@ -84,10 +85,14 @@ impl NodeDescriptionTopologyExt for NymNodeDescription {
 #[derive(Debug, Clone)]
 pub struct DescribedNodes {
     nodes: HashMap<NodeId, NymNodeDescription>,
+    addresses_cache: HashMap<IpAddr, NodeId>,
 }
 
 impl DescribedNodes {
     pub fn force_update(&mut self, node: NymNodeDescription) {
+        for ip in &node.description.host_information.ip_address {
+            self.addresses_cache.insert(*ip, node.node_id);
+        }
         self.nodes.insert(node.node_id, node);
     }
 
@@ -129,6 +134,10 @@ impl DescribedNodes {
             .filter(|n| n.contract_node_type == DescribedNodeType::NymNode)
             .filter(|n| n.description.declared_role.can_operate_exit_gateway())
     }
+
+    pub fn node_with_address(&self, address: IpAddr) -> Option<NodeId> {
+        self.addresses_cache.get(&address).copied()
+    }
 }
 
 pub struct NodeDescriptionProvider {
@@ -396,9 +405,20 @@ impl CacheItemProvider for NodeDescriptionProvider {
         .collect::<HashMap<_, _>>()
         .await;
 
-        info!("refreshed self described data for {} nodes", nodes.len());
+        let mut addresses_cache = HashMap::new();
+        for node in nodes.values() {
+            for ip in &node.description.host_information.ip_address {
+                addresses_cache.insert(*ip, node.node_id);
+            }
+        }
 
-        Ok(DescribedNodes { nodes })
+        info!("refreshed self described data for {} nodes", nodes.len());
+        info!("with {} unique ip addresses", addresses_cache.len());
+
+        Ok(DescribedNodes {
+            nodes,
+            addresses_cache,
+        })
     }
 }
 

nym-api/src/nym_nodes/handlers/unstable/mod.rs

@@ -20,6 +20,7 @@
 //!   - `/mixnodes/<tier>` => only returns mixnode role data
 //!   - `/gateway/<tier>` => only returns (entry) gateway role data
 
+use crate::node_status_api::models::{AxumErrorResponse, AxumResult};
 use crate::nym_nodes::handlers::unstable::full_fat::nodes_detailed;
 use crate::nym_nodes::handlers::unstable::semi_skimmed::nodes_expanded;
 use crate::nym_nodes::handlers::unstable::skimmed::{
@@ -29,10 +30,14 @@ use crate::nym_nodes::handlers::unstable::skimmed::{
 };
 use crate::support::http::helpers::PaginationRequest;
 use crate::support::http::state::AppState;
-use axum::routing::get;
-use axum::Router;
-use nym_api_requests::nym_nodes::NodeRoleQueryParam;
+use axum::extract::State;
+use axum::routing::{get, post};
+use axum::{Json, Router};
+use nym_api_requests::nym_nodes::{
+    NodeRoleQueryParam, NodesByAddressesRequestBody, NodesByAddressesResponse,
+};
 use serde::Deserialize;
+use std::collections::HashMap;
 use tower_http::compression::CompressionLayer;
 
 pub(crate) mod full_fat;
@@ -74,6 +79,7 @@ pub(crate) fn nym_node_routes_unstable() -> Router<AppState> {
         .nest("/full-fat", Router::new().route("/", get(nodes_detailed)))
         .route("/gateways/skimmed", get(skimmed::deprecated_gateways_basic))
         .route("/mixnodes/skimmed", get(skimmed::deprecated_mixnodes_basic))
+        .route("/by-addresses", post(nodes_by_addresses))
         .layer(CompressionLayer::new())
 }
 
@@ -129,3 +135,36 @@ impl<'a> From<&'a NodesParams> for PaginationRequest {
         }
     }
 }
+
+#[utoipa::path(
+    tag = "Unstable Nym Nodes",
+    post,
+    request_body = NodesByAddressesRequestBody,
+    path = "/by-addresses",
+    context_path = "/v1/unstable/nym-nodes",
+    responses(
+        (status = 200, body = NodesByAddressesResponse)
+    )
+)]
+async fn nodes_by_addresses(
+    state: State<AppState>,
+    Json(body): Json<NodesByAddressesRequestBody>,
+) -> AxumResult<Json<NodesByAddressesResponse>> {
+    // if the request is too big, simply reject it
+    if body.addresses.len() > 100 {
+        return Err(AxumErrorResponse::bad_request(
+            "requested too many addresses",
+        ));
+    }
+
+    // TODO: perhaps introduce different cache because realistically nym-api will receive
+    // request for the same couple addresses from all nodes in quick succession
+    let describe_cache = state.describe_nodes_cache_data().await?;
+
+    let mut existence = HashMap::new();
+    for address in body.addresses {
+        existence.insert(address, describe_cache.node_with_address(address));
+    }
+
+    Ok(Json(NodesByAddressesResponse { existence }))
+}

nym-node/src/cli/commands/run/mod.rs

@@ -4,8 +4,8 @@
 use crate::config::upgrade_helpers::try_load_current_config;
 use crate::error::NymNodeError;
 use crate::node::bonding_information::BondingInformation;
+use crate::node::mixnet::packet_forwarding::global::is_global_ip;
 use crate::node::NymNode;
-use nym_config::helpers::SPECIAL_ADDRESSES;
 use std::fs;
 use std::net::IpAddr;
 use tracing::{debug, info, trace, warn};
@@ -17,7 +17,7 @@ pub(crate) use args::Args;
 fn check_public_ips(ips: &[IpAddr], local: bool) -> Result<(), NymNodeError> {
     let mut suspicious_ip = Vec::new();
     for ip in ips {
-        if SPECIAL_ADDRESSES.contains(ip) {
+        if !is_global_ip(ip) {
             if !local {
                 return Err(NymNodeError::InvalidPublicIp { address: *ip });
             }
@@ -92,6 +92,11 @@ pub(crate) async fn execute(mut args: Args) -> Result<(), NymNodeError> {
     }
     check_public_ips(&config.host.public_ips, local)?;
 
+    let mut config = config;
+    if local {
+        config.debug.testnet = true
+    }
+
     let nym_node = NymNode::new(config)
         .await?
         .with_accepted_operator_terms_and_conditions(accepted_operator_terms_and_conditions);

nym-node/src/config/gateway_tasks.rs

@@ -47,6 +47,10 @@ pub struct Debug {
     /// Number of messages from offline client that can be pulled at once (i.e. with a single SQL query) from the storage.
     pub message_retrieval_limit: i64,
 
+    /// Specifies the minimum performance of mixnodes in the network that are to be used in internal topologies
+    /// of the services providers
+    pub minimum_mix_performance: u8,
+
     /// Defines the maximum age of a signed authentication request before it's deemed too stale to process.
     pub maximum_auth_request_age: Duration,
 
@@ -59,6 +63,7 @@ pub struct Debug {
 
 impl Debug {
     pub const DEFAULT_MESSAGE_RETRIEVAL_LIMIT: i64 = 100;
+    pub const DEFAULT_MINIMUM_MIX_PERFORMANCE: u8 = 50;
     pub const DEFAULT_MAXIMUM_AUTH_REQUEST_AGE: Duration = Duration::from_secs(30);
 }
 
@@ -67,6 +72,7 @@ impl Default for Debug {
         Debug {
             message_retrieval_limit: Self::DEFAULT_MESSAGE_RETRIEVAL_LIMIT,
             maximum_auth_request_age: Self::DEFAULT_MAXIMUM_AUTH_REQUEST_AGE,
+            minimum_mix_performance: Self::DEFAULT_MINIMUM_MIX_PERFORMANCE,
             stale_messages: Default::default(),
             client_bandwidth: Default::default(),
             zk_nym_tickets: Default::default(),

nym-node/src/config/mod.rs

@@ -780,16 +780,26 @@ pub struct Debug {
     /// Specifies the time to live of the internal topology provider cache.
     #[serde(with = "humantime_serde")]
     pub topology_cache_ttl: Duration,
+
+    /// Specifies the time between attempting to resolve any pending unknown nodes in the routing filter
+    #[serde(with = "humantime_serde")]
+    pub routing_nodes_check_interval: Duration,
+
+    /// Specifies whether this node runs in testnet mode thus allowing it to route packets on local interfaces
+    pub testnet: bool,
 }
 
 impl Debug {
-    pub const DEFAULT_TOPOLOGY_CACHE_TTL: Duration = Duration::from_secs(5 * 60);
+    pub const DEFAULT_TOPOLOGY_CACHE_TTL: Duration = Duration::from_secs(10 * 60);
+    pub const DEFAULT_ROUTING_NODES_CHECK_INTERVAL: Duration = Duration::from_secs(5 * 60);
 }
 
 impl Default for Debug {
     fn default() -> Self {
         Debug {
             topology_cache_ttl: Self::DEFAULT_TOPOLOGY_CACHE_TTL,
+            routing_nodes_check_interval: Self::DEFAULT_ROUTING_NODES_CHECK_INTERVAL,
+            testnet: false,
         }
     }
 }

nym-node/src/error.rs

@@ -4,6 +4,7 @@
 use crate::node::http::error::NymNodeHttpError;
 use crate::wireguard::error::WireguardError;
 use nym_ip_packet_router::error::ClientCoreError;
+use nym_validator_client::ValidatorClientError;
 use std::io;
 use std::net::IpAddr;
 use std::path::PathBuf;
@@ -141,6 +142,11 @@ pub enum NymNodeError {
         source: ipnetwork::IpNetworkError,
     },
 
+    #[error(
+        "failed to retrieve initial network topology - can't start the node without it: {source}"
+    )]
+    InitialTopologyQueryFailure { source: ValidatorClientError },
+
     #[error(transparent)]
     GatewayFailure(#[from] nym_gateway::GatewayError),
 

nym-node/src/node/mixnet/packet_forwarding/global.rs

@@ -0,0 +1,79 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+
+// use `ip` feature without nightly
+// issue: https://github.com/rust-lang/rust/issues/27709
+pub(crate) const fn is_global_ip(ip: &IpAddr) -> bool {
+    match ip {
+        IpAddr::V4(addr) => is_global_ipv4(addr),
+        IpAddr::V6(addr) => is_global_ipv6(addr),
+    }
+}
+
+const fn is_shared_ipv4(ip: &Ipv4Addr) -> bool {
+    ip.octets()[0] == 100 && (ip.octets()[1] & 0b1100_0000 == 0b0100_0000)
+}
+
+const fn is_benchmarking_ipv4(ip: &Ipv4Addr) -> bool {
+    ip.octets()[0] == 198 && (ip.octets()[1] & 0xfe) == 18
+}
+
+const fn is_reserved_ipv4(ip: &Ipv4Addr) -> bool {
+    ip.octets()[0] & 240 == 240 && !ip.is_broadcast()
+}
+
+const fn is_global_ipv4(ip: &Ipv4Addr) -> bool {
+    !(ip.octets()[0] == 0 // "This network"
+        || ip.is_private()
+        || is_shared_ipv4(ip)
+        || ip.is_loopback()
+        || ip.is_link_local()
+        // addresses reserved for future protocols (`192.0.0.0/24`)
+        // .9 and .10 are documented as globally reachable so they're excluded
+        || (
+        ip.octets()[0] == 192 && ip.octets()[1] == 0 && ip.octets()[2] == 0
+            && ip.octets()[3] != 9 && ip.octets()[3] != 10
+    )
+        || ip.is_documentation()
+        || is_benchmarking_ipv4(ip)
+        || is_reserved_ipv4(ip)
+        || ip.is_broadcast())
+}
+
+const fn is_documentation_ipv6(ip: &Ipv6Addr) -> bool {
+    (ip.segments()[0] == 0x2001) && (ip.segments()[1] == 0xdb8)
+}
+
+const fn is_global_ipv6(ip: &Ipv6Addr) -> bool {
+    !(ip.is_unspecified()
+        || ip.is_loopback()
+        // IPv4-mapped Address (`::ffff:0:0/96`)
+        || matches!(ip.segments(), [0, 0, 0, 0, 0, 0xffff, _, _])
+        // IPv4-IPv6 Translat. (`64:ff9b:1::/48`)
+        || matches!(ip.segments(), [0x64, 0xff9b, 1, _, _, _, _, _])
+        // Discard-Only Address Block (`100::/64`)
+        || matches!(ip.segments(), [0x100, 0, 0, 0, _, _, _, _])
+        // IETF Protocol Assignments (`2001::/23`)
+        || (matches!(ip.segments(), [0x2001, b, _, _, _, _, _, _] if b < 0x200)
+        && !(
+        // Port Control Protocol Anycast (`2001:1::1`)
+        u128::from_be_bytes(ip.octets()) == 0x2001_0001_0000_0000_0000_0000_0000_0001
+            // Traversal Using Relays around NAT Anycast (`2001:1::2`)
+            || u128::from_be_bytes(ip.octets()) == 0x2001_0001_0000_0000_0000_0000_0000_0002
+            // AMT (`2001:3::/32`)
+            || matches!(ip.segments(), [0x2001, 3, _, _, _, _, _, _])
+            // AS112-v6 (`2001:4:112::/48`)
+            || matches!(ip.segments(), [0x2001, 4, 0x112, _, _, _, _, _])
+            // ORCHIDv2 (`2001:20::/28`)
+            // Drone Remote ID Protocol Entity Tags (DETs) Prefix (`2001:30::/28`)`
+            || matches!(ip.segments(), [0x2001, b, _, _, _, _, _, _] if b >= 0x20 && b <= 0x3F)
+    ))
+        // 6to4 (`2002::/16`) – it's not explicitly documented as globally reachable,
+        // IANA says N/A.
+        || matches!(ip.segments(), [0x2002, _, _, _, _, _, _, _])
+        || is_documentation_ipv6(ip)
+        || ip.is_unique_local()
+        || ip.is_unicast_link_local())
+}

nym-node/src/node/mixnet/packet_forwarding/mod.rs

@@ -1,6 +1,8 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
+use crate::node::mixnet::packet_forwarding::global::is_global_ip;
+use crate::node::shared_network::RoutingFilter;
 use futures::StreamExt;
 use nym_mixnet_client::forwarder::{
     mix_forwarding_channels, MixForwardingReceiver, MixForwardingSender, PacketToForward,
@@ -11,14 +13,20 @@ use nym_nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue};
 use nym_sphinx_forwarding::packet::MixPacket;
 use nym_task::ShutdownToken;
 use std::io;
+use std::net::IpAddr;
 use tokio::time::Instant;
 use tracing::{debug, error, trace, warn};
 
+pub(crate) mod global;
+
 pub struct PacketForwarder<C> {
+    testnet: bool,
+
     delay_queue: NonExhaustiveDelayQueue<MixPacket>,
     mixnet_client: C,
 
     metrics: NymNodeMetrics,
+    routing_filter: RoutingFilter,
 
     packet_sender: MixForwardingSender,
     packet_receiver: MixForwardingReceiver,
@@ -26,13 +34,21 @@ pub struct PacketForwarder<C> {
 }
 
 impl<C> PacketForwarder<C> {
-    pub fn new(client: C, metrics: NymNodeMetrics, shutdown: ShutdownToken) -> Self {
+    pub fn new(
+        client: C,
+        testnet: bool,
+        routing_filter: RoutingFilter,
+        metrics: NymNodeMetrics,
+        shutdown: ShutdownToken,
+    ) -> Self {
         let (packet_sender, packet_receiver) = mix_forwarding_channels();
 
         PacketForwarder {
+            testnet,
             delay_queue: NonExhaustiveDelayQueue::new(),
             mixnet_client: client,
             metrics,
+            routing_filter,
             packet_sender,
             packet_receiver,
             shutdown,
@@ -43,11 +59,29 @@ impl<C> PacketForwarder<C> {
         self.packet_sender.clone()
     }
 
+    fn should_route(&mut self, ip_addr: IpAddr) -> bool {
+        // only allow non-global ips on testnets
+        if self.testnet && !is_global_ip(&ip_addr) {
+            return true;
+        }
+
+        self.routing_filter.attempt_resolve(ip_addr).should_route()
+    }
+
     fn forward_packet(&mut self, packet: MixPacket)
     where
         C: SendWithoutResponse,
     {
         let next_hop = packet.next_hop();
+
+        if !self.should_route(next_hop.as_ref().ip()) {
+            debug!("dropping packet as the egress address does not belong to any known node");
+            self.metrics
+                .mixnet
+                .egress_dropped_forward_packet(next_hop.into());
+            return;
+        }
+
         let packet_type = packet.packet_type();
         let packet = packet.into_packet();
 

nym-node/src/node/mod.rs

@@ -28,7 +28,9 @@ use crate::node::metrics::handler::pending_egress_packets_updater::PendingEgress
 use crate::node::mixnet::packet_forwarding::PacketForwarder;
 use crate::node::mixnet::shared::ProcessingConfig;
 use crate::node::mixnet::SharedFinalHopData;
-use crate::node::shared_topology::NymNodeTopologyProvider;
+use crate::node::shared_network::{
+    CachedNetwork, CachedTopologyProvider, NetworkRefresher, RoutingFilter,
+};
 use nym_bin_common::bin_info;
 use nym_crypto::asymmetric::{ed25519, x25519};
 use nym_gateway::node::{ActiveClientsStore, GatewayTasksBuilder};
@@ -67,7 +69,7 @@ pub mod helpers;
 pub(crate) mod http;
 pub(crate) mod metrics;
 pub(crate) mod mixnet;
-mod shared_topology;
+mod shared_network;
 
 pub struct GatewayTasksData {
     mnemonic: Arc<Zeroizing<bip39::Mnemonic>>,
@@ -530,16 +532,15 @@ impl NymNode {
         self.x25519_noise_keys.public_key()
     }
 
-    // the reason it's here as opposed to in the gateway directly,
-    // is that other nym-node tasks will also eventually need it
-    // (such as the ones for obtaining noise keys of other nodes)
-    fn build_topology_provider(&self) -> Result<NymNodeTopologyProvider, NymNodeError> {
-        Ok(NymNodeTopologyProvider::new(
-            self.as_gateway_topology_node()?,
-            self.config.debug.topology_cache_ttl,
+    async fn build_network_refresher(&self) -> Result<NetworkRefresher, NymNodeError> {
+        NetworkRefresher::initialise_new(
             self.user_agent(),
             self.config.mixnet.nym_api_urls.clone(),
-        ))
+            self.config.debug.topology_cache_ttl,
+            self.config.debug.routing_nodes_check_interval,
+            self.shutdown_manager.clone_token("network-refresher"),
+        )
+        .await
     }
 
     fn as_gateway_topology_node(&self) -> Result<nym_topology::RoutingNode, NymNodeError> {
@@ -583,13 +584,19 @@ impl NymNode {
 
     async fn start_gateway_tasks(
         &mut self,
+        cached_network: CachedNetwork,
         metrics_sender: MetricEventsSender,
         active_clients_store: ActiveClientsStore,
         mix_packet_sender: MixForwardingSender,
         task_client: TaskClient,
     ) -> Result<(), NymNodeError> {
         let config = gateway_tasks_config(&self.config);
-        let topology_provider = Box::new(self.build_topology_provider()?);
+
+        let topology_provider = Box::new(CachedTopologyProvider::new(
+            self.as_gateway_topology_node()?,
+            cached_network,
+            self.config.gateway_tasks.debug.minimum_mix_performance,
+        ));
 
         let mut gateway_tasks_builder = GatewayTasksBuilder::new(
             config.gateway,
@@ -952,6 +959,7 @@ impl NymNode {
     pub(crate) fn start_mixnet_listener(
         &self,
         active_clients_store: &ActiveClientsStore,
+        routing_filter: RoutingFilter,
         shutdown: ShutdownToken,
     ) -> (MixForwardingSender, ActiveConnections) {
         let processing_config = ProcessingConfig::new(&self.config);
@@ -980,6 +988,8 @@ impl NymNode {
 
         let mut packet_forwarder = PacketForwarder::new(
             mixnet_client,
+            self.config.debug.testnet,
+            routing_filter,
             self.metrics.clone(),
             shutdown.clone_with_suffix("mix-packet-forwarder"),
         );
@@ -1028,13 +1038,14 @@ impl NymNode {
         });
 
         self.try_refresh_remote_nym_api_cache().await;
-
         self.start_verloc_measurements();
 
+        let network_refresher = self.build_network_refresher().await?;
         let active_clients_store = ActiveClientsStore::new();
 
         let (mix_packet_sender, active_egress_mixnet_connections) = self.start_mixnet_listener(
             &active_clients_store,
+            network_refresher.routing_filter(),
             self.shutdown_manager.clone_token("mixnet-traffic"),
         );
 
@@ -1045,6 +1056,7 @@ impl NymNode {
         );
 
         self.start_gateway_tasks(
+            network_refresher.cached_network(),
             metrics_sender,
             active_clients_store,
             mix_packet_sender,
@@ -1052,6 +1064,8 @@ impl NymNode {
         )
         .await?;
 
+        network_refresher.start();
+
         self.shutdown_manager.close();
         self.shutdown_manager.wait_for_shutdown_signal().await;
 

nym-node/src/node/shared_network.rs

@@ -0,0 +1,434 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::error::NymNodeError;
+use arc_swap::ArcSwap;
+use async_trait::async_trait;
+use nym_gateway::node::UserAgent;
+use nym_node_metrics::prometheus_wrapper::{PrometheusMetric, PROMETHEUS_METRICS};
+use nym_task::ShutdownToken;
+use nym_topology::node::RoutingNode;
+use nym_topology::{EpochRewardedSet, NymTopology, Role, TopologyProvider};
+use nym_validator_client::nym_nodes::{NodesByAddressesResponse, SkimmedNode};
+use nym_validator_client::{NymApiClient, ValidatorClientError};
+use std::collections::HashSet;
+use std::net::IpAddr;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::RwLock;
+use tokio::time::interval;
+use tracing::log::error;
+use tracing::{debug, trace, warn};
+use url::Url;
+
+#[derive(Clone)]
+pub(crate) struct RoutingFilter {
+    resolved: KnownNodes,
+
+    // while this is technically behind a lock, it should not be called too often as once resolved it will
+    // be present on the arcswap in either allowed or denied section
+    pending: UnknownNodes,
+}
+
+impl RoutingFilter {
+    fn new_empty() -> Self {
+        RoutingFilter {
+            resolved: Default::default(),
+            pending: Default::default(),
+        }
+    }
+
+    pub(crate) fn attempt_resolve(&self, ip: IpAddr) -> Resolution {
+        if self.resolved.inner.allowed.load().contains(&ip) {
+            Resolution::Accept
+        } else if self.resolved.inner.denied.load().contains(&ip) {
+            Resolution::Deny
+        } else {
+            self.pending.try_insert(ip);
+            Resolution::Unknown
+        }
+    }
+}
+
+#[derive(Clone, Default)]
+struct UnknownNodes(Arc<RwLock<HashSet<IpAddr>>>);
+
+impl UnknownNodes {
+    fn try_insert(&self, ip: IpAddr) {
+        // if we can immediately grab the lock to push it into the pending queue, amazing, let's do it
+        // otherwise we can do it next time we see this ip
+        // (if we can't hold the lock, it means it's being updated at this very moment which is actually a good thing)
+        if let Ok(mut guard) = self.0.try_write() {
+            guard.insert(ip);
+        }
+    }
+
+    async fn clear(&self) {
+        self.0.write().await.clear();
+    }
+
+    async fn nodes(&self) -> HashSet<IpAddr> {
+        self.0.read().await.clone()
+    }
+}
+
+// for now we don't care about keys, etc.
+// we only want to know if given ip belongs to a known node
+#[derive(Debug, Default, Clone)]
+pub(crate) struct KnownNodes {
+    inner: Arc<KnownNodesInner>,
+}
+
+#[derive(Debug, Default)]
+struct KnownNodesInner {
+    allowed: ArcSwap<HashSet<IpAddr>>,
+    denied: ArcSwap<HashSet<IpAddr>>,
+}
+
+pub(crate) enum Resolution {
+    Unknown,
+    Deny,
+    Accept,
+}
+
+impl From<bool> for Resolution {
+    fn from(value: bool) -> Self {
+        if value {
+            Resolution::Accept
+        } else {
+            Resolution::Deny
+        }
+    }
+}
+
+impl Resolution {
+    pub(crate) fn should_route(&self) -> bool {
+        matches!(self, Resolution::Accept)
+    }
+}
+
+impl KnownNodes {
+    fn swap_allowed(&self, new: HashSet<IpAddr>) {
+        self.inner.allowed.store(Arc::new(new))
+    }
+
+    fn swap_denied(&self, new: HashSet<IpAddr>) {
+        self.inner.denied.store(Arc::new(new))
+    }
+}
+
+struct NodesQuerier {
+    client: NymApiClient,
+    nym_api_urls: Vec<Url>,
+    currently_used_api: usize,
+}
+
+impl NodesQuerier {
+    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.client
+            .change_nym_api(self.nym_api_urls[self.currently_used_api].clone())
+    }
+
+    async fn rewarded_set(&mut self) -> Result<EpochRewardedSet, ValidatorClientError> {
+        let res = self
+            .client
+            .get_current_rewarded_set()
+            .await
+            .inspect_err(|err| error!("failed to get current rewarded set: {err}"));
+
+        if res.is_err() {
+            self.use_next_nym_api()
+        }
+        res
+    }
+
+    async fn current_nymnodes(&mut self) -> Result<Vec<SkimmedNode>, ValidatorClientError> {
+        let res = self
+            .client
+            .get_all_basic_nodes()
+            .await
+            .inspect_err(|err| error!("failed to get network nodes: {err}"));
+
+        if res.is_err() {
+            self.use_next_nym_api()
+        }
+        res
+    }
+
+    async fn query_for_info(
+        &mut self,
+        ips: Vec<IpAddr>,
+    ) -> Result<NodesByAddressesResponse, ValidatorClientError> {
+        let res = self
+            .client
+            .nodes_by_addresses(ips)
+            .await
+            .inspect_err(|err| error!("failed to obtain node information: {err}"));
+
+        if res.is_err() {
+            self.use_next_nym_api()
+        }
+        res
+    }
+}
+
+#[derive(Clone)]
+pub struct CachedTopologyProvider {
+    gateway_node: Arc<RoutingNode>,
+    cached_network: CachedNetwork,
+    min_mix_performance: u8,
+}
+
+impl CachedTopologyProvider {
+    pub(crate) fn new(
+        gateway_node: RoutingNode,
+        cached_network: CachedNetwork,
+        min_mix_performance: u8,
+    ) -> Self {
+        CachedTopologyProvider {
+            gateway_node: Arc::new(gateway_node),
+            cached_network,
+            min_mix_performance,
+        }
+    }
+}
+
+#[async_trait]
+impl TopologyProvider for CachedTopologyProvider {
+    async fn get_new_topology(&mut self) -> Option<NymTopology> {
+        let network_guard = self.cached_network.inner.read().await;
+        let self_node = self.gateway_node.identity_key;
+
+        let mut topology = NymTopology::new_empty(network_guard.rewarded_set.clone())
+            .with_additional_nodes(network_guard.network_nodes.iter().filter(|node| {
+                if node.supported_roles.mixnode {
+                    node.performance.round_to_integer() >= self.min_mix_performance
+                } else {
+                    true
+                }
+            }));
+
+        if !topology.has_node_details(self.gateway_node.node_id) {
+            debug!("{self_node} didn't exist in topology. inserting it.",);
+            topology.insert_node_details(self.gateway_node.as_ref().clone());
+        }
+        topology.force_set_active(self.gateway_node.node_id, Role::EntryGateway);
+
+        Some(topology)
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct CachedNetwork {
+    inner: Arc<RwLock<CachedNetworkInner>>,
+}
+
+impl CachedNetwork {
+    fn new_empty() -> Self {
+        CachedNetwork {
+            inner: Arc::new(RwLock::new(CachedNetworkInner {
+                rewarded_set: Default::default(),
+                network_nodes: vec![],
+            })),
+        }
+    }
+}
+
+struct CachedNetworkInner {
+    rewarded_set: EpochRewardedSet,
+    network_nodes: Vec<SkimmedNode>,
+}
+
+pub struct NetworkRefresher {
+    querier: NodesQuerier,
+    full_refresh_interval: Duration,
+    pending_check_interval: Duration,
+    shutdown_token: ShutdownToken,
+
+    network: CachedNetwork,
+    routing_filter: RoutingFilter,
+}
+
+impl NetworkRefresher {
+    pub(crate) async fn initialise_new(
+        user_agent: UserAgent,
+        nym_api_urls: Vec<Url>,
+        full_refresh_interval: Duration,
+        pending_check_interval: Duration,
+        shutdown_token: ShutdownToken,
+    ) -> Result<Self, NymNodeError> {
+        let mut this = NetworkRefresher {
+            querier: NodesQuerier {
+                client: NymApiClient::new_with_user_agent(nym_api_urls[0].clone(), user_agent),
+                nym_api_urls,
+                currently_used_api: 0,
+            },
+            full_refresh_interval,
+            pending_check_interval,
+            shutdown_token,
+            network: CachedNetwork::new_empty(),
+            routing_filter: RoutingFilter::new_empty(),
+        };
+
+        this.obtain_initial_network().await?;
+        Ok(this)
+    }
+
+    fn allowed_nodes_copy(&self) -> HashSet<IpAddr> {
+        self.routing_filter
+            .resolved
+            .inner
+            .allowed
+            .load_full()
+            .as_ref()
+            .clone()
+    }
+
+    fn denied_nodes_copy(&self) -> HashSet<IpAddr> {
+        self.routing_filter
+            .resolved
+            .inner
+            .denied
+            .load_full()
+            .as_ref()
+            .clone()
+    }
+
+    async fn inspect_pending(&mut self) {
+        let to_resolve = self.routing_filter.pending.nodes().await;
+
+        // no pending requests to resolve
+        if to_resolve.is_empty() {
+            return;
+        }
+
+        let mut allowed = self.allowed_nodes_copy();
+        let mut denied = self.denied_nodes_copy();
+
+        // short circuit: check if the pending nodes are not already resolved
+        // (it could happen due to lack of full sync between pending lock and arcswap(s))
+        if to_resolve
+            .iter()
+            .all(|p| allowed.contains(p) || denied.contains(p))
+        {
+            return;
+        }
+
+        // 1. attempt to use the new nym-api query to get information just by ips
+        let nodes = to_resolve.into_iter().collect();
+        if let Ok(res) = self.querier.query_for_info(nodes).await {
+            for (ip, maybe_id) in res.existence {
+                if maybe_id.is_some() {
+                    allowed.insert(ip);
+                } else {
+                    denied.insert(ip);
+                }
+            }
+
+            self.routing_filter.resolved.swap_allowed(allowed);
+            self.routing_filter.resolved.swap_denied(denied);
+            self.routing_filter.pending.clear().await;
+            return;
+        }
+
+        // 2. we assume nym-api doesn't support that query yet - we have to do the full refresh
+        self.refresh_network_nodes().await;
+    }
+
+    async fn refresh_network_nodes_inner(&mut self) -> Result<(), ValidatorClientError> {
+        let rewarded_set = self.querier.rewarded_set().await?;
+        let nodes = self.querier.current_nymnodes().await?;
+
+        // collect all known/allowed nodes information
+        let known_nodes = nodes
+            .iter()
+            .flat_map(|n| n.ip_addresses.iter())
+            .copied()
+            .collect::<HashSet<_>>();
+
+        let pending = self.routing_filter.pending.nodes().await;
+        let mut current_denied = self.denied_nodes_copy();
+
+        for allowed in &known_nodes {
+            // if some node has become known, it should be removed from the denied set
+            current_denied.remove(allowed);
+        }
+
+        // any pending node, if not present in the new set of allowed nodes, should be added in the denied set
+        for pending_node in pending {
+            if !known_nodes.contains(&pending_node) {
+                current_denied.insert(pending_node);
+            }
+        }
+
+        self.routing_filter.resolved.swap_allowed(known_nodes);
+        self.routing_filter.resolved.swap_denied(current_denied);
+        self.routing_filter.pending.clear().await;
+
+        let mut network_guard = self.network.inner.write().await;
+        network_guard.network_nodes = nodes;
+        network_guard.rewarded_set = rewarded_set;
+
+        Ok(())
+    }
+
+    async fn refresh_network_nodes(&mut self) {
+        let timer =
+            PROMETHEUS_METRICS.start_timer(PrometheusMetric::ProcessTopologyQueryResolutionLatency);
+
+        if self.refresh_network_nodes_inner().await.is_err() {
+            // don't use the histogram observation as some queries didn't complete
+            if let Some(obs) = timer {
+                obs.stop_and_discard();
+            }
+        }
+    }
+
+    pub(crate) async fn obtain_initial_network(&mut self) -> Result<(), NymNodeError> {
+        self.refresh_network_nodes_inner()
+            .await
+            .map_err(|source| NymNodeError::InitialTopologyQueryFailure { source })
+    }
+
+    pub(crate) fn routing_filter(&self) -> RoutingFilter {
+        self.routing_filter.clone()
+    }
+
+    pub(crate) fn cached_network(&self) -> CachedNetwork {
+        self.network.clone()
+    }
+
+    pub(crate) async fn run(&mut self) {
+        let mut full_refresh_interval = interval(self.full_refresh_interval);
+        full_refresh_interval.reset();
+
+        let mut pending_check_interval = interval(self.pending_check_interval);
+        pending_check_interval.reset();
+
+        while !self.shutdown_token.is_cancelled() {
+            tokio::select! {
+                biased;
+                _ = self.shutdown_token.cancelled() => {
+                   trace!("NetworkRefresher: Received shutdown");
+                }
+                _ = pending_check_interval.tick() => {
+                    self.inspect_pending().await;
+                }
+                _ = full_refresh_interval.tick() => {
+                    self.refresh_network_nodes().await;
+                }
+            }
+        }
+        trace!("NetworkRefresher: Exiting");
+    }
+
+    pub(crate) fn start(mut self) {
+        tokio::spawn(async move { self.run().await });
+    }
+}

nym-node/src/node/shared_topology.rs

@@ -1,106 +0,0 @@
-// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: GPL-3.0-only
-
-use async_trait::async_trait;
-use nym_gateway::node::{NymApiTopologyProvider, NymApiTopologyProviderConfig, UserAgent};
-use nym_node_metrics::prometheus_wrapper::{PrometheusMetric, PROMETHEUS_METRICS};
-use nym_topology::node::RoutingNode;
-use nym_topology::{NymTopology, Role, TopologyProvider};
-use std::sync::Arc;
-use std::time::Duration;
-use time::OffsetDateTime;
-use tokio::sync::Mutex;
-use tracing::debug;
-use url::Url;
-// I wouldn't be surprised if this became the start of the node topology cache
-
-#[derive(Clone)]
-pub struct NymNodeTopologyProvider {
-    inner: Arc<Mutex<NymNodeTopologyProviderInner>>,
-}
-
-impl NymNodeTopologyProvider {
-    pub fn new(
-        gateway_node: RoutingNode,
-        cache_ttl: Duration,
-        user_agent: UserAgent,
-        nym_api_url: Vec<Url>,
-    ) -> NymNodeTopologyProvider {
-        NymNodeTopologyProvider {
-            inner: Arc::new(Mutex::new(NymNodeTopologyProviderInner {
-                inner: NymApiTopologyProvider::new(
-                    NymApiTopologyProviderConfig {
-                        min_mixnode_performance: 50,
-                        min_gateway_performance: 0,
-                        use_extended_topology: false,
-                        ignore_egress_epoch_role: true,
-                    },
-                    nym_api_url,
-                    Some(user_agent),
-                ),
-                cache_ttl,
-                cached_at: OffsetDateTime::UNIX_EPOCH,
-                cached: None,
-                gateway_node,
-            })),
-        }
-    }
-}
-
-struct NymNodeTopologyProviderInner {
-    inner: NymApiTopologyProvider,
-    cache_ttl: Duration,
-    cached_at: OffsetDateTime,
-    cached: Option<NymTopology>,
-    gateway_node: RoutingNode,
-}
-
-impl NymNodeTopologyProviderInner {
-    fn cached_topology(&self) -> Option<NymTopology> {
-        if let Some(cached_topology) = &self.cached {
-            if self.cached_at + self.cache_ttl > OffsetDateTime::now_utc() {
-                return Some(cached_topology.clone());
-            }
-        }
-
-        None
-    }
-
-    async fn update_cache(&mut self) -> Option<NymTopology> {
-        let updated_cache = match self.inner.get_new_topology().await {
-            None => None,
-            Some(mut base) => {
-                if !base.has_node_details(self.gateway_node.node_id) {
-                    debug!(
-                        "{} didn't exist in topology. inserting it.",
-                        self.gateway_node.identity_key
-                    );
-                    base.insert_node_details(self.gateway_node.clone());
-                }
-                base.force_set_active(self.gateway_node.node_id, Role::EntryGateway);
-                Some(base)
-            }
-        };
-
-        self.cached_at = OffsetDateTime::now_utc();
-        self.cached = updated_cache.clone();
-
-        updated_cache
-    }
-}
-
-#[async_trait]
-impl TopologyProvider for NymNodeTopologyProvider {
-    async fn get_new_topology(&mut self) -> Option<NymTopology> {
-        let mut guard = self.inner.lock().await;
-        // check the cache
-        if let Some(cached) = guard.cached_topology() {
-            return Some(cached);
-        }
-
-        // the observation will be included on drop
-        let _timer =
-            PROMETHEUS_METRICS.start_timer(PrometheusMetric::ProcessTopologyQueryResolutionLatency);
-        guard.update_cache().await
-    }
-}