diff --git a/common/client-libs/validator-client/src/client.rs b/common/client-libs/validator-client/src/client.rs index e5a9a7ea8a..c41ba519d4 100644 --- a/common/client-libs/validator-client/src/client.rs +++ b/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, + ) -> Result { + Ok(self.nym_api.nodes_by_addresses(addresses).await?) + } } diff --git a/common/client-libs/validator-client/src/nym_api/mod.rs b/common/client-libs/validator-client/src/nym_api/mod.rs index e1bac67513..7011a1904a 100644 --- a/common/client-libs/validator-client/src/nym_api/mod.rs +++ b/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, + ) -> Result { + 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 { self.get_json( diff --git a/common/client-libs/validator-client/src/nym_api/routes.rs b/common/client-libs/validator-client/src/nym_api/routes.rs index 767acc0b8d..1dae2254f4 100644 --- a/common/client-libs/validator-client/src/nym_api/routes.rs +++ b/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"; diff --git a/common/config/src/helpers.rs b/common/config/src/helpers.rs index cfed0dc1c7..d1ea90010b 100644 --- a/common/config/src/helpers.rs +++ b/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. diff --git a/common/nymsphinx/addressing/src/nodes.rs b/common/nymsphinx/addressing/src/nodes.rs index c748a72dcd..4171410b6c 100644 --- a/common/nymsphinx/addressing/src/nodes.rs +++ b/common/nymsphinx/addressing/src/nodes.rs @@ -161,6 +161,12 @@ impl From for SocketAddr { } } +impl AsRef for NymNodeRoutingAddress { + fn as_ref(&self) -> &SocketAddr { + &self.0 + } +} + impl TryInto for NymNodeRoutingAddress { type Error = NymNodeRoutingAddressError; diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 2150d587cb..4354721bc5 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -254,6 +254,15 @@ impl NymTopology { } } + pub fn with_additional_nodes(mut self, nodes: impl Iterator) -> Self + where + N: TryInto, + >::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) } diff --git a/nym-api/nym-api-requests/src/nym_nodes.rs b/nym-api/nym-api-requests/src/nym_nodes.rs index 96c981c8a1..544e07a861 100644 --- a/nym-api/nym-api-requests/src/nym_nodes.rs +++ b/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, } + +#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, ToSchema)] +pub struct NodesByAddressesRequestBody { + #[schema(value_type = Vec)] + pub addresses: Vec, +} + +#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, ToSchema)] +pub struct NodesByAddressesResponse { + #[schema(value_type = HashMap>)] + pub existence: HashMap>, +} diff --git a/nym-api/src/node_describe_cache/mod.rs b/nym-api/src/node_describe_cache/mod.rs index 9cecb37e1e..8fa5adf452 100644 --- a/nym-api/src/node_describe_cache/mod.rs +++ b/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, + addresses_cache: HashMap, } 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 { + self.addresses_cache.get(&address).copied() + } } pub struct NodeDescriptionProvider { @@ -396,9 +405,20 @@ impl CacheItemProvider for NodeDescriptionProvider { .collect::>() .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, + }) } } diff --git a/nym-api/src/nym_nodes/handlers/unstable/mod.rs b/nym-api/src/nym_nodes/handlers/unstable/mod.rs index d1ee06ed1e..f858065d2b 100644 --- a/nym-api/src/nym_nodes/handlers/unstable/mod.rs +++ b/nym-api/src/nym_nodes/handlers/unstable/mod.rs @@ -20,6 +20,7 @@ //! - `/mixnodes/` => only returns mixnode role data //! - `/gateway/` => 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 { .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, + Json(body): Json, +) -> AxumResult> { + // 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 })) +} diff --git a/nym-node/src/cli/commands/run/mod.rs b/nym-node/src/cli/commands/run/mod.rs index 9dfd1061f0..8b568150ed 100644 --- a/nym-node/src/cli/commands/run/mod.rs +++ b/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); diff --git a/nym-node/src/config/gateway_tasks.rs b/nym-node/src/config/gateway_tasks.rs index 4d83c4632d..d35a978333 100644 --- a/nym-node/src/config/gateway_tasks.rs +++ b/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(), diff --git a/nym-node/src/config/mod.rs b/nym-node/src/config/mod.rs index b46d19e370..c71f003fbc 100644 --- a/nym-node/src/config/mod.rs +++ b/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, } } } diff --git a/nym-node/src/error.rs b/nym-node/src/error.rs index 3c50451e07..5eb5545e21 100644 --- a/nym-node/src/error.rs +++ b/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), diff --git a/nym-node/src/node/mixnet/packet_forwarding/global.rs b/nym-node/src/node/mixnet/packet_forwarding/global.rs new file mode 100644 index 0000000000..6f27444a57 --- /dev/null +++ b/nym-node/src/node/mixnet/packet_forwarding/global.rs @@ -0,0 +1,79 @@ +// Copyright 2025 - Nym Technologies SA +// 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()) +} diff --git a/nym-node/src/node/mixnet/packet_forwarding.rs b/nym-node/src/node/mixnet/packet_forwarding/mod.rs similarity index 85% rename from nym-node/src/node/mixnet/packet_forwarding.rs rename to nym-node/src/node/mixnet/packet_forwarding/mod.rs index cb85fc5af9..a61f5517ec 100644 --- a/nym-node/src/node/mixnet/packet_forwarding.rs +++ b/nym-node/src/node/mixnet/packet_forwarding/mod.rs @@ -1,6 +1,8 @@ // Copyright 2024 - Nym Technologies SA // 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 { + testnet: bool, + delay_queue: NonExhaustiveDelayQueue, mixnet_client: C, metrics: NymNodeMetrics, + routing_filter: RoutingFilter, packet_sender: MixForwardingSender, packet_receiver: MixForwardingReceiver, @@ -26,13 +34,21 @@ pub struct PacketForwarder { } impl PacketForwarder { - 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 PacketForwarder { 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(); diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index b803d772d0..3a8d7bf867 100644 --- a/nym-node/src/node/mod.rs +++ b/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>, @@ -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 { - Ok(NymNodeTopologyProvider::new( - self.as_gateway_topology_node()?, - self.config.debug.topology_cache_ttl, + async fn build_network_refresher(&self) -> Result { + 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 { @@ -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; diff --git a/nym-node/src/node/shared_network.rs b/nym-node/src/node/shared_network.rs new file mode 100644 index 0000000000..c127aa1586 --- /dev/null +++ b/nym-node/src/node/shared_network.rs @@ -0,0 +1,434 @@ +// Copyright 2024 - Nym Technologies SA +// 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>>); + +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 { + 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, +} + +#[derive(Debug, Default)] +struct KnownNodesInner { + allowed: ArcSwap>, + denied: ArcSwap>, +} + +pub(crate) enum Resolution { + Unknown, + Deny, + Accept, +} + +impl From 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) { + self.inner.allowed.store(Arc::new(new)) + } + + fn swap_denied(&self, new: HashSet) { + self.inner.denied.store(Arc::new(new)) + } +} + +struct NodesQuerier { + client: NymApiClient, + nym_api_urls: Vec, + 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 { + 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, 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, + ) -> Result { + 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, + 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 { + 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>, +} + +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, +} + +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, + full_refresh_interval: Duration, + pending_check_interval: Duration, + shutdown_token: ShutdownToken, + ) -> Result { + 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 { + self.routing_filter + .resolved + .inner + .allowed + .load_full() + .as_ref() + .clone() + } + + fn denied_nodes_copy(&self) -> HashSet { + 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::>(); + + 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 }); + } +} diff --git a/nym-node/src/node/shared_topology.rs b/nym-node/src/node/shared_topology.rs deleted file mode 100644 index f3ba72fd00..0000000000 --- a/nym-node/src/node/shared_topology.rs +++ /dev/null @@ -1,106 +0,0 @@ -// Copyright 2024 - Nym Technologies SA -// 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>, -} - -impl NymNodeTopologyProvider { - pub fn new( - gateway_node: RoutingNode, - cache_ttl: Duration, - user_agent: UserAgent, - nym_api_url: Vec, - ) -> 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, - gateway_node: RoutingNode, -} - -impl NymNodeTopologyProviderInner { - fn cached_topology(&self) -> Option { - 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 { - 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 { - 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 - } -}