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/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/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/error.rs b/service-providers/ip-packet-router/src/error.rs index 370b6014db..b71dac5ffc 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,12 @@ pub enum IpPacketRouterError { #[error("can't setup an exit policy without any upstream urls")] NoUpstreamExitPolicy, + + #[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 c514be4f1b..bcc0bf0601 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, }; @@ -43,7 +33,7 @@ impl OnStartData { } } -pub struct IpPacketRouterBuilder { +pub struct IpPacketRouter { #[allow(unused)] config: Config, wait_for_gateway: bool, @@ -53,7 +43,7 @@ pub struct IpPacketRouterBuilder { on_start: Option>, } -impl IpPacketRouterBuilder { +impl IpPacketRouter { pub fn new(config: Config) -> Self { Self { 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 @@ -144,28 +136,28 @@ impl IpPacketRouterBuilder { // 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_clients_rx) = mixnet_listener::ConnectedClients::new(); + // 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(), - connected_clients: Default::default(), - connected_client_rx, + connected_clients: connected_clients_rx, }; tun_listener.start(); let request_filter = request_filter::RequestFilter::new(&self.config).await?; request_filter.start_update_tasks().await; - let ip_packet_router_service = IpPacketRouter { + let mixnet_listener = mixnet_listener::MixnetListener { _config: self.config, request_filter: request_filter.clone(), 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}"); @@ -181,320 +173,6 @@ impl IpPacketRouterBuilder { } } - ip_packet_router_service.run().await + mixnet_listener.run().await } } - -#[cfg(target_os = "linux")] -struct IpPacketRouter { - _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 IpPacketRouter { - 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..d3dab9bde1 100644 --- a/service-providers/ip-packet-router/src/lib.rs +++ b/service-providers/ip-packet-router/src/lib.rs @@ -2,13 +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; 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..66911843c0 --- /dev/null +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -0,0 +1,454 @@ +use std::{collections::HashMap, net::IpAddr}; + +use futures::StreamExt; +use nym_ip_packet_requests::{ + DynamicConnectFailureReason, IpPacketRequest, IpPacketRequestData, IpPacketResponse, + StaticConnectFailureReason, +}; +use nym_sdk::mixnet::{MixnetMessageSender, Recipient}; +use nym_sphinx::receiver::ReconstructedMessage; +use nym_task::TaskHandle; +#[cfg(target_os = "linux")] +use tokio::io::AsyncWriteExt; + +use crate::{ + constants::{CLIENT_INACTIVITY_TIMEOUT, DISCONNECT_TIMER_INTERVAL}, + error::{IpPacketRouterError, Result}, + request_filter::{self}, + util::generate_new_ip, + util::{ + create_message::create_input_message, + 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: ConnectedClients, +} + +pub(crate) struct ConnectedClients { + clients: HashMap, + 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, ConnectedClientsListener) { + let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel(); + ( + Self { + clients: Default::default(), + connected_client_tx, + }, + ConnectedClientsListener { + clients: Default::default(), + 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 { + pub(crate) nym_address: Recipient, + pub(crate) mix_hops: Option, + 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> { + 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; + 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.is_ip_connected(&requested_ip); + + // Check that the nym address isn't already registered + 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"); + 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 + .connect(requested_ip, reply_to, reply_to_hops); + 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> { + 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; + 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 + // with return traffic + + if let Some(existing_ip) = self.connected_clients.get_ip(&reply_to) { + log::info!("Found existing client for nym address"); + 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, + reply_to, + existing_ip, + ))); + } + + 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, + reply_to, + DynamicConnectFailureReason::NoAvailableIp, + ))); + }; + + self.connected_clients + .connect(new_ip, reply_to, reply_to_hops); + 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> { + 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 self.connected_clients.update_activity(&src_addr).is_err() { + 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> { + 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, + } + } + + // 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 + .get_client_by_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); + + while !task_client.is_shutdown() { + tokio::select! { + _ = task_client.recv() => { + log::debug!("IpPacketRouter [main loop]: received shutdown"); + }, + _ = disconnect_timer.tick() => { + self.connected_clients.disconnect_inactive_clients(); + }, + msg = self.mixnet_client.next() => { + if let Some(msg) = msg { + match self.on_reconstructed_message(msg).await { + Ok(Some(response)) => { + if let Err(err) = self.handle_response(response).await { + log::error!("Mixnet listener failed to handle response: {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(DisconnectEvent), + Connect(Box), +} + +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 93a16d316d..5dcc59d205 100644 --- a/service-providers/ip-packet-router/src/tun_listener.rs +++ b/service-providers/ip-packet-router/src/tun_listener.rs @@ -1,13 +1,14 @@ -use std::{collections::HashMap, net::IpAddr}; - use nym_ip_packet_requests::IpPacketResponse; -use nym_sdk::mixnet::{InputMessage, MixnetMessageSender}; -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, ip_packet_router, util::parse_ip::parse_dst_addr}; +use crate::{ + error::{IpPacketRouterError, Result}, + mixnet_listener::{self}, + util::{create_message::create_input_message, parse_ip::parse_dst_addr}, +}; // Reads packet from TUN and writes to mixnet client #[cfg(target_os = "linux")] @@ -15,60 +16,61 @@ 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")] 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 run(mut self) -> Result<()> { let mut buf = [0u8; 65535]; while !self.task_client.is_shutdown() { tokio::select! { _ = self.task_client.recv() => { log::trace!("TunListener: received shutdown"); }, - event = self.connected_client_rx.recv() => match event { - Some(ip_packet_router::ConnectedClientEvent::Connect(ip, nym_addr)) => { - log::trace!("Connect client: {ip}"); - self.connected_clients.insert(ip, ip_packet_router::ConnectedClient { - nym_address: *nym_addr, - last_activity: std::time::Instant::now(), - }); + // 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; }, - Some(ip_packet_router::ConnectedClientEvent::Disconnect(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) => { 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..d04d85bd22 --- /dev/null +++ b/service-providers/ip-packet-router/src/util/create_message.rs @@ -0,0 +1,18 @@ +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; + InputMessage::new_regular_with_custom_hops( + nym_address, + response_packet, + lane, + packet_type, + mix_hops, + ) +} 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 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;