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) => {