From 527fc5dfdd5146dc49e35491da395c4e9abd34ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 26 Oct 2023 15:53:22 +0100 Subject: [PATCH] [mixnode] using the ingress filter for incoming connections --- .../mixnode-common/src/forward_travel/mod.rs | 16 +++++++------- mixnode/src/node/listener/mod.rs | 22 +++++++++++++++++-- mixnode/src/node/mod.rs | 12 +++++++--- 3 files changed, 37 insertions(+), 13 deletions(-) diff --git a/common/mixnode-common/src/forward_travel/mod.rs b/common/mixnode-common/src/forward_travel/mod.rs index b00d40ac5d..7e4545f985 100644 --- a/common/mixnode-common/src/forward_travel/mod.rs +++ b/common/mixnode-common/src/forward_travel/mod.rs @@ -22,8 +22,8 @@ use url::Url; pub mod error; -pub type Ingress = AllowedPaths; -pub type Egress = AllowedPaths; +pub type AllowedIngress = AllowedPaths; +pub type AllowedEgress = AllowedPaths; pub struct AllowedAddressesProvider { current_epoch: EpochId, @@ -35,8 +35,8 @@ pub struct AllowedAddressesProvider { /// URLs to the nyxd validators for obtaining unfiltered network topology. nyxd_endpoints: Vec, - ingress: Ingress, - egress: Egress, + ingress: AllowedIngress, + egress: AllowedEgress, } impl AllowedAddressesProvider { @@ -81,14 +81,14 @@ impl AllowedAddressesProvider { Err(last_error.into()) } - pub fn ingress(&self) -> Ingress { - Ingress { + pub fn ingress(&self) -> AllowedIngress { + AllowedIngress { inner: Arc::clone(&self.ingress.inner), } } - pub fn egress(&self) -> Egress { - Egress { + pub fn egress(&self) -> AllowedEgress { + AllowedEgress { inner: Arc::clone(&self.egress.inner), } } diff --git a/mixnode/src/node/listener/mod.rs b/mixnode/src/node/listener/mod.rs index 995a6e7c73..be6806bff6 100644 --- a/mixnode/src/node/listener/mod.rs +++ b/mixnode/src/node/listener/mod.rs @@ -3,6 +3,7 @@ use crate::node::listener::connection_handler::ConnectionHandler; use log::{error, info, warn}; +use nym_mixnode_common::forward_travel::AllowedIngress; use std::net::SocketAddr; use std::process; use tokio::net::TcpListener; @@ -14,12 +15,21 @@ pub(crate) mod connection_handler; pub(crate) struct Listener { address: SocketAddr, + allowed_ingress: AllowedIngress, shutdown: TaskClient, } impl Listener { - pub(crate) fn new(address: SocketAddr, shutdown: TaskClient) -> Self { - Listener { address, shutdown } + pub(crate) fn new( + address: SocketAddr, + allowed_ingress: AllowedIngress, + shutdown: TaskClient, + ) -> Self { + Listener { + address, + allowed_ingress, + shutdown, + } } async fn run(&mut self, connection_handler: ConnectionHandler) { @@ -41,6 +51,14 @@ impl Listener { connection = listener.accept() => { match connection { Ok((socket, remote_addr)) => { + // even though this is an async method, we shouldn't be blocked for too long + // since we're just getting a read permit to rwlock, + // while the write task should only be running every hour or so (epoch length) + if !self.allowed_ingress.is_allowed(remote_addr.ip()).await { + warn!("received an incoming connection from {remote_addr}, but this address does not belong to any node on the previous layer - dropping the connection"); + continue + } + let handler = connection_handler.clone(); tokio::spawn(handler.handle_connection(socket, remote_addr, self.shutdown.clone())); } diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 9db96b2848..aa3b6c68a2 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -17,7 +17,7 @@ use nym_bin_common::output_format::OutputFormat; use nym_bin_common::version_checker::parse_version; use nym_config::defaults::NymNetworkDetails; use nym_crypto::asymmetric::{encryption, identity}; -use nym_mixnode_common::forward_travel::{AllowedAddressesProvider, Egress, Ingress}; +use nym_mixnode_common::forward_travel::{AllowedAddressesProvider, AllowedEgress, AllowedIngress}; use nym_mixnode_common::verloc::{self, AtomicVerlocResult, VerlocMeasurer}; use nym_task::{TaskClient, TaskManager}; use rand::seq::SliceRandom; @@ -103,6 +103,7 @@ impl MixNode { &self, node_stats_update_sender: node_statistics::UpdateSender, delay_forwarding_channel: PacketDelayForwardSender, + ingress: AllowedIngress, shutdown: TaskClient, ) { info!("Starting socket listener..."); @@ -117,13 +118,13 @@ impl MixNode { self.config.mixnode.mix_port, ); - Listener::new(listening_address, shutdown).start(connection_handler); + Listener::new(listening_address, ingress, shutdown).start(connection_handler); } fn start_allowed_addresses_provider( &self, task_client: TaskClient, - ) -> Result<(Ingress, Egress), MixnodeError> { + ) -> Result<(AllowedIngress, AllowedEgress), MixnodeError> { let identity = self.identity_keypair.public_key().to_base58_string(); let nyxd_endpoints = self.config.mixnode.nyxd_urls.clone(); @@ -251,6 +252,10 @@ impl MixNode { let shutdown = TaskManager::default(); + let (ingress, egress) = self.start_allowed_addresses_provider( + shutdown.subscribe().named("AllowedAddressesProvider"), + )?; + let (node_stats_pointer, node_stats_update_sender) = self .start_node_stats_controller(shutdown.subscribe().named("node_statistics::Controller")); let delay_forwarding_channel = self.start_packet_delay_forwarder( @@ -260,6 +265,7 @@ impl MixNode { self.start_socket_listener( node_stats_update_sender, delay_forwarding_channel, + ingress, shutdown.subscribe().named("Listener"), ); let atomic_verloc_results =