diff --git a/common/ip-packet-requests/src/request.rs b/common/ip-packet-requests/src/request.rs index 9900ce3784..68a95e0c4b 100644 --- a/common/ip-packet-requests/src/request.rs +++ b/common/ip-packet-requests/src/request.rs @@ -121,6 +121,8 @@ pub enum IpPacketRequestData { Data(DataRequest), } +// A static connect request is when the client provides the internal IP address it will use on the +// ip packet router. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct StaticConnectRequest { pub request_id: u64, @@ -135,6 +137,8 @@ pub struct StaticConnectRequest { pub reply_to_avg_mix_delays: Option, } +// A dynamic connect request is when the client does not provide the internal IP address it will use +// on the ip packet router, and instead requests one to be assigned to it. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct DynamicConnectRequest { pub request_id: u64, @@ -148,6 +152,8 @@ pub struct DynamicConnectRequest { pub reply_to_avg_mix_delays: Option, } +// A disconnect request is when the client wants to disconnect from the ip packet router and free +// up the allocated IP address. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct DisconnectRequest { pub request_id: u64, @@ -155,6 +161,7 @@ pub struct DisconnectRequest { pub reply_to: Recipient, } +// A data request is when the client wants to send an IP packet to a destination. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)] pub struct DataRequest { pub ip_packet: bytes::Bytes, diff --git a/service-providers/ip-packet-router/src/connected_client_handler.rs b/service-providers/ip-packet-router/src/connected_client_handler.rs index 6b955a7b70..ce7009f066 100644 --- a/service-providers/ip-packet-router/src/connected_client_handler.rs +++ b/service-providers/ip-packet-router/src/connected_client_handler.rs @@ -1,15 +1,12 @@ -use std::{collections::HashMap, net::IpAddr}; +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only use nym_ip_packet_requests::response::IpPacketResponse; use nym_sdk::mixnet::{MixnetMessageSender, Recipient}; -use nym_task::TaskClient; -#[cfg(target_os = "linux")] -use tokio::io::AsyncReadExt; use crate::{ error::{IpPacketRouterError, Result}, - mixnet_listener::{self}, - util::{create_message::create_input_message, parse_ip::parse_dst_addr}, + util::create_message::create_input_message, }; // Data flow @@ -21,26 +18,38 @@ use crate::{ pub(crate) struct ConnectedClientHandler { nym_address: Recipient, mix_hops: Option, - tun_rx: tokio::sync::mpsc::UnboundedReceiver>, + forward_from_tun_rx: tokio::sync::mpsc::UnboundedReceiver>, mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, close_rx: tokio::sync::oneshot::Receiver<()>, } impl ConnectedClientHandler { - pub(crate) fn new( - nym_address: Recipient, - mix_hops: Option, - tun_rx: tokio::sync::mpsc::UnboundedReceiver>, + pub(crate) fn launch( + reply_to: Recipient, + reply_to_hops: Option, mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, - close_rx: tokio::sync::oneshot::Receiver<()>, - ) -> Self { - ConnectedClientHandler { - nym_address, - mix_hops, - tun_rx, + ) -> ( + tokio::sync::mpsc::UnboundedSender>, + tokio::sync::oneshot::Sender<()>, + ) { + let (close_tx, close_rx) = tokio::sync::oneshot::channel(); + let (forward_from_tun_tx, forward_from_tun_rx) = tokio::sync::mpsc::unbounded_channel(); + + let connected_client_handler = ConnectedClientHandler { + nym_address: reply_to, + mix_hops: reply_to_hops, + forward_from_tun_rx, mixnet_client_sender, close_rx, - } + }; + + tokio::spawn(async move { + if let Err(err) = connected_client_handler.run().await { + log::error!("connected client handler has failed: {err}") + } + }); + + (forward_from_tun_tx, close_tx) } async fn handle_packet(&mut self, packet: Vec) -> Result<()> { @@ -52,41 +61,33 @@ impl ConnectedClientHandler { self.mixnet_client_sender .send(input_message) .await - .map_err(|err| IpPacketRouterError::FailedToSendPacketToMixnet { source: err })?; - - Ok(()) + .map_err(|err| IpPacketRouterError::FailedToSendPacketToMixnet { source: err }) } async fn run(mut self) -> Result<()> { loop { tokio::select! { _ = &mut self.close_rx => { + // WIP(JON): downgrade to trace once confirmed to work log::warn!("ConnectedClientHandler: received shutdown"); break; }, - packet = self.tun_rx.recv() => match packet { + packet = self.forward_from_tun_rx.recv() => match packet { Some(packet) => { if let Err(err) = self.handle_packet(packet).await { log::error!("connected client handler: failed to handle packet: {err}"); } }, None => { - log::error!("connected client handler: tun channel closed"); + log::debug!("connected client handler: tun channel closed"); break; } } } } + // WIP(JON): downgrade to debug once confirmed to work log::warn!("ConnectedClientHandler: exiting"); Ok(()) } - - pub(crate) fn start(self) { - tokio::spawn(async move { - if let Err(err) = self.run().await { - log::error!("connected client handler 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 844e40c8ca..c6cd08e59d 100644 --- a/service-providers/ip-packet-router/src/ip_packet_router.rs +++ b/service-providers/ip-packet-router/src/ip_packet_router.rs @@ -142,11 +142,9 @@ impl IpPacketRouter { // Channel used by the IpPacketRouter to signal connected and disconnected clients to the // TunListener 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: connected_clients_rx, }; @@ -162,7 +160,6 @@ impl IpPacketRouter { mixnet_client, task_handle, 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 c380b78660..d49ece059a 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -31,16 +31,6 @@ use crate::{ }, }; -#[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: ConnectedClients, -} - pub(crate) struct ConnectedClients { clients: HashMap, connected_client_tx: tokio::sync::mpsc::UnboundedSender, @@ -156,22 +146,19 @@ impl ConnectedClients { } pub(crate) struct ConnectedClient { + // The nym address of the connected client that we are communicating with on the other side of + // the mixnet pub(crate) nym_address: Recipient, - pub(crate) mix_hops: Option, - pub(crate) last_activity: std::time::Instant, - // Send to connected clients listener to stop - // This is inside an Option only because we want to send in Drop - pub(crate) close_tx: Option>, -} -impl Drop for ConnectedClient { - fn drop(&mut self) { - log::info!("Dropping connected client: {}", self.nym_address); - if let Some(close_tx) = self.close_tx.take() { - log::info!("Sending close signal to connected client handler"); - close_tx.send(()).unwrap(); - } - } + // Number of mix node hops that the client has requested to use + pub(crate) mix_hops: Option, + + // Keep track of last activity so we can disconnect inactive clients + pub(crate) last_activity: std::time::Instant, + + // Send to connected clients listener to stop. This is option only because we need to take + // ownership of it when the client is dropped. + pub(crate) close_tx: Option>, } impl ConnectedClient { @@ -180,8 +167,44 @@ impl ConnectedClient { } } +impl Drop for ConnectedClient { + fn drop(&mut self) { + // WIP(JON): downgrade to trace once confirmed to work + log::info!("Dropping connected client: {}", self.nym_address); + if let Some(close_tx) = self.close_tx.take() { + // WIP(JON): downgrade to trace once confirmed to work + log::info!("Sending close signal to connected client handler"); + close_tx.send(()).unwrap(); + } + } +} + +#[cfg(target_os = "linux")] +pub(crate) struct MixnetListener { + // The configuration for the mixnet listener + pub(crate) _config: Config, + + // The request filter that we use to check if a packet should be forwarded + pub(crate) request_filter: request_filter::RequestFilter, + + // The TUN device that we use to send and receive packets from the internet + pub(crate) tun_writer: tokio::io::WriteHalf, + + // The mixnet client that we use to send and receive packets from the mixnet + pub(crate) mixnet_client: nym_sdk::mixnet::MixnetClient, + + // The task handle for the main loop + pub(crate) task_handle: TaskHandle, + + // The map of connected clients that the mixnet listener keeps track of. It monitors + // activity and disconnects clients that have been inactive for too long. + pub(crate) connected_clients: ConnectedClients, +} + #[cfg(target_os = "linux")] impl MixnetListener { + // Receving a static connect request from a client with an IP provided that we assign to them, + // if it's available. If it's not available, we send a failure response. async fn on_static_connect_request( &mut self, connect_request: nym_ip_packet_requests::request::StaticConnectRequest, @@ -220,20 +243,15 @@ impl MixnetListener { (false, false) => { log::info!("Connecting a new client"); - // Start the ConnectedClientHandler for the new client - let (close_tx, close_rx) = tokio::sync::oneshot::channel(); - let (forward_from_tun_tx, forward_from_tun_rx) = - tokio::sync::mpsc::unbounded_channel(); - let connected_client_handler = - connected_client_handler::ConnectedClientHandler::new( + // Spawn the ConnectedClientHandler for the new client + let (forward_from_tun_tx, close_tx) = + connected_client_handler::ConnectedClientHandler::launch( reply_to, reply_to_hops, - forward_from_tun_rx, self.mixnet_client.split_sender(), - close_rx, ); - connected_client_handler.start(); + // Register the new client in the set of connected clients self.connected_clients.connect( requested_ip, reply_to, @@ -307,18 +325,15 @@ impl MixnetListener { ))); }; - // Start the ConnectedClientHandler for the new client - let (close_tx, close_rx) = tokio::sync::oneshot::channel(); - let (forward_from_tun_tx, forward_from_tun_rx) = tokio::sync::mpsc::unbounded_channel(); - let connected_client_handler = connected_client_handler::ConnectedClientHandler::new( - reply_to, - reply_to_hops, - forward_from_tun_rx, - self.mixnet_client.split_sender(), - close_rx, - ); - connected_client_handler.start(); + // Spawn the ConnectedClientHandler for the new client + let (forward_from_tun_tx, close_tx) = + connected_client_handler::ConnectedClientHandler::launch( + reply_to, + reply_to_hops, + self.mixnet_client.split_sender(), + ); + // Register the new client in the set of connected clients self.connected_clients.connect( new_ip, reply_to, diff --git a/service-providers/ip-packet-router/src/tun_listener.rs b/service-providers/ip-packet-router/src/tun_listener.rs index d235b20df4..27d197c777 100644 --- a/service-providers/ip-packet-router/src/tun_listener.rs +++ b/service-providers/ip-packet-router/src/tun_listener.rs @@ -1,15 +1,14 @@ use std::{collections::HashMap, net::IpAddr}; -use nym_ip_packet_requests::response::IpPacketResponse; -use nym_sdk::mixnet::{MixnetMessageSender, Recipient}; +use nym_sdk::mixnet::Recipient; use nym_task::TaskClient; #[cfg(target_os = "linux")] use tokio::io::AsyncReadExt; use crate::{ - error::{IpPacketRouterError, Result}, + error::Result, mixnet_listener::{self}, - util::{create_message::create_input_message, parse_ip::parse_dst_addr}, + util::parse_ip::parse_dst_addr, }; pub(crate) struct ConnectedClientMirror { @@ -76,7 +75,7 @@ impl ConnectedClientsListener { #[cfg(target_os = "linux")] pub(crate) struct TunListener { pub(crate) tun_reader: tokio::io::ReadHalf, - pub(crate) mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, + // pub(crate) mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, pub(crate) task_client: TaskClient, pub(crate) connected_clients: ConnectedClientsListener, } @@ -98,18 +97,6 @@ impl TunListener { { let packet = buf[..len].to_vec(); forward_from_tun_tx.send(packet).unwrap(); - - // 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"); }