[mixnode] using the ingress filter for incoming connections

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<Url>,
 
-    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),
         }
     }

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

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 =