From b1fa70a232a7c92c3bbe332ffa2572041c0df3b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Thu, 30 Nov 2023 11:40:43 +0100 Subject: [PATCH] Sort out task flow in ip packet router (#4200) * Create separate tun listener task * log tweaks * typo * typo in log * Add sync event * typo in log * Add note * Switch to unbounded channel * tweak log * Handle tun device directly in ip-packet-router * clippy * update comment * Fix mac build * allow unused imports on non-linux * listen for shutdown in tun listener * log tweaks --- Cargo.lock | 5 +- common/tun/Cargo.toml | 2 +- common/tun/src/linux/tun_device.rs | 26 ++-- service-providers/ip-packet-router/Cargo.toml | 3 + .../ip-packet-router/src/error.rs | 3 + service-providers/ip-packet-router/src/lib.rs | 138 ++++++++++++++---- 6 files changed, 136 insertions(+), 41 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9dfe502368..0b8497f9a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6652,6 +6652,7 @@ dependencies = [ "tap", "thiserror", "tokio", + "tokio-tun", "url", ] @@ -10944,9 +10945,9 @@ dependencies = [ [[package]] name = "tokio-tun" -version = "0.9.1" +version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4a67d1405a577ba1f4cd61f46608f1db2cadbb6a9549c3fc2eed7f1195393c9" +checksum = "bf2efaf33e86779a3a68b1f1d6e9e13a346c1c75ee3cab7a4293235c463b2668" dependencies = [ "libc", "nix 0.27.1", diff --git a/common/tun/Cargo.toml b/common/tun/Cargo.toml index ec0cd09774..2cde7db280 100644 --- a/common/tun/Cargo.toml +++ b/common/tun/Cargo.toml @@ -18,4 +18,4 @@ log.workspace = true nym-wireguard-types = { path = "../wireguard-types", optional = true } [target.'cfg(target_os = "linux")'.dependencies] -tokio-tun = "0.9.0" +tokio-tun = "0.11.2" diff --git a/common/tun/src/linux/tun_device.rs b/common/tun/src/linux/tun_device.rs index fdebfbae03..20f514f4a9 100644 --- a/common/tun/src/linux/tun_device.rs +++ b/common/tun/src/linux/tun_device.rs @@ -112,15 +112,7 @@ impl TunDevice { routing_mode: RoutingMode, config: TunDeviceConfig, ) -> (Self, TunTaskTx, TunTaskResponseRx) { - let TunDeviceConfig { - base_name, - ip, - netmask, - } = config; - let name = format!("{base_name}%d"); - - let tun = setup_tokio_tun_device(&name, ip, netmask); - log::info!("Created TUN device: {}", tun.name()); + let tun = Self::new_device_only(config); // Channels to communicate with the other tasks let (tun_task_tx, tun_task_rx) = tun_task_channel(); @@ -136,6 +128,19 @@ impl TunDevice { (tun_device, tun_task_tx, tun_task_response_rx) } + pub fn new_device_only(config: TunDeviceConfig) -> tokio_tun::Tun { + let TunDeviceConfig { + base_name, + ip, + netmask, + } = config; + let name = format!("{base_name}%d"); + + let tun = setup_tokio_tun_device(&name, ip, netmask); + log::info!("Created TUN device: {}", tun.name()); + tun + } + // Send outbound packets out on the wild internet async fn handle_tun_write(&mut self, data: TunTaskPayload) -> Result<(), TunDeviceError> { let (tag, packet) = data; @@ -182,6 +187,7 @@ impl TunDevice { } RoutingMode::Passthrough => { + // TODO: skip the parsing at the top of the function log::debug!("Forward packet without checking anything"); return self .tun_task_response_tx @@ -215,7 +221,7 @@ impl TunDevice { // Writing to the TUN device Some(data) = self.tun_task_rx.recv() => { if let Err(err) = self.handle_tun_write(data).await { - log::error!("ifcae: handle_tun_write failed: {err}"); + log::error!("iface: handle_tun_write failed: {err}"); } } } diff --git a/service-providers/ip-packet-router/Cargo.toml b/service-providers/ip-packet-router/Cargo.toml index 44644623ba..01ae4fb897 100644 --- a/service-providers/ip-packet-router/Cargo.toml +++ b/service-providers/ip-packet-router/Cargo.toml @@ -35,3 +35,6 @@ tap.workspace = true thiserror = { workspace = true } tokio = { workspace = true, features = ["rt-multi-thread", "net", "io-util"] } url.workspace = true + +[target.'cfg(target_os = "linux")'.dependencies] +tokio-tun = "0.11.2" diff --git a/service-providers/ip-packet-router/src/error.rs b/service-providers/ip-packet-router/src/error.rs index 797dc78b81..370b6014db 100644 --- a/service-providers/ip-packet-router/src/error.rs +++ b/service-providers/ip-packet-router/src/error.rs @@ -50,6 +50,9 @@ pub enum IpPacketRouterError { source: tokio::sync::mpsc::error::TrySendError<(u64, Vec)>, }, + #[error("failed to write packet to tun")] + FailedToWritePacketToTun, + #[error("the provided socket address, '{addr}' is not covered by the exit policy!")] AddressNotCoveredByExitPolicy { addr: SocketAddr }, diff --git a/service-providers/ip-packet-router/src/lib.rs b/service-providers/ip-packet-router/src/lib.rs index 0a993528c6..619c037b93 100644 --- a/service-providers/ip-packet-router/src/lib.rs +++ b/service-providers/ip-packet-router/src/lib.rs @@ -1,4 +1,5 @@ #![cfg_attr(not(target_os = "linux"), allow(dead_code))] +#![cfg_attr(not(target_os = "linux"), allow(unused_imports))] use std::{ collections::HashMap, @@ -24,6 +25,8 @@ use nym_sdk::{ use nym_sphinx::receiver::ReconstructedMessage; use nym_task::{connections::TransmissionLane, TaskClient, TaskHandle}; use request_filter::RequestFilter; +#[cfg(target_os = "linux")] +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use crate::config::BaseClientConfig; @@ -154,11 +157,21 @@ impl IpPacketRouterBuilder { ip: TUN_DEVICE_ADDRESS.parse().unwrap(), netmask: TUN_DEVICE_NETMASK.parse().unwrap(), }; - let (tun, tun_task_tx, tun_task_response_rx) = nym_tun::tun_device::TunDevice::new( - nym_tun::tun_device::RoutingMode::new_passthrough(), - config, - ); - tun.start(); + let (tun_reader, tun_writer) = + tokio::io::split(nym_tun::tun_device::TunDevice::new_device_only(config)); + + // 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 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, + }; + tun_listener.start(); let request_filter = request_filter::RequestFilter::new(&self.config).await?; request_filter.start_update_tasks().await; @@ -166,11 +179,11 @@ impl IpPacketRouterBuilder { let ip_packet_router_service = IpPacketRouter { _config: self.config, request_filter: request_filter.clone(), - tun_task_tx, - tun_task_response_rx, + tun_writer, mixnet_client, task_handle, connected_clients: Default::default(), + connected_client_tx, }; log::info!("The address of this client is: {self_address}"); @@ -190,16 +203,16 @@ impl IpPacketRouterBuilder { } } -#[cfg_attr(not(target_os = "linux"), allow(unused))] +#[cfg(target_os = "linux")] struct IpPacketRouter { _config: Config, request_filter: request_filter::RequestFilter, - tun_task_tx: nym_tun::tun_task_channel::TunTaskTx, - tun_task_response_rx: nym_tun::tun_task_channel::TunTaskResponseRx, + tun_writer: tokio::io::WriteHalf, mixnet_client: nym_sdk::mixnet::MixnetClient, task_handle: TaskHandle, connected_clients: HashMap, + connected_client_tx: tokio::sync::mpsc::UnboundedSender, } struct ConnectedClient { @@ -207,7 +220,7 @@ struct ConnectedClient { last_activity: std::time::Instant, } -#[cfg_attr(not(target_os = "linux"), allow(unused))] +#[cfg(target_os = "linux")] impl IpPacketRouter { async fn on_static_connect_request( &mut self, @@ -254,6 +267,12 @@ impl IpPacketRouter { 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, ))) @@ -332,6 +351,9 @@ impl IpPacketRouter { 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, ))) @@ -341,7 +363,7 @@ impl IpPacketRouter { &mut self, data_request: nym_ip_packet_requests::DataRequest, ) -> Result, IpPacketRouterError> { - log::info!("Received data request"); + 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. @@ -382,9 +404,11 @@ impl IpPacketRouter { } // TODO: consider changing from Vec to bytes::Bytes? - self.tun_task_tx - .try_send((0, data_request.ip_packet.into())) - .map_err(|err| IpPacketRouterError::FailedToSendPacketToTun { source: err })?; + let packet = data_request.ip_packet; + self.tun_writer + .write_all(&packet) + .await + .map_err(|_| IpPacketRouterError::FailedToWritePacketToTun)?; Ok(None) } @@ -444,6 +468,7 @@ impl IpPacketRouter { 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() => { @@ -479,9 +504,52 @@ impl IpPacketRouter { break; }; }, - packet = self.tun_task_response_rx.recv() => { - if let Some((_tag, packet)) = packet { - let Some(dst_addr) = parse_dst_addr(&packet) else { + + } + } + log::debug!("IpPacketRouter: stopping"); + Ok(()) + } +} + +// Reads packet from TUN and writes to mixnet client +#[cfg(target_os = "linux")] +struct TunListener { + tun_reader: tokio::io::ReadHalf, + mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, + task_client: TaskClient, + + // A mirror of the one in IpPacketRouter + connected_clients: HashMap, + connected_client_rx: tokio::sync::mpsc::UnboundedReceiver, +} + +#[cfg(target_os = "linux")] +impl TunListener { + async fn run(mut self) -> Result<(), IpPacketRouterError> { + 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(ConnectedClientEvent::Connect(ip, nym_addr)) => { + log::trace!("Connect client: {ip}"); + self.connected_clients.insert(ip, ConnectedClient { + nym_address: *nym_addr, + last_activity: std::time::Instant::now(), + }); + }, + Some(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; }; @@ -491,6 +559,7 @@ impl IpPacketRouter { 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"); @@ -498,23 +567,36 @@ impl IpPacketRouter { }; 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.mixnet_client_sender.send(input_message).await { + log::error!("TunListener: failed to send packet to mixnet: {err}"); }; } else { - log::error!("IpPacketRouter [main loop]: no nym-address recipient for packet"); + log::info!("No registered nym-address for packet - dropping"); } - } else { - log::trace!("IpPacketRouter [main loop]: stopping since channel closed"); - break; + }, + Err(err) => { + log::warn!("iface: read error: {err}"); + // break; } } - } } - log::info!("IpPacketRouter: stopping"); + log::debug!("TunListener: stopping"); Ok(()) } + + fn start(self) { + tokio::spawn(async move { + if let Err(err) = self.run().await { + log::error!("tun listener router has failed: {err}") + } + }); + } +} + +enum ConnectedClientEvent { + Disconnect(IpAddr), + Connect(IpAddr, Box), } struct ParsedPacket<'a> { @@ -531,8 +613,8 @@ fn parse_packet(packet: &[u8]) -> Result { })?; let (packet_type, dst_port) = match headers.transport { - Some(etherparse::TransportSlice::Udp(header)) => ("ipv4", Some(header.destination_port())), - Some(etherparse::TransportSlice::Tcp(header)) => ("ipv6", Some(header.destination_port())), + Some(etherparse::TransportSlice::Udp(header)) => ("udp", Some(header.destination_port())), + Some(etherparse::TransportSlice::Tcp(header)) => ("tcp", Some(header.destination_port())), Some(etherparse::TransportSlice::Icmpv4(_)) => ("icmpv4", None), Some(etherparse::TransportSlice::Icmpv6(_)) => ("icmpv6", None), Some(etherparse::TransportSlice::Unknown(_)) => ("unknown", None),