Handle mixhops in tun_listener

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<bincode::ErrorKind> },
+
     #[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<u8>)>,
-    },
-
     #[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<T> = std::result::Result<T, IpPacketRouterError>;

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<mixnet_listener::ConnectedClientEvent>,
 }
 
+fn create_input_message(
+    nym_address: Recipient,
+    response_packet: Vec<u8>,
+    mix_hops: Option<u8>,
+) -> 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) => {