From c2321c20eb84d9f0917b30926088553f02e33804 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 08:45:55 +0100 Subject: [PATCH 1/9] Rename to MixnetListener --- service-providers/ip-packet-router/src/ip_packet_router.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/service-providers/ip-packet-router/src/ip_packet_router.rs b/service-providers/ip-packet-router/src/ip_packet_router.rs index c514be4f1b..55b45abe36 100644 --- a/service-providers/ip-packet-router/src/ip_packet_router.rs +++ b/service-providers/ip-packet-router/src/ip_packet_router.rs @@ -158,7 +158,7 @@ impl IpPacketRouterBuilder { let request_filter = request_filter::RequestFilter::new(&self.config).await?; request_filter.start_update_tasks().await; - let ip_packet_router_service = IpPacketRouter { + let ip_packet_router_service = MixnetListener { _config: self.config, request_filter: request_filter.clone(), tun_writer, @@ -186,7 +186,7 @@ impl IpPacketRouterBuilder { } #[cfg(target_os = "linux")] -struct IpPacketRouter { +struct MixnetListener { _config: Config, request_filter: request_filter::RequestFilter, tun_writer: tokio::io::WriteHalf, @@ -203,7 +203,7 @@ pub(crate) struct ConnectedClient { } #[cfg(target_os = "linux")] -impl IpPacketRouter { +impl MixnetListener { async fn on_static_connect_request( &mut self, connect_request: nym_ip_packet_requests::StaticConnectRequest, From 8be372acff68b41ac25cb36983d116f2814b0e84 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 08:59:09 +0100 Subject: [PATCH 2/9] Extract out mixnet_listener.rs --- .../ip-packet-router/src/ip_packet_router.rs | 340 +----------------- service-providers/ip-packet-router/src/lib.rs | 5 +- .../ip-packet-router/src/mixnet_listener.rs | 335 +++++++++++++++++ .../ip-packet-router/src/tun_listener.rs | 12 +- .../src/util/generate_new_ip.rs | 2 +- 5 files changed, 354 insertions(+), 340 deletions(-) create mode 100644 service-providers/ip-packet-router/src/mixnet_listener.rs diff --git a/service-providers/ip-packet-router/src/ip_packet_router.rs b/service-providers/ip-packet-router/src/ip_packet_router.rs index 55b45abe36..15160ccd58 100644 --- a/service-providers/ip-packet-router/src/ip_packet_router.rs +++ b/service-providers/ip-packet-router/src/ip_packet_router.rs @@ -1,29 +1,19 @@ #![cfg_attr(not(target_os = "linux"), allow(dead_code))] #![cfg_attr(not(target_os = "linux"), allow(unused_imports))] -use std::{collections::HashMap, net::IpAddr, path::Path}; +use std::path::Path; -use futures::{channel::oneshot, StreamExt}; +use futures::channel::oneshot; use nym_client_core::{ client::mix_traffic::transceiver::GatewayTransceiver, HardcodedTopologyProvider, TopologyProvider, }; -use nym_ip_packet_requests::{ - DynamicConnectFailureReason, IpPacketRequest, IpPacketRequestData, IpPacketResponse, - StaticConnectFailureReason, -}; -use nym_sdk::mixnet::{InputMessage, MixnetMessageSender, Recipient}; -use nym_sphinx::receiver::ReconstructedMessage; -use nym_task::{connections::TransmissionLane, TaskClient, TaskHandle}; -#[cfg(target_os = "linux")] -use tokio::io::AsyncWriteExt; +use nym_sdk::mixnet::Recipient; +use nym_task::{TaskClient, TaskHandle}; use crate::{ - constants::{CLIENT_INACTIVITY_TIMEOUT, DISCONNECT_TIMER_INTERVAL}, error::IpPacketRouterError, request_filter::{self, RequestFilter}, - util::generate_new_ip, - util::parse_ip::{parse_packet, ParsedPacket}, Config, }; @@ -118,6 +108,8 @@ impl IpPacketRouterBuilder { #[cfg(target_os = "linux")] pub async fn run_service_provider(self) -> Result<(), IpPacketRouterError> { // Used to notify tasks to shutdown. Not all tasks fully supports this (yet). + + use crate::{mixnet_listener, tun_listener}; let task_handle: TaskHandle = self.shutdown.map(Into::into).unwrap_or_default(); // Connect to the mixnet @@ -146,7 +138,7 @@ impl IpPacketRouterBuilder { // TunListener let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel(); - let tun_listener = crate::tun_listener::TunListener { + let tun_listener = tun_listener::TunListener { tun_reader, mixnet_client_sender: mixnet_client.split_sender(), task_client: task_handle.get_handle(), @@ -158,7 +150,7 @@ impl IpPacketRouterBuilder { let request_filter = request_filter::RequestFilter::new(&self.config).await?; request_filter.start_update_tasks().await; - let ip_packet_router_service = MixnetListener { + let mixnet_listener = mixnet_listener::MixnetListener { _config: self.config, request_filter: request_filter.clone(), tun_writer, @@ -181,320 +173,6 @@ impl IpPacketRouterBuilder { } } - ip_packet_router_service.run().await + mixnet_listener.run().await } } - -#[cfg(target_os = "linux")] -struct MixnetListener { - _config: Config, - request_filter: request_filter::RequestFilter, - tun_writer: tokio::io::WriteHalf, - mixnet_client: nym_sdk::mixnet::MixnetClient, - task_handle: TaskHandle, - - connected_clients: HashMap, - connected_client_tx: tokio::sync::mpsc::UnboundedSender, -} - -pub(crate) struct ConnectedClient { - pub(crate) nym_address: Recipient, - pub(crate) last_activity: std::time::Instant, -} - -#[cfg(target_os = "linux")] -impl MixnetListener { - async fn on_static_connect_request( - &mut self, - connect_request: nym_ip_packet_requests::StaticConnectRequest, - ) -> Result, IpPacketRouterError> { - log::info!( - "Received static connect request from {sender_address}", - sender_address = connect_request.reply_to - ); - - let request_id = connect_request.request_id; - let requested_ip = connect_request.ip; - let reply_to = connect_request.reply_to; - // TODO: ignoring reply_to_hops and reply_to_avg_mix_delays for now - - // Check that the IP is available in the set of connected clients - let is_ip_taken = self.connected_clients.contains_key(&requested_ip); - - // Check that the nym address isn't already registered - let is_nym_address_taken = self - .connected_clients - .values() - .any(|client| client.nym_address == reply_to); - - match (is_ip_taken, is_nym_address_taken) { - (true, true) => { - log::info!("Connecting an already connected client"); - // Update the last activity time for the client - if let Some(client) = self.connected_clients.get_mut(&requested_ip) { - client.last_activity = std::time::Instant::now(); - } else { - log::error!("Failed to update last activity time for client"); - } - Ok(Some(IpPacketResponse::new_static_connect_success( - request_id, reply_to, - ))) - } - (false, false) => { - log::info!("Connecting a new client"); - self.connected_clients.insert( - requested_ip, - ConnectedClient { - nym_address: reply_to, - last_activity: std::time::Instant::now(), - }, - ); - self.connected_client_tx - .send(ConnectedClientEvent::Connect( - requested_ip, - Box::new(reply_to), - )) - .unwrap(); - Ok(Some(IpPacketResponse::new_static_connect_success( - request_id, reply_to, - ))) - } - (true, false) => { - log::info!("Requested IP is not available"); - Ok(Some(IpPacketResponse::new_static_connect_failure( - request_id, - reply_to, - StaticConnectFailureReason::RequestedIpAlreadyInUse, - ))) - } - (false, true) => { - log::info!("Nym address is already registered"); - Ok(Some(IpPacketResponse::new_static_connect_failure( - request_id, - reply_to, - StaticConnectFailureReason::RequestedNymAddressAlreadyInUse, - ))) - } - } - } - - async fn on_dynamic_connect_request( - &mut self, - connect_request: nym_ip_packet_requests::DynamicConnectRequest, - ) -> Result, IpPacketRouterError> { - log::info!( - "Received dynamic connect request from {sender_address}", - sender_address = connect_request.reply_to - ); - - let request_id = connect_request.request_id; - let reply_to = connect_request.reply_to; - // TODO: ignoring reply_to_hops and reply_to_avg_mix_delays for now - - // Check if it's the same client connecting again, then we just reuse the same IP - // TODO: this is problematic. Until we sign connect requests this means you can spam people - // with return traffic - let existing_ip = self.connected_clients.iter().find_map(|(ip, client)| { - if client.nym_address == reply_to { - Some(*ip) - } else { - None - } - }); - - if let Some(existing_ip) = existing_ip { - log::info!("Found existing client for nym address"); - // Update the last activity time for the client - if let Some(client) = self.connected_clients.get_mut(&existing_ip) { - client.last_activity = std::time::Instant::now(); - } else { - log::error!("Failed to update last activity time for client"); - } - return Ok(Some(IpPacketResponse::new_dynamic_connect_success( - request_id, - reply_to, - existing_ip, - ))); - } - - let Some(new_ip) = generate_new_ip::find_new_ip(&self.connected_clients) else { - log::info!("No available IP address"); - return Ok(Some(IpPacketResponse::new_dynamic_connect_failure( - request_id, - reply_to, - DynamicConnectFailureReason::NoAvailableIp, - ))); - }; - - self.connected_clients.insert( - new_ip, - ConnectedClient { - nym_address: reply_to, - last_activity: std::time::Instant::now(), - }, - ); - self.connected_client_tx - .send(ConnectedClientEvent::Connect(new_ip, Box::new(reply_to))) - .unwrap(); - Ok(Some(IpPacketResponse::new_dynamic_connect_success( - request_id, reply_to, new_ip, - ))) - } - - async fn on_data_request( - &mut self, - data_request: nym_ip_packet_requests::DataRequest, - ) -> Result, IpPacketRouterError> { - log::trace!("Received data request"); - - // We don't forward packets that we are not able to parse. BUT, there might be a good - // reason to still forward them. - // - // For example, if we are running in a mode where we are only supposed to forward - // packets to a specific destination, we might want to forward them anyway. - // - // TODO: look into this - let ParsedPacket { - packet_type, - src_addr, - dst_addr, - dst, - } = parse_packet(&data_request.ip_packet)?; - - let dst_str = dst.map_or(dst_addr.to_string(), |dst| dst.to_string()); - log::info!("Received packet: {packet_type}: {src_addr} -> {dst_str}"); - - // Check if there is a connected client for this src_addr. If there is, update the last activity time - // for the client. If there isn't, drop the packet. - if let Some(client) = self.connected_clients.get_mut(&src_addr) { - client.last_activity = std::time::Instant::now(); - } else { - log::info!("Dropping packet: no connected client for {src_addr}"); - return Ok(None); - } - - // Filter check - if let Some(dst) = dst { - if !self.request_filter.check_address(&dst).await { - log::warn!("Failed filter check: {dst}"); - // TODO: we could consider sending back a response here - return Err(IpPacketRouterError::AddressFailedFilterCheck { addr: dst }); - } - } else { - // TODO: we should also filter packets without port number - log::warn!("Ignoring filter check for packet without port number! TODO!"); - } - - // TODO: consider changing from Vec to bytes::Bytes? - let packet = data_request.ip_packet; - self.tun_writer - .write_all(&packet) - .await - .map_err(|_| IpPacketRouterError::FailedToWritePacketToTun)?; - - Ok(None) - } - async fn on_reconstructed_message( - &mut self, - reconstructed: ReconstructedMessage, - ) -> Result, IpPacketRouterError> { - log::debug!( - "Received message with sender_tag: {:?}", - reconstructed.sender_tag - ); - - // Check version of request - if let Some(version) = reconstructed.message.first() { - // The idea is that in the future we can add logic here to parse older versions to stay - // backwards compatible. - if *version != nym_ip_packet_requests::CURRENT_VERSION { - log::warn!("Received packet with invalid version"); - return Err(IpPacketRouterError::InvalidPacketVersion(*version)); - } - } - - let request = IpPacketRequest::from_reconstructed_message(&reconstructed) - .map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { source: err })?; - - match request.data { - IpPacketRequestData::StaticConnect(connect_request) => { - self.on_static_connect_request(connect_request).await - } - IpPacketRequestData::DynamicConnect(connect_request) => { - self.on_dynamic_connect_request(connect_request).await - } - IpPacketRequestData::Data(data_request) => self.on_data_request(data_request).await, - } - } - - async fn run(mut self) -> Result<(), IpPacketRouterError> { - let mut task_client = self.task_handle.fork("main_loop"); - let mut disconnect_timer = tokio::time::interval(DISCONNECT_TIMER_INTERVAL); - - while !task_client.is_shutdown() { - tokio::select! { - _ = task_client.recv() => { - log::debug!("IpPacketRouter [main loop]: received shutdown"); - }, - _ = disconnect_timer.tick() => { - let now = std::time::Instant::now(); - let inactive_clients: Vec = self.connected_clients.iter() - .filter_map(|(ip, client)| { - if now.duration_since(client.last_activity) > CLIENT_INACTIVITY_TIMEOUT { - Some(*ip) - } else { - None - } - }) - .collect(); - for ip in inactive_clients { - log::info!("Disconnect inactive client: {ip}"); - self.connected_clients.remove(&ip); - self.connected_client_tx.send(ConnectedClientEvent::Disconnect(ip)).unwrap(); - } - }, - msg = self.mixnet_client.next() => { - if let Some(msg) = msg { - match self.on_reconstructed_message(msg).await { - Ok(Some(response)) => { - let Some(recipient) = response.recipient() else { - log::error!("IpPacketRouter [main loop]: failed to get recipient from response"); - continue; - }; - let response_packet = response.to_bytes(); - let Ok(response_packet) = response_packet else { - log::error!("Failed to serialize response packet"); - continue; - }; - let lane = TransmissionLane::General; - let packet_type = None; - let input_message = InputMessage::new_regular(*recipient, response_packet, lane, packet_type); - if let Err(err) = self.mixnet_client.send(input_message).await { - log::error!("IpPacketRouter [main loop]: failed to send packet to mixnet: {err}"); - }; - }, - Ok(None) => { - continue; - }, - Err(err) => { - log::error!("Error handling mixnet message: {err}"); - } - - }; - } else { - log::trace!("IpPacketRouter [main loop]: stopping since channel closed"); - break; - }; - }, - - } - } - log::debug!("IpPacketRouter: stopping"); - Ok(()) - } -} - -pub(crate) enum ConnectedClientEvent { - Disconnect(IpAddr), - Connect(IpAddr, Box), -} diff --git a/service-providers/ip-packet-router/src/lib.rs b/service-providers/ip-packet-router/src/lib.rs index 7739f4cff6..dcc00c985b 100644 --- a/service-providers/ip-packet-router/src/lib.rs +++ b/service-providers/ip-packet-router/src/lib.rs @@ -4,11 +4,12 @@ pub use crate::config::Config; pub use ip_packet_router::{IpPacketRouterBuilder, OnStartData}; -pub mod config; mod constants; -pub mod error; mod ip_packet_router; mod mixnet_client; +mod mixnet_listener; mod request_filter; mod tun_listener; mod util; +pub mod config; +pub mod error; diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs new file mode 100644 index 0000000000..f4e4b63a52 --- /dev/null +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -0,0 +1,335 @@ +use std::{collections::HashMap, net::IpAddr}; + +use futures::StreamExt; +use nym_ip_packet_requests::{ + DynamicConnectFailureReason, IpPacketRequest, IpPacketRequestData, IpPacketResponse, + StaticConnectFailureReason, +}; +use nym_sdk::mixnet::{InputMessage, MixnetMessageSender, Recipient}; +use nym_sphinx::receiver::ReconstructedMessage; +use nym_task::{connections::TransmissionLane, TaskHandle}; +#[cfg(target_os = "linux")] +use tokio::io::AsyncWriteExt; + +use crate::{ + constants::{CLIENT_INACTIVITY_TIMEOUT, DISCONNECT_TIMER_INTERVAL}, + error::IpPacketRouterError, + request_filter::{self}, + util::generate_new_ip, + util::parse_ip::{parse_packet, ParsedPacket}, + Config, +}; + +#[cfg(target_os = "linux")] +pub(crate) struct MixnetListener { + pub(crate) _config: Config, + pub(crate) request_filter: request_filter::RequestFilter, + pub(crate) tun_writer: tokio::io::WriteHalf, + pub(crate) mixnet_client: nym_sdk::mixnet::MixnetClient, + pub(crate) task_handle: TaskHandle, + + pub(crate) connected_clients: HashMap, + pub(crate) connected_client_tx: tokio::sync::mpsc::UnboundedSender, +} + +pub(crate) struct ConnectedClient { + pub(crate) nym_address: Recipient, + pub(crate) last_activity: std::time::Instant, +} + +#[cfg(target_os = "linux")] +impl MixnetListener { + async fn on_static_connect_request( + &mut self, + connect_request: nym_ip_packet_requests::StaticConnectRequest, + ) -> Result, IpPacketRouterError> { + log::info!( + "Received static connect request from {sender_address}", + sender_address = connect_request.reply_to + ); + + let request_id = connect_request.request_id; + let requested_ip = connect_request.ip; + let reply_to = connect_request.reply_to; + // TODO: ignoring reply_to_hops and reply_to_avg_mix_delays for now + + // Check that the IP is available in the set of connected clients + let is_ip_taken = self.connected_clients.contains_key(&requested_ip); + + // Check that the nym address isn't already registered + let is_nym_address_taken = self + .connected_clients + .values() + .any(|client| client.nym_address == reply_to); + + match (is_ip_taken, is_nym_address_taken) { + (true, true) => { + log::info!("Connecting an already connected client"); + // Update the last activity time for the client + if let Some(client) = self.connected_clients.get_mut(&requested_ip) { + client.last_activity = std::time::Instant::now(); + } else { + log::error!("Failed to update last activity time for client"); + } + Ok(Some(IpPacketResponse::new_static_connect_success( + request_id, reply_to, + ))) + } + (false, false) => { + log::info!("Connecting a new client"); + self.connected_clients.insert( + requested_ip, + ConnectedClient { + nym_address: reply_to, + last_activity: std::time::Instant::now(), + }, + ); + self.connected_client_tx + .send(ConnectedClientEvent::Connect( + requested_ip, + Box::new(reply_to), + )) + .unwrap(); + Ok(Some(IpPacketResponse::new_static_connect_success( + request_id, reply_to, + ))) + } + (true, false) => { + log::info!("Requested IP is not available"); + Ok(Some(IpPacketResponse::new_static_connect_failure( + request_id, + reply_to, + StaticConnectFailureReason::RequestedIpAlreadyInUse, + ))) + } + (false, true) => { + log::info!("Nym address is already registered"); + Ok(Some(IpPacketResponse::new_static_connect_failure( + request_id, + reply_to, + StaticConnectFailureReason::RequestedNymAddressAlreadyInUse, + ))) + } + } + } + + async fn on_dynamic_connect_request( + &mut self, + connect_request: nym_ip_packet_requests::DynamicConnectRequest, + ) -> Result, IpPacketRouterError> { + log::info!( + "Received dynamic connect request from {sender_address}", + sender_address = connect_request.reply_to + ); + + let request_id = connect_request.request_id; + let reply_to = connect_request.reply_to; + // TODO: ignoring reply_to_hops and reply_to_avg_mix_delays for now + + // Check if it's the same client connecting again, then we just reuse the same IP + // TODO: this is problematic. Until we sign connect requests this means you can spam people + // with return traffic + let existing_ip = self.connected_clients.iter().find_map(|(ip, client)| { + if client.nym_address == reply_to { + Some(*ip) + } else { + None + } + }); + + if let Some(existing_ip) = existing_ip { + log::info!("Found existing client for nym address"); + // Update the last activity time for the client + if let Some(client) = self.connected_clients.get_mut(&existing_ip) { + client.last_activity = std::time::Instant::now(); + } else { + log::error!("Failed to update last activity time for client"); + } + return Ok(Some(IpPacketResponse::new_dynamic_connect_success( + request_id, + reply_to, + existing_ip, + ))); + } + + let Some(new_ip) = generate_new_ip::find_new_ip(&self.connected_clients) else { + log::info!("No available IP address"); + return Ok(Some(IpPacketResponse::new_dynamic_connect_failure( + request_id, + reply_to, + DynamicConnectFailureReason::NoAvailableIp, + ))); + }; + + self.connected_clients.insert( + new_ip, + ConnectedClient { + nym_address: reply_to, + last_activity: std::time::Instant::now(), + }, + ); + self.connected_client_tx + .send(ConnectedClientEvent::Connect(new_ip, Box::new(reply_to))) + .unwrap(); + Ok(Some(IpPacketResponse::new_dynamic_connect_success( + request_id, reply_to, new_ip, + ))) + } + + async fn on_data_request( + &mut self, + data_request: nym_ip_packet_requests::DataRequest, + ) -> Result, IpPacketRouterError> { + log::trace!("Received data request"); + + // We don't forward packets that we are not able to parse. BUT, there might be a good + // reason to still forward them. + // + // For example, if we are running in a mode where we are only supposed to forward + // packets to a specific destination, we might want to forward them anyway. + // + // TODO: look into this + let ParsedPacket { + packet_type, + src_addr, + dst_addr, + dst, + } = parse_packet(&data_request.ip_packet)?; + + let dst_str = dst.map_or(dst_addr.to_string(), |dst| dst.to_string()); + log::info!("Received packet: {packet_type}: {src_addr} -> {dst_str}"); + + // Check if there is a connected client for this src_addr. If there is, update the last activity time + // for the client. If there isn't, drop the packet. + if let Some(client) = self.connected_clients.get_mut(&src_addr) { + client.last_activity = std::time::Instant::now(); + } else { + log::info!("Dropping packet: no connected client for {src_addr}"); + return Ok(None); + } + + // Filter check + if let Some(dst) = dst { + if !self.request_filter.check_address(&dst).await { + log::warn!("Failed filter check: {dst}"); + // TODO: we could consider sending back a response here + return Err(IpPacketRouterError::AddressFailedFilterCheck { addr: dst }); + } + } else { + // TODO: we should also filter packets without port number + log::warn!("Ignoring filter check for packet without port number! TODO!"); + } + + // TODO: consider changing from Vec to bytes::Bytes? + let packet = data_request.ip_packet; + self.tun_writer + .write_all(&packet) + .await + .map_err(|_| IpPacketRouterError::FailedToWritePacketToTun)?; + + Ok(None) + } + async fn on_reconstructed_message( + &mut self, + reconstructed: ReconstructedMessage, + ) -> Result, IpPacketRouterError> { + log::debug!( + "Received message with sender_tag: {:?}", + reconstructed.sender_tag + ); + + // Check version of request + if let Some(version) = reconstructed.message.first() { + // The idea is that in the future we can add logic here to parse older versions to stay + // backwards compatible. + if *version != nym_ip_packet_requests::CURRENT_VERSION { + log::warn!("Received packet with invalid version"); + return Err(IpPacketRouterError::InvalidPacketVersion(*version)); + } + } + + let request = IpPacketRequest::from_reconstructed_message(&reconstructed) + .map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { source: err })?; + + match request.data { + IpPacketRequestData::StaticConnect(connect_request) => { + self.on_static_connect_request(connect_request).await + } + IpPacketRequestData::DynamicConnect(connect_request) => { + self.on_dynamic_connect_request(connect_request).await + } + IpPacketRequestData::Data(data_request) => self.on_data_request(data_request).await, + } + } + + pub(crate) async fn run(mut self) -> Result<(), IpPacketRouterError> { + let mut task_client = self.task_handle.fork("main_loop"); + let mut disconnect_timer = tokio::time::interval(DISCONNECT_TIMER_INTERVAL); + + while !task_client.is_shutdown() { + tokio::select! { + _ = task_client.recv() => { + log::debug!("IpPacketRouter [main loop]: received shutdown"); + }, + _ = disconnect_timer.tick() => { + let now = std::time::Instant::now(); + let inactive_clients: Vec = self.connected_clients.iter() + .filter_map(|(ip, client)| { + if now.duration_since(client.last_activity) > CLIENT_INACTIVITY_TIMEOUT { + Some(*ip) + } else { + None + } + }) + .collect(); + for ip in inactive_clients { + log::info!("Disconnect inactive client: {ip}"); + self.connected_clients.remove(&ip); + self.connected_client_tx.send(ConnectedClientEvent::Disconnect(ip)).unwrap(); + } + }, + msg = self.mixnet_client.next() => { + if let Some(msg) = msg { + match self.on_reconstructed_message(msg).await { + Ok(Some(response)) => { + let Some(recipient) = response.recipient() else { + log::error!("IpPacketRouter [main loop]: failed to get recipient from response"); + continue; + }; + let response_packet = response.to_bytes(); + let Ok(response_packet) = response_packet else { + log::error!("Failed to serialize response packet"); + continue; + }; + let lane = TransmissionLane::General; + let packet_type = None; + let input_message = InputMessage::new_regular(*recipient, response_packet, lane, packet_type); + if let Err(err) = self.mixnet_client.send(input_message).await { + log::error!("IpPacketRouter [main loop]: failed to send packet to mixnet: {err}"); + }; + }, + Ok(None) => { + continue; + }, + Err(err) => { + log::error!("Error handling mixnet message: {err}"); + } + + }; + } else { + log::trace!("IpPacketRouter [main loop]: stopping since channel closed"); + break; + }; + }, + + } + } + log::debug!("IpPacketRouter: stopping"); + Ok(()) + } +} + +pub(crate) enum ConnectedClientEvent { + Disconnect(IpAddr), + Connect(IpAddr, Box), +} diff --git a/service-providers/ip-packet-router/src/tun_listener.rs b/service-providers/ip-packet-router/src/tun_listener.rs index 93a16d316d..15e09c13b5 100644 --- a/service-providers/ip-packet-router/src/tun_listener.rs +++ b/service-providers/ip-packet-router/src/tun_listener.rs @@ -7,7 +7,7 @@ use nym_task::{connections::TransmissionLane, TaskClient}; use tokio::io::AsyncReadExt; use tokio::sync::mpsc::UnboundedReceiver; -use crate::{error::IpPacketRouterError, ip_packet_router, util::parse_ip::parse_dst_addr}; +use crate::{error::IpPacketRouterError, mixnet_listener, util::parse_ip::parse_dst_addr}; // Reads packet from TUN and writes to mixnet client #[cfg(target_os = "linux")] @@ -17,8 +17,8 @@ pub(crate) struct TunListener { pub(crate) task_client: TaskClient, // A mirror of the one in IpPacketRouter - pub(crate) connected_clients: HashMap, - pub(crate) connected_client_rx: UnboundedReceiver, + pub(crate) connected_clients: HashMap, + pub(crate) connected_client_rx: UnboundedReceiver, } #[cfg(target_os = "linux")] @@ -31,14 +31,14 @@ impl TunListener { log::trace!("TunListener: received shutdown"); }, event = self.connected_client_rx.recv() => match event { - Some(ip_packet_router::ConnectedClientEvent::Connect(ip, nym_addr)) => { + Some(mixnet_listener::ConnectedClientEvent::Connect(ip, nym_addr)) => { log::trace!("Connect client: {ip}"); - self.connected_clients.insert(ip, ip_packet_router::ConnectedClient { + self.connected_clients.insert(ip, mixnet_listener::ConnectedClient { nym_address: *nym_addr, last_activity: std::time::Instant::now(), }); }, - Some(ip_packet_router::ConnectedClientEvent::Disconnect(ip)) => { + Some(mixnet_listener::ConnectedClientEvent::Disconnect(ip)) => { log::trace!("Disconnect client: {ip}"); self.connected_clients.remove(&ip); }, diff --git a/service-providers/ip-packet-router/src/util/generate_new_ip.rs b/service-providers/ip-packet-router/src/util/generate_new_ip.rs index 54aa175076..5824ce7bba 100644 --- a/service-providers/ip-packet-router/src/util/generate_new_ip.rs +++ b/service-providers/ip-packet-router/src/util/generate_new_ip.rs @@ -3,7 +3,7 @@ use std::{ net::{IpAddr, Ipv4Addr}, }; -use crate::{constants::TUN_DEVICE_ADDRESS, ip_packet_router::ConnectedClient}; +use crate::{constants::TUN_DEVICE_ADDRESS, mixnet_listener::ConnectedClient}; // Find an available IP address in self.connected_clients // TODO: make this nicer From a190506b41301472bf82707117eb4f11e8292b51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 09:04:17 +0100 Subject: [PATCH 3/9] Rename IpPacketRouterBuilder to IpPacketRouter --- gateway/src/node/mod.rs | 8 ++++---- .../ip-packet-router/src/ip_packet_router.rs | 4 ++-- service-providers/ip-packet-router/src/lib.rs | 6 +++--- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index a77a358a8c..fa0f8b35b0 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -347,19 +347,19 @@ impl Gateway { // TODO: well, wire it up internally to gateway traffic, shutdowns, etc. let (on_start_tx, on_start_rx) = oneshot::channel(); - let mut ip_builder = - nym_ip_packet_router::IpPacketRouterBuilder::new(ip_opts.config.clone()) + let mut ip_packet_router = + nym_ip_packet_router::IpPacketRouter::new(ip_opts.config.clone()) .with_shutdown(shutdown) .with_custom_gateway_transceiver(Box::new(transceiver)) .with_wait_for_gateway(true) .with_on_start(on_start_tx); if let Some(custom_mixnet) = &ip_opts.custom_mixnet_path { - ip_builder = ip_builder.with_stored_topology(custom_mixnet)? + ip_packet_router = ip_packet_router.with_stored_topology(custom_mixnet)? } tokio::spawn(async move { - if let Err(err) = ip_builder.run_service_provider().await { + if let Err(err) = ip_packet_router.run_service_provider().await { // no need to panic as we have passed a task client to the ip packet router so // we're most likely already in the process of shutting down error!("ip packet router has failed: {err}") diff --git a/service-providers/ip-packet-router/src/ip_packet_router.rs b/service-providers/ip-packet-router/src/ip_packet_router.rs index 15160ccd58..7cdd586eb0 100644 --- a/service-providers/ip-packet-router/src/ip_packet_router.rs +++ b/service-providers/ip-packet-router/src/ip_packet_router.rs @@ -33,7 +33,7 @@ impl OnStartData { } } -pub struct IpPacketRouterBuilder { +pub struct IpPacketRouter { #[allow(unused)] config: Config, wait_for_gateway: bool, @@ -43,7 +43,7 @@ pub struct IpPacketRouterBuilder { on_start: Option>, } -impl IpPacketRouterBuilder { +impl IpPacketRouter { pub fn new(config: Config) -> Self { Self { config, diff --git a/service-providers/ip-packet-router/src/lib.rs b/service-providers/ip-packet-router/src/lib.rs index dcc00c985b..d3dab9bde1 100644 --- a/service-providers/ip-packet-router/src/lib.rs +++ b/service-providers/ip-packet-router/src/lib.rs @@ -2,14 +2,14 @@ #![cfg_attr(not(target_os = "linux"), allow(unused_imports))] pub use crate::config::Config; -pub use ip_packet_router::{IpPacketRouterBuilder, OnStartData}; +pub use ip_packet_router::{IpPacketRouter, OnStartData}; +pub mod config; mod constants; +pub mod error; mod ip_packet_router; mod mixnet_client; mod mixnet_listener; mod request_filter; mod tun_listener; mod util; -pub mod config; -pub mod error; From 0e26a6efdf239023a1dc08493be21aa5c9fc3a05 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 09:45:44 +0100 Subject: [PATCH 4/9] Register num_hops --- common/ip-packet-requests/src/lib.rs | 10 +++++ .../ip-packet-router/src/mixnet_listener.rs | 38 ++++++++++++++----- .../ip-packet-router/src/tun_listener.rs | 7 ++-- 3 files changed, 42 insertions(+), 13 deletions(-) diff --git a/common/ip-packet-requests/src/lib.rs b/common/ip-packet-requests/src/lib.rs index 290deec697..bbdcd12dc9 100644 --- a/common/ip-packet-requests/src/lib.rs +++ b/common/ip-packet-requests/src/lib.rs @@ -108,16 +108,26 @@ pub enum IpPacketRequestData { pub struct StaticConnectRequest { pub request_id: u64, pub ip: IpAddr, + // The nym-address the response should be sent back to pub reply_to: Recipient, + // The number of mix node hops that responses should take, in addition to the entry and exit + // node. Zero means only client -> entry -> exit -> client. pub reply_to_hops: Option, + // The average delay at each mix node, in milliseconds. Currently this is not supported by the + // ip packet router. pub reply_to_avg_mix_delays: Option, } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct DynamicConnectRequest { pub request_id: u64, + // The nym-address the response should be sent back to pub reply_to: Recipient, + // The number of mix node hops that responses should take, in addition to the entry and exit + // node. Zero means only client -> entry -> exit -> client. pub reply_to_hops: Option, + // The average delay at each mix node, in milliseconds. Currently this is not supported by the + // ip packet router. pub reply_to_avg_mix_delays: Option, } diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index f4e4b63a52..5b230e341e 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -34,6 +34,7 @@ pub(crate) struct MixnetListener { pub(crate) struct ConnectedClient { pub(crate) nym_address: Recipient, + pub(crate) mix_hops: Option, pub(crate) last_activity: std::time::Instant, } @@ -51,7 +52,8 @@ impl MixnetListener { let request_id = connect_request.request_id; let requested_ip = connect_request.ip; let reply_to = connect_request.reply_to; - // TODO: ignoring reply_to_hops and reply_to_avg_mix_delays for now + let reply_to_hops = connect_request.reply_to_hops; + // TODO: ignoring reply_to_avg_mix_delays for now // Check that the IP is available in the set of connected clients let is_ip_taken = self.connected_clients.contains_key(&requested_ip); @@ -81,14 +83,16 @@ impl MixnetListener { requested_ip, ConnectedClient { nym_address: reply_to, + mix_hops: reply_to_hops, last_activity: std::time::Instant::now(), }, ); self.connected_client_tx - .send(ConnectedClientEvent::Connect( - requested_ip, - Box::new(reply_to), - )) + .send(ConnectedClientEvent::Connect(ConnectEvent { + ip: requested_ip, + nym_address: reply_to, + mix_hops: reply_to_hops, + })) .unwrap(); Ok(Some(IpPacketResponse::new_static_connect_success( request_id, reply_to, @@ -124,7 +128,8 @@ impl MixnetListener { let request_id = connect_request.request_id; let reply_to = connect_request.reply_to; - // TODO: ignoring reply_to_hops and reply_to_avg_mix_delays for now + let reply_to_hops = connect_request.reply_to_hops; + // TODO: ignoring reply_to_avg_mix_delays for now // Check if it's the same client connecting again, then we just reuse the same IP // TODO: this is problematic. Until we sign connect requests this means you can spam people @@ -165,11 +170,16 @@ impl MixnetListener { new_ip, ConnectedClient { nym_address: reply_to, + mix_hops: reply_to_hops, last_activity: std::time::Instant::now(), }, ); self.connected_client_tx - .send(ConnectedClientEvent::Connect(new_ip, Box::new(reply_to))) + .send(ConnectedClientEvent::Connect(ConnectEvent { + ip: new_ip, + nym_address: reply_to, + mix_hops: reply_to_hops, + })) .unwrap(); Ok(Some(IpPacketResponse::new_dynamic_connect_success( request_id, reply_to, new_ip, @@ -285,7 +295,7 @@ impl MixnetListener { for ip in inactive_clients { log::info!("Disconnect inactive client: {ip}"); self.connected_clients.remove(&ip); - self.connected_client_tx.send(ConnectedClientEvent::Disconnect(ip)).unwrap(); + self.connected_client_tx.send(ConnectedClientEvent::Disconnect(DisconnectEvent(ip))).unwrap(); } }, msg = self.mixnet_client.next() => { @@ -330,6 +340,14 @@ impl MixnetListener { } pub(crate) enum ConnectedClientEvent { - Disconnect(IpAddr), - Connect(IpAddr, Box), + Disconnect(DisconnectEvent), + Connect(ConnectEvent), +} + +pub(crate) struct DisconnectEvent(pub(crate) IpAddr); + +pub(crate) struct ConnectEvent { + pub(crate) ip: IpAddr, + pub(crate) nym_address: Recipient, + pub(crate) mix_hops: Option, } diff --git a/service-providers/ip-packet-router/src/tun_listener.rs b/service-providers/ip-packet-router/src/tun_listener.rs index 15e09c13b5..28ed7ab087 100644 --- a/service-providers/ip-packet-router/src/tun_listener.rs +++ b/service-providers/ip-packet-router/src/tun_listener.rs @@ -31,14 +31,15 @@ impl TunListener { log::trace!("TunListener: received shutdown"); }, event = self.connected_client_rx.recv() => match event { - Some(mixnet_listener::ConnectedClientEvent::Connect(ip, nym_addr)) => { + Some(mixnet_listener::ConnectedClientEvent::Connect(mixnet_listener::ConnectEvent { ip, nym_address, mix_hops })) => { log::trace!("Connect client: {ip}"); self.connected_clients.insert(ip, mixnet_listener::ConnectedClient { - nym_address: *nym_addr, + nym_address, + mix_hops, last_activity: std::time::Instant::now(), }); }, - Some(mixnet_listener::ConnectedClientEvent::Disconnect(ip)) => { + Some(mixnet_listener::ConnectedClientEvent::Disconnect(mixnet_listener::DisconnectEvent(ip))) => { log::trace!("Disconnect client: {ip}"); self.connected_clients.remove(&ip); }, From 34be9dc60f3c0745f67d1cbaa96c12aca60fdfee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 10:34:17 +0100 Subject: [PATCH 5/9] Handle mixhops in tun_listener --- .../ip-packet-router/src/error.rs | 14 +- .../ip-packet-router/src/tun_listener.rs | 131 +++++++++++++----- 2 files changed, 102 insertions(+), 43 deletions(-) diff --git a/service-providers/ip-packet-router/src/error.rs b/service-providers/ip-packet-router/src/error.rs index 370b6014db..c355cd6314 100644 --- a/service-providers/ip-packet-router/src/error.rs +++ b/service-providers/ip-packet-router/src/error.rs @@ -33,6 +33,9 @@ pub enum IpPacketRouterError { #[error("received packet has an invalid version: {0}")] InvalidPacketVersion(u8), + #[error("failed to serialize response packet: {source}")] + FailedToSerializeResponsePacket { source: Box }, + #[error("failed to deserialize tagged packet: {source}")] FailedToDeserializeTaggedPacket { source: bincode::Error }, @@ -45,14 +48,12 @@ pub enum IpPacketRouterError { #[error("parsed packet is missing transport header")] PacketMissingTransportHeader, - #[error("failed to send packet to tun device: {source}")] - FailedToSendPacketToTun { - source: tokio::sync::mpsc::error::TrySendError<(u64, Vec)>, - }, - #[error("failed to write packet to tun")] FailedToWritePacketToTun, + #[error("failed to send packet to mixnet: {source}")] + FailedToSendPacketToMixnet { source: nym_sdk::Error }, + #[error("the provided socket address, '{addr}' is not covered by the exit policy!")] AddressNotCoveredByExitPolicy { addr: SocketAddr }, @@ -73,4 +74,7 @@ pub enum IpPacketRouterError { #[error("can't setup an exit policy without any upstream urls")] NoUpstreamExitPolicy, + } + +pub type Result = std::result::Result; diff --git a/service-providers/ip-packet-router/src/tun_listener.rs b/service-providers/ip-packet-router/src/tun_listener.rs index 28ed7ab087..a1756c61fd 100644 --- a/service-providers/ip-packet-router/src/tun_listener.rs +++ b/service-providers/ip-packet-router/src/tun_listener.rs @@ -1,13 +1,17 @@ use std::{collections::HashMap, net::IpAddr}; use nym_ip_packet_requests::IpPacketResponse; -use nym_sdk::mixnet::{InputMessage, MixnetMessageSender}; +use nym_sdk::mixnet::{InputMessage, MixnetMessageSender, Recipient}; use nym_task::{connections::TransmissionLane, TaskClient}; #[cfg(target_os = "linux")] use tokio::io::AsyncReadExt; use tokio::sync::mpsc::UnboundedReceiver; -use crate::{error::IpPacketRouterError, mixnet_listener, util::parse_ip::parse_dst_addr}; +use crate::{ + error::{IpPacketRouterError, Result}, + mixnet_listener, + util::parse_ip::parse_dst_addr, +}; // Reads packet from TUN and writes to mixnet client #[cfg(target_os = "linux")] @@ -21,9 +25,89 @@ pub(crate) struct TunListener { pub(crate) connected_client_rx: UnboundedReceiver, } +fn create_input_message( + nym_address: Recipient, + response_packet: Vec, + mix_hops: Option, +) -> InputMessage { + let lane = TransmissionLane::General; + let packet_type = None; + if let Some(mix_hops) = mix_hops { + InputMessage::new_regular_with_custom_hops( + nym_address, + response_packet, + lane, + packet_type, + mix_hops, + ) + } else { + InputMessage::new_regular(nym_address, response_packet, lane, packet_type) + } +} + #[cfg(target_os = "linux")] impl TunListener { - async fn run(mut self) -> Result<(), IpPacketRouterError> { + async fn handle_packet(&mut self, buf: &[u8], len: usize) -> Result<()> { + let Some(dst_addr) = parse_dst_addr(&buf[..len]) else { + log::warn!("Failed to parse packet"); + return Ok(()); + }; + + if let Some(mixnet_listener::ConnectedClient { + nym_address, + mix_hops, + .. + }) = self.connected_clients.get(&dst_addr) + { + let packet = buf[..len].to_vec(); + let response_packet = IpPacketResponse::new_ip_packet(packet.into()) + .to_bytes() + .map_err(|err| IpPacketRouterError::FailedToSerializeResponsePacket { + source: err, + })?; + let input_message = create_input_message(*nym_address, response_packet, *mix_hops); + + self.mixnet_client_sender + .send(input_message) + .await + .map_err(|err| IpPacketRouterError::FailedToSendPacketToMixnet { source: err })?; + } else { + log::info!("No registered nym-address for packet - dropping"); + } + + Ok(()) + } + + async fn handle_connected_client_event( + &mut self, + event: mixnet_listener::ConnectedClientEvent, + ) { + match event { + mixnet_listener::ConnectedClientEvent::Connect(mixnet_listener::ConnectEvent { + ip, + nym_address, + mix_hops, + }) => { + log::trace!("Connect client: {ip}"); + self.connected_clients.insert( + ip, + mixnet_listener::ConnectedClient { + nym_address, + mix_hops, + last_activity: std::time::Instant::now(), + }, + ); + } + mixnet_listener::ConnectedClientEvent::Disconnect( + mixnet_listener::DisconnectEvent(ip), + ) => { + log::trace!("Disconnect client: {ip}"); + self.connected_clients.remove(&ip); + } + } + } + + async fn run(mut self) -> Result<()> { let mut buf = [0u8; 65535]; while !self.task_client.is_shutdown() { tokio::select! { @@ -31,45 +115,16 @@ impl TunListener { log::trace!("TunListener: received shutdown"); }, event = self.connected_client_rx.recv() => match event { - Some(mixnet_listener::ConnectedClientEvent::Connect(mixnet_listener::ConnectEvent { ip, nym_address, mix_hops })) => { - log::trace!("Connect client: {ip}"); - self.connected_clients.insert(ip, mixnet_listener::ConnectedClient { - nym_address, - mix_hops, - last_activity: std::time::Instant::now(), - }); + Some(event) => self.handle_connected_client_event(event).await, + None => { + log::error!("TunListener: connected client channel closed"); + break; }, - Some(mixnet_listener::ConnectedClientEvent::Disconnect(mixnet_listener::DisconnectEvent(ip))) => { - log::trace!("Disconnect client: {ip}"); - self.connected_clients.remove(&ip); - }, - None => {}, }, len = self.tun_reader.read(&mut buf) => match len { Ok(len) => { - let Some(dst_addr) = parse_dst_addr(&buf[..len]) else { - log::warn!("Failed to parse packet"); - continue; - }; - - let recipient = self.connected_clients.get(&dst_addr).map(|c| c.nym_address); - - if let Some(recipient) = recipient { - let lane = TransmissionLane::General; - let packet_type = None; - let packet = buf[..len].to_vec(); - let response_packet = IpPacketResponse::new_ip_packet(packet.into()).to_bytes(); - let Ok(response_packet) = response_packet else { - log::error!("Failed to serialize response packet"); - continue; - }; - let input_message = InputMessage::new_regular(recipient, response_packet, lane, packet_type); - - if let Err(err) = self.mixnet_client_sender.send(input_message).await { - log::error!("TunListener: failed to send packet to mixnet: {err}"); - }; - } else { - log::info!("No registered nym-address for packet - dropping"); + if let Err(err) = self.handle_packet(&buf, len).await { + log::error!("tun: failed to handle packet: {err}"); } }, Err(err) => { From 66e36a7ed5023d40be2026338ce6f2e4023df865 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 11:08:36 +0100 Subject: [PATCH 6/9] Use mix hops for handling responses --- .../ip-packet-router/src/error.rs | 2 + .../ip-packet-router/src/mixnet_listener.rs | 76 ++++++++++++------- .../ip-packet-router/src/tun_listener.rs | 39 +++------- .../src/util/create_message.rs | 22 ++++++ .../ip-packet-router/src/util/mod.rs | 1 + 5 files changed, 82 insertions(+), 58 deletions(-) create mode 100644 service-providers/ip-packet-router/src/util/create_message.rs diff --git a/service-providers/ip-packet-router/src/error.rs b/service-providers/ip-packet-router/src/error.rs index c355cd6314..528c382bba 100644 --- a/service-providers/ip-packet-router/src/error.rs +++ b/service-providers/ip-packet-router/src/error.rs @@ -75,6 +75,8 @@ pub enum IpPacketRouterError { #[error("can't setup an exit policy without any upstream urls")] NoUpstreamExitPolicy, + #[error("no recipient in response packet")] + NoRecipientInResponse, } pub type Result = std::result::Result; diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index 5b230e341e..36b242307d 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -5,18 +5,21 @@ use nym_ip_packet_requests::{ DynamicConnectFailureReason, IpPacketRequest, IpPacketRequestData, IpPacketResponse, StaticConnectFailureReason, }; -use nym_sdk::mixnet::{InputMessage, MixnetMessageSender, Recipient}; +use nym_sdk::mixnet::{MixnetMessageSender, Recipient}; use nym_sphinx::receiver::ReconstructedMessage; -use nym_task::{connections::TransmissionLane, TaskHandle}; +use nym_task::TaskHandle; #[cfg(target_os = "linux")] use tokio::io::AsyncWriteExt; use crate::{ constants::{CLIENT_INACTIVITY_TIMEOUT, DISCONNECT_TIMER_INTERVAL}, - error::IpPacketRouterError, + error::{IpPacketRouterError, Result}, request_filter::{self}, util::generate_new_ip, - util::parse_ip::{parse_packet, ParsedPacket}, + util::{ + create_message::create_input_message, + parse_ip::{parse_packet, ParsedPacket}, + }, Config, }; @@ -43,7 +46,7 @@ impl MixnetListener { async fn on_static_connect_request( &mut self, connect_request: nym_ip_packet_requests::StaticConnectRequest, - ) -> Result, IpPacketRouterError> { + ) -> Result> { log::info!( "Received static connect request from {sender_address}", sender_address = connect_request.reply_to @@ -88,11 +91,11 @@ impl MixnetListener { }, ); self.connected_client_tx - .send(ConnectedClientEvent::Connect(ConnectEvent { + .send(ConnectedClientEvent::Connect(Box::new(ConnectEvent { ip: requested_ip, nym_address: reply_to, mix_hops: reply_to_hops, - })) + }))) .unwrap(); Ok(Some(IpPacketResponse::new_static_connect_success( request_id, reply_to, @@ -120,7 +123,7 @@ impl MixnetListener { async fn on_dynamic_connect_request( &mut self, connect_request: nym_ip_packet_requests::DynamicConnectRequest, - ) -> Result, IpPacketRouterError> { + ) -> Result> { log::info!( "Received dynamic connect request from {sender_address}", sender_address = connect_request.reply_to @@ -175,11 +178,11 @@ impl MixnetListener { }, ); self.connected_client_tx - .send(ConnectedClientEvent::Connect(ConnectEvent { + .send(ConnectedClientEvent::Connect(Box::new(ConnectEvent { ip: new_ip, nym_address: reply_to, mix_hops: reply_to_hops, - })) + }))) .unwrap(); Ok(Some(IpPacketResponse::new_dynamic_connect_success( request_id, reply_to, new_ip, @@ -189,7 +192,7 @@ impl MixnetListener { async fn on_data_request( &mut self, data_request: nym_ip_packet_requests::DataRequest, - ) -> Result, IpPacketRouterError> { + ) -> Result> { log::trace!("Received data request"); // We don't forward packets that we are not able to parse. BUT, there might be a good @@ -242,7 +245,7 @@ impl MixnetListener { async fn on_reconstructed_message( &mut self, reconstructed: ReconstructedMessage, - ) -> Result, IpPacketRouterError> { + ) -> Result> { log::debug!( "Received message with sender_tag: {:?}", reconstructed.sender_tag @@ -272,7 +275,34 @@ impl MixnetListener { } } - pub(crate) async fn run(mut self) -> Result<(), IpPacketRouterError> { + // When an incoming mixnet message triggers a response that we send back, such as during + // connect handshake. + async fn handle_response(&self, response: IpPacketResponse) -> Result<()> { + let Some(recipient) = response.recipient() else { + log::error!("no recipient in response packet, this should NOT happen!"); + return Err(IpPacketRouterError::NoRecipientInResponse); + }; + + let response_packet = response.to_bytes().map_err(|err| { + log::error!("Failed to serialize response packet"); + IpPacketRouterError::FailedToSerializeResponsePacket { source: err } + })?; + + // We could avoid this lookup if we check this when we create the response. + let mix_hops = self + .connected_clients + .values() + .find(|client| client.nym_address == *recipient) + .and_then(|c| c.mix_hops); + + let input_message = create_input_message(*recipient, response_packet, mix_hops); + self.mixnet_client + .send(input_message) + .await + .map_err(|err| IpPacketRouterError::FailedToSendPacketToMixnet { source: err }) + } + + pub(crate) async fn run(mut self) -> Result<()> { let mut task_client = self.task_handle.fork("main_loop"); let mut disconnect_timer = tokio::time::interval(DISCONNECT_TIMER_INTERVAL); @@ -302,21 +332,9 @@ impl MixnetListener { if let Some(msg) = msg { match self.on_reconstructed_message(msg).await { Ok(Some(response)) => { - let Some(recipient) = response.recipient() else { - log::error!("IpPacketRouter [main loop]: failed to get recipient from response"); - continue; - }; - let response_packet = response.to_bytes(); - let Ok(response_packet) = response_packet else { - log::error!("Failed to serialize response packet"); - continue; - }; - let lane = TransmissionLane::General; - let packet_type = None; - let input_message = InputMessage::new_regular(*recipient, response_packet, lane, packet_type); - if let Err(err) = self.mixnet_client.send(input_message).await { - log::error!("IpPacketRouter [main loop]: failed to send packet to mixnet: {err}"); - }; + if let Err(err) = self.handle_response(response).await { + log::error!("Mixnet listener failed to handle response: {err}"); + } }, Ok(None) => { continue; @@ -341,7 +359,7 @@ impl MixnetListener { pub(crate) enum ConnectedClientEvent { Disconnect(DisconnectEvent), - Connect(ConnectEvent), + Connect(Box), } pub(crate) struct DisconnectEvent(pub(crate) IpAddr); diff --git a/service-providers/ip-packet-router/src/tun_listener.rs b/service-providers/ip-packet-router/src/tun_listener.rs index a1756c61fd..3b51c23789 100644 --- a/service-providers/ip-packet-router/src/tun_listener.rs +++ b/service-providers/ip-packet-router/src/tun_listener.rs @@ -1,16 +1,16 @@ use std::{collections::HashMap, net::IpAddr}; use nym_ip_packet_requests::IpPacketResponse; -use nym_sdk::mixnet::{InputMessage, MixnetMessageSender, Recipient}; -use nym_task::{connections::TransmissionLane, TaskClient}; +use nym_sdk::mixnet::MixnetMessageSender; +use nym_task::TaskClient; #[cfg(target_os = "linux")] use tokio::io::AsyncReadExt; use tokio::sync::mpsc::UnboundedReceiver; use crate::{ error::{IpPacketRouterError, Result}, - mixnet_listener, - util::parse_ip::parse_dst_addr, + mixnet_listener::{self, ConnectEvent}, + util::{create_message::create_input_message, parse_ip::parse_dst_addr}, }; // Reads packet from TUN and writes to mixnet client @@ -25,26 +25,6 @@ pub(crate) struct TunListener { pub(crate) connected_client_rx: UnboundedReceiver, } -fn create_input_message( - nym_address: Recipient, - response_packet: Vec, - mix_hops: Option, -) -> InputMessage { - let lane = TransmissionLane::General; - let packet_type = None; - if let Some(mix_hops) = mix_hops { - InputMessage::new_regular_with_custom_hops( - nym_address, - response_packet, - lane, - packet_type, - mix_hops, - ) - } else { - InputMessage::new_regular(nym_address, response_packet, lane, packet_type) - } -} - #[cfg(target_os = "linux")] impl TunListener { async fn handle_packet(&mut self, buf: &[u8], len: usize) -> Result<()> { @@ -83,11 +63,12 @@ impl TunListener { event: mixnet_listener::ConnectedClientEvent, ) { match event { - mixnet_listener::ConnectedClientEvent::Connect(mixnet_listener::ConnectEvent { - ip, - nym_address, - mix_hops, - }) => { + mixnet_listener::ConnectedClientEvent::Connect(connected_event) => { + let ConnectEvent { + ip, + nym_address, + mix_hops, + } = *connected_event; log::trace!("Connect client: {ip}"); self.connected_clients.insert( ip, diff --git a/service-providers/ip-packet-router/src/util/create_message.rs b/service-providers/ip-packet-router/src/util/create_message.rs new file mode 100644 index 0000000000..7ca7934ff5 --- /dev/null +++ b/service-providers/ip-packet-router/src/util/create_message.rs @@ -0,0 +1,22 @@ +use nym_sdk::mixnet::{InputMessage, Recipient}; +use nym_task::connections::TransmissionLane; + +pub(crate) fn create_input_message( + nym_address: Recipient, + response_packet: Vec, + mix_hops: Option, +) -> InputMessage { + let lane = TransmissionLane::General; + let packet_type = None; + if let Some(mix_hops) = mix_hops { + InputMessage::new_regular_with_custom_hops( + nym_address, + response_packet, + lane, + packet_type, + mix_hops, + ) + } else { + InputMessage::new_regular(nym_address, response_packet, lane, packet_type) + } +} diff --git a/service-providers/ip-packet-router/src/util/mod.rs b/service-providers/ip-packet-router/src/util/mod.rs index 62f035c4fb..c096a4a930 100644 --- a/service-providers/ip-packet-router/src/util/mod.rs +++ b/service-providers/ip-packet-router/src/util/mod.rs @@ -1,2 +1,3 @@ +pub(crate) mod create_message; pub(crate) mod generate_new_ip; pub(crate) mod parse_ip; From 49aaf860a8cff5b9f060935fa0ee4663525c2e0a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 11:38:32 +0100 Subject: [PATCH 7/9] Extract out ConnectedClients type --- .../ip-packet-router/src/error.rs | 3 + .../ip-packet-router/src/ip_packet_router.rs | 7 +- .../ip-packet-router/src/mixnet_listener.rs | 204 +++++++++++------- 3 files changed, 134 insertions(+), 80 deletions(-) diff --git a/service-providers/ip-packet-router/src/error.rs b/service-providers/ip-packet-router/src/error.rs index 528c382bba..b71dac5ffc 100644 --- a/service-providers/ip-packet-router/src/error.rs +++ b/service-providers/ip-packet-router/src/error.rs @@ -77,6 +77,9 @@ pub enum IpPacketRouterError { #[error("no recipient in response packet")] NoRecipientInResponse, + + #[error("failed to update client activity")] + FailedToUpdateClientActivity, } pub type Result = std::result::Result; diff --git a/service-providers/ip-packet-router/src/ip_packet_router.rs b/service-providers/ip-packet-router/src/ip_packet_router.rs index 7cdd586eb0..907424fed2 100644 --- a/service-providers/ip-packet-router/src/ip_packet_router.rs +++ b/service-providers/ip-packet-router/src/ip_packet_router.rs @@ -136,7 +136,8 @@ impl IpPacketRouter { // Channel used by the IpPacketRouter to signal connected and disconnected clients to the // TunListener - let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel(); + let (connected_clients, connected_client_rx) = mixnet_listener::ConnectedClients::new(); + // let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel(); let tun_listener = tun_listener::TunListener { tun_reader, @@ -156,8 +157,8 @@ impl IpPacketRouter { tun_writer, mixnet_client, task_handle, - connected_clients: Default::default(), - connected_client_tx, + connected_clients, + // connected_client_tx, }; log::info!("The address of this client is: {self_address}"); diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index 36b242307d..827749f6c0 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -31,8 +31,109 @@ pub(crate) struct MixnetListener { pub(crate) mixnet_client: nym_sdk::mixnet::MixnetClient, pub(crate) task_handle: TaskHandle, - pub(crate) connected_clients: HashMap, - pub(crate) connected_client_tx: tokio::sync::mpsc::UnboundedSender, + // pub(crate) connected_clients: HashMap, + // pub(crate) connected_client_tx: tokio::sync::mpsc::UnboundedSender, + pub(crate) connected_clients: ConnectedClients, +} + +pub(crate) struct ConnectedClients { + clients: HashMap, + connected_client_tx: tokio::sync::mpsc::UnboundedSender, +} + +impl ConnectedClients { + pub(crate) fn new() -> ( + Self, + tokio::sync::mpsc::UnboundedReceiver, + ) { + let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel(); + ( + Self { + clients: Default::default(), + connected_client_tx, + }, + connected_client_rx, + ) + } + + fn is_ip_connected(&self, ip: &IpAddr) -> bool { + self.clients.contains_key(ip) + } + + fn is_nym_address_connected(&self, nym_address: &Recipient) -> bool { + self.clients + .values() + .any(|client| client.nym_address == *nym_address) + } + + fn get_ip(&self, nym_address: &Recipient) -> Option { + self.clients.iter().find_map(|(ip, client)| { + if client.nym_address == *nym_address { + Some(*ip) + } else { + None + } + }) + } + + fn get_client_by_nym_address(&self, nym_address: &Recipient) -> Option<&ConnectedClient> { + self.clients + .values() + .find(|client| client.nym_address == *nym_address) + } + + fn connect(&mut self, ip: IpAddr, nym_address: Recipient, mix_hops: Option) { + self.clients.insert( + ip, + ConnectedClient { + nym_address, + mix_hops, + last_activity: std::time::Instant::now(), + }, + ); + self.connected_client_tx + .send(ConnectedClientEvent::Connect(Box::new(ConnectEvent { + ip, + nym_address, + mix_hops, + }))) + .unwrap(); + } + + fn update_activity(&mut self, ip: &IpAddr) -> Result<()> { + if let Some(client) = self.clients.get_mut(ip) { + client.last_activity = std::time::Instant::now(); + Ok(()) + } else { + Err(IpPacketRouterError::FailedToUpdateClientActivity) + } + } + + fn disconnect_inactive_clients(&mut self) { + let now = std::time::Instant::now(); + let inactive_clients: Vec = self + .clients + .iter() + .filter_map(|(ip, client)| { + if now.duration_since(client.last_activity) > CLIENT_INACTIVITY_TIMEOUT { + Some(*ip) + } else { + None + } + }) + .collect(); + for ip in inactive_clients { + log::info!("Disconnect inactive client: {ip}"); + self.clients.remove(&ip); + self.connected_client_tx + .send(ConnectedClientEvent::Disconnect(DisconnectEvent(ip))) + .unwrap(); + } + } + + fn find_new_ip(&self) -> Option { + generate_new_ip::find_new_ip(&self.clients) + } } pub(crate) struct ConnectedClient { @@ -59,44 +160,29 @@ impl MixnetListener { // TODO: ignoring reply_to_avg_mix_delays for now // Check that the IP is available in the set of connected clients - let is_ip_taken = self.connected_clients.contains_key(&requested_ip); + let is_ip_taken = self.connected_clients.is_ip_connected(&requested_ip); // Check that the nym address isn't already registered - let is_nym_address_taken = self - .connected_clients - .values() - .any(|client| client.nym_address == reply_to); + let is_nym_address_taken = self.connected_clients.is_nym_address_connected(&reply_to); match (is_ip_taken, is_nym_address_taken) { (true, true) => { log::info!("Connecting an already connected client"); - // Update the last activity time for the client - if let Some(client) = self.connected_clients.get_mut(&requested_ip) { - client.last_activity = std::time::Instant::now(); - } else { - log::error!("Failed to update last activity time for client"); - } + if self + .connected_clients + .update_activity(&requested_ip) + .is_err() + { + log::error!("Failed to update activity for client"); + }; Ok(Some(IpPacketResponse::new_static_connect_success( request_id, reply_to, ))) } (false, false) => { log::info!("Connecting a new client"); - self.connected_clients.insert( - requested_ip, - ConnectedClient { - nym_address: reply_to, - mix_hops: reply_to_hops, - last_activity: std::time::Instant::now(), - }, - ); - self.connected_client_tx - .send(ConnectedClientEvent::Connect(Box::new(ConnectEvent { - ip: requested_ip, - nym_address: reply_to, - mix_hops: reply_to_hops, - }))) - .unwrap(); + self.connected_clients + .connect(requested_ip, reply_to, reply_to_hops); Ok(Some(IpPacketResponse::new_static_connect_success( request_id, reply_to, ))) @@ -137,21 +223,15 @@ impl MixnetListener { // Check if it's the same client connecting again, then we just reuse the same IP // TODO: this is problematic. Until we sign connect requests this means you can spam people // with return traffic - let existing_ip = self.connected_clients.iter().find_map(|(ip, client)| { - if client.nym_address == reply_to { - Some(*ip) - } else { - None - } - }); - if let Some(existing_ip) = existing_ip { + if let Some(existing_ip) = self.connected_clients.get_ip(&reply_to) { log::info!("Found existing client for nym address"); - // Update the last activity time for the client - if let Some(client) = self.connected_clients.get_mut(&existing_ip) { - client.last_activity = std::time::Instant::now(); - } else { - log::error!("Failed to update last activity time for client"); + if self + .connected_clients + .update_activity(&existing_ip) + .is_err() + { + log::error!("Failed to update activity for client"); } return Ok(Some(IpPacketResponse::new_dynamic_connect_success( request_id, @@ -160,7 +240,7 @@ impl MixnetListener { ))); } - let Some(new_ip) = generate_new_ip::find_new_ip(&self.connected_clients) else { + let Some(new_ip) = self.connected_clients.find_new_ip() else { log::info!("No available IP address"); return Ok(Some(IpPacketResponse::new_dynamic_connect_failure( request_id, @@ -169,21 +249,8 @@ impl MixnetListener { ))); }; - self.connected_clients.insert( - new_ip, - ConnectedClient { - nym_address: reply_to, - mix_hops: reply_to_hops, - last_activity: std::time::Instant::now(), - }, - ); - self.connected_client_tx - .send(ConnectedClientEvent::Connect(Box::new(ConnectEvent { - ip: new_ip, - nym_address: reply_to, - mix_hops: reply_to_hops, - }))) - .unwrap(); + self.connected_clients + .connect(new_ip, reply_to, reply_to_hops); Ok(Some(IpPacketResponse::new_dynamic_connect_success( request_id, reply_to, new_ip, ))) @@ -214,9 +281,7 @@ impl MixnetListener { // Check if there is a connected client for this src_addr. If there is, update the last activity time // for the client. If there isn't, drop the packet. - if let Some(client) = self.connected_clients.get_mut(&src_addr) { - client.last_activity = std::time::Instant::now(); - } else { + if self.connected_clients.update_activity(&src_addr).is_err() { log::info!("Dropping packet: no connected client for {src_addr}"); return Ok(None); } @@ -291,8 +356,7 @@ impl MixnetListener { // We could avoid this lookup if we check this when we create the response. let mix_hops = self .connected_clients - .values() - .find(|client| client.nym_address == *recipient) + .get_client_by_nym_address(recipient) .and_then(|c| c.mix_hops); let input_message = create_input_message(*recipient, response_packet, mix_hops); @@ -312,21 +376,7 @@ impl MixnetListener { log::debug!("IpPacketRouter [main loop]: received shutdown"); }, _ = disconnect_timer.tick() => { - let now = std::time::Instant::now(); - let inactive_clients: Vec = self.connected_clients.iter() - .filter_map(|(ip, client)| { - if now.duration_since(client.last_activity) > CLIENT_INACTIVITY_TIMEOUT { - Some(*ip) - } else { - None - } - }) - .collect(); - for ip in inactive_clients { - log::info!("Disconnect inactive client: {ip}"); - self.connected_clients.remove(&ip); - self.connected_client_tx.send(ConnectedClientEvent::Disconnect(DisconnectEvent(ip))).unwrap(); - } + self.connected_clients.disconnect_inactive_clients(); }, msg = self.mixnet_client.next() => { if let Some(msg) = msg { From c8017db6c4f20ef5647d3df1ff7fa363822ed72a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 12:11:16 +0100 Subject: [PATCH 8/9] Extract out ConmnectedClientsListener --- .../ip-packet-router/src/ip_packet_router.rs | 5 +- .../ip-packet-router/src/mixnet_listener.rs | 49 ++++++++++++++++--- .../ip-packet-router/src/tun_listener.rs | 45 ++--------------- 3 files changed, 48 insertions(+), 51 deletions(-) diff --git a/service-providers/ip-packet-router/src/ip_packet_router.rs b/service-providers/ip-packet-router/src/ip_packet_router.rs index 907424fed2..bcc0bf0601 100644 --- a/service-providers/ip-packet-router/src/ip_packet_router.rs +++ b/service-providers/ip-packet-router/src/ip_packet_router.rs @@ -136,15 +136,14 @@ impl IpPacketRouter { // Channel used by the IpPacketRouter to signal connected and disconnected clients to the // TunListener - let (connected_clients, connected_client_rx) = mixnet_listener::ConnectedClients::new(); + let (connected_clients, connected_clients_rx) = mixnet_listener::ConnectedClients::new(); // let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel(); let tun_listener = tun_listener::TunListener { tun_reader, mixnet_client_sender: mixnet_client.split_sender(), task_client: task_handle.get_handle(), - connected_clients: Default::default(), - connected_client_rx, + connected_clients: connected_clients_rx, }; tun_listener.start(); diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index 827749f6c0..66911843c0 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -30,9 +30,6 @@ pub(crate) struct MixnetListener { pub(crate) tun_writer: tokio::io::WriteHalf, pub(crate) mixnet_client: nym_sdk::mixnet::MixnetClient, pub(crate) task_handle: TaskHandle, - - // pub(crate) connected_clients: HashMap, - // pub(crate) connected_client_tx: tokio::sync::mpsc::UnboundedSender, pub(crate) connected_clients: ConnectedClients, } @@ -41,18 +38,54 @@ pub(crate) struct ConnectedClients { connected_client_tx: tokio::sync::mpsc::UnboundedSender, } +pub(crate) struct ConnectedClientsListener { + clients: HashMap, + pub(crate) connected_client_rx: tokio::sync::mpsc::UnboundedReceiver, +} + +impl ConnectedClientsListener { + pub(crate) fn get(&self, ip: &IpAddr) -> Option<&ConnectedClient> { + self.clients.get(ip) + } + + pub(crate) fn update(&mut self, event: ConnectedClientEvent) { + match event { + ConnectedClientEvent::Connect(connected_event) => { + let ConnectEvent { + ip, + nym_address, + mix_hops, + } = *connected_event; + log::trace!("Connect client: {ip}"); + self.clients.insert( + ip, + ConnectedClient { + nym_address, + mix_hops, + last_activity: std::time::Instant::now(), + }, + ); + } + ConnectedClientEvent::Disconnect(DisconnectEvent(ip)) => { + log::trace!("Disconnect client: {ip}"); + self.clients.remove(&ip); + } + } + } +} + impl ConnectedClients { - pub(crate) fn new() -> ( - Self, - tokio::sync::mpsc::UnboundedReceiver, - ) { + pub(crate) fn new() -> (Self, ConnectedClientsListener) { let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel(); ( Self { clients: Default::default(), connected_client_tx, }, - connected_client_rx, + ConnectedClientsListener { + clients: Default::default(), + connected_client_rx, + }, ) } diff --git a/service-providers/ip-packet-router/src/tun_listener.rs b/service-providers/ip-packet-router/src/tun_listener.rs index 3b51c23789..5dcc59d205 100644 --- a/service-providers/ip-packet-router/src/tun_listener.rs +++ b/service-providers/ip-packet-router/src/tun_listener.rs @@ -1,15 +1,12 @@ -use std::{collections::HashMap, net::IpAddr}; - use nym_ip_packet_requests::IpPacketResponse; use nym_sdk::mixnet::MixnetMessageSender; use nym_task::TaskClient; #[cfg(target_os = "linux")] use tokio::io::AsyncReadExt; -use tokio::sync::mpsc::UnboundedReceiver; use crate::{ error::{IpPacketRouterError, Result}, - mixnet_listener::{self, ConnectEvent}, + mixnet_listener::{self}, util::{create_message::create_input_message, parse_ip::parse_dst_addr}, }; @@ -19,10 +16,7 @@ pub(crate) struct TunListener { pub(crate) tun_reader: tokio::io::ReadHalf, pub(crate) mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, pub(crate) task_client: TaskClient, - - // A mirror of the one in IpPacketRouter - pub(crate) connected_clients: HashMap, - pub(crate) connected_client_rx: UnboundedReceiver, + pub(crate) connected_clients: mixnet_listener::ConnectedClientsListener, } #[cfg(target_os = "linux")] @@ -58,36 +52,6 @@ impl TunListener { Ok(()) } - async fn handle_connected_client_event( - &mut self, - event: mixnet_listener::ConnectedClientEvent, - ) { - match event { - mixnet_listener::ConnectedClientEvent::Connect(connected_event) => { - let ConnectEvent { - ip, - nym_address, - mix_hops, - } = *connected_event; - log::trace!("Connect client: {ip}"); - self.connected_clients.insert( - ip, - mixnet_listener::ConnectedClient { - nym_address, - mix_hops, - last_activity: std::time::Instant::now(), - }, - ); - } - mixnet_listener::ConnectedClientEvent::Disconnect( - mixnet_listener::DisconnectEvent(ip), - ) => { - log::trace!("Disconnect client: {ip}"); - self.connected_clients.remove(&ip); - } - } - } - async fn run(mut self) -> Result<()> { let mut buf = [0u8; 65535]; while !self.task_client.is_shutdown() { @@ -95,8 +59,9 @@ impl TunListener { _ = self.task_client.recv() => { log::trace!("TunListener: received shutdown"); }, - event = self.connected_client_rx.recv() => match event { - Some(event) => self.handle_connected_client_event(event).await, + // TODO: ConnectedClientsListener::update should poll the channel instead + event = self.connected_clients.connected_client_rx.recv() => match event { + Some(event) => self.connected_clients.update(event), None => { log::error!("TunListener: connected client channel closed"); break; From ac14382a081b77be6e7b1b9680be94f8c92ca11f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 4 Dec 2023 22:20:39 +0100 Subject: [PATCH 9/9] Change to Option in new_regular_with_custom_hops --- .../client-core/src/client/inbound_messages.rs | 8 ++++---- .../src/util/create_message.rs | 18 +++++++----------- 2 files changed, 11 insertions(+), 15 deletions(-) diff --git a/common/client-core/src/client/inbound_messages.rs b/common/client-core/src/client/inbound_messages.rs index f567c84fb7..baf163913f 100644 --- a/common/client-core/src/client/inbound_messages.rs +++ b/common/client-core/src/client/inbound_messages.rs @@ -110,13 +110,13 @@ impl InputMessage { data: Vec, lane: TransmissionLane, packet_type: Option, - mix_hops: u8, + mix_hops: Option, ) -> Self { let message = InputMessage::Regular { recipient, data, lane, - mix_hops: Some(mix_hops), + mix_hops, }; if let Some(packet_type) = packet_type { InputMessage::new_wrapper(message, packet_type) @@ -154,14 +154,14 @@ impl InputMessage { reply_surbs: u32, lane: TransmissionLane, packet_type: Option, - mix_hops: u8, + mix_hops: Option, ) -> Self { let message = InputMessage::Anonymous { recipient, data, reply_surbs, lane, - mix_hops: Some(mix_hops), + mix_hops, }; if let Some(packet_type) = packet_type { InputMessage::new_wrapper(message, packet_type) diff --git a/service-providers/ip-packet-router/src/util/create_message.rs b/service-providers/ip-packet-router/src/util/create_message.rs index 7ca7934ff5..d04d85bd22 100644 --- a/service-providers/ip-packet-router/src/util/create_message.rs +++ b/service-providers/ip-packet-router/src/util/create_message.rs @@ -8,15 +8,11 @@ pub(crate) fn create_input_message( ) -> InputMessage { let lane = TransmissionLane::General; let packet_type = None; - if let Some(mix_hops) = mix_hops { - InputMessage::new_regular_with_custom_hops( - nym_address, - response_packet, - lane, - packet_type, - mix_hops, - ) - } else { - InputMessage::new_regular(nym_address, response_packet, lane, packet_type) - } + InputMessage::new_regular_with_custom_hops( + nym_address, + response_packet, + lane, + packet_type, + mix_hops, + ) }