From aebd386382e0bfdc66680ae1d6c895ffed68429a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bogdan-=C8=98tefan=20Neac=C5=9Fu?= Date: Fri, 1 Mar 2024 19:28:21 +0200 Subject: [PATCH] Add IPv6 support to IPPR (#4431) --- common/ip-packet-requests/src/lib.rs | 13 +- common/ip-packet-requests/src/request.rs | 27 +- common/ip-packet-requests/src/response.rs | 10 +- common/tun/src/linux/tun_device.rs | 52 ++-- .../ip-packet-router/src/constants.rs | 8 +- .../ip-packet-router/src/error.rs | 4 + .../ip-packet-router/src/ip_packet_router.rs | 8 +- .../ip-packet-router/src/mixnet_listener.rs | 230 +++++++++++------- .../ip-packet-router/src/tun_listener.rs | 42 +++- .../src/util/generate_new_ip.rs | 53 ++-- 10 files changed, 293 insertions(+), 154 deletions(-) diff --git a/common/ip-packet-requests/src/lib.rs b/common/ip-packet-requests/src/lib.rs index fc97d13042..f61e3f266f 100644 --- a/common/ip-packet-requests/src/lib.rs +++ b/common/ip-packet-requests/src/lib.rs @@ -1,8 +1,19 @@ +use serde::{Deserialize, Serialize}; +use std::net::{Ipv4Addr, Ipv6Addr}; + pub mod codec; pub mod request; pub mod response; -pub const CURRENT_VERSION: u8 = 3; +// version 3: initial version +// version 4: IPv6 support +pub const CURRENT_VERSION: u8 = 4; + +#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct IPPair { + pub ipv4: Ipv4Addr, + pub ipv6: Ipv6Addr, +} fn make_bincode_serializer() -> impl bincode::Options { use bincode::Options; diff --git a/common/ip-packet-requests/src/request.rs b/common/ip-packet-requests/src/request.rs index 6f4b17ffea..011ff5bfe6 100644 --- a/common/ip-packet-requests/src/request.rs +++ b/common/ip-packet-requests/src/request.rs @@ -1,9 +1,15 @@ -use std::net::IpAddr; +use std::fmt::{Display, Formatter}; use nym_sphinx::addressing::clients::Recipient; use serde::{Deserialize, Serialize}; -use crate::{make_bincode_serializer, CURRENT_VERSION}; +use crate::{make_bincode_serializer, IPPair, CURRENT_VERSION}; + +impl Display for IPPair { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + writeln!(f, "IPv4: {}, IPV6: {}", self.ipv4, self.ipv6) + } +} fn generate_random() -> u64 { use rand::RngCore; @@ -19,7 +25,7 @@ pub struct IpPacketRequest { impl IpPacketRequest { pub fn new_static_connect_request( - ip: IpAddr, + ips: IPPair, reply_to: Recipient, reply_to_hops: Option, reply_to_avg_mix_delays: Option, @@ -31,7 +37,7 @@ impl IpPacketRequest { version: CURRENT_VERSION, data: IpPacketRequestData::StaticConnect(StaticConnectRequest { request_id, - ip, + ips, reply_to, reply_to_hops, reply_to_avg_mix_delays, @@ -137,7 +143,7 @@ pub enum IpPacketRequestData { pub struct StaticConnectRequest { pub request_id: u64, - pub ip: IpAddr, + pub ips: IPPair, // The nym-address the response should be sent back to pub reply_to: Recipient, @@ -210,6 +216,8 @@ pub struct HealthRequest { #[cfg(test)] mod tests { use super::*; + use std::net::{Ipv4Addr, Ipv6Addr}; + use std::str::FromStr; #[test] fn check_size_of_request() { @@ -218,15 +226,18 @@ mod tests { data: IpPacketRequestData::StaticConnect( StaticConnectRequest { request_id: 123, - ip: IpAddr::from([10, 0, 0, 1]), + ips: IPPair { + ipv4: Ipv4Addr::from_str("10.0.0.1").unwrap(), + ipv6: Ipv6Addr::from_str("2001:db8:a160::1").unwrap(), + }, reply_to: Recipient::try_from_base58_string("D1rrpsysCGCYXy9saP8y3kmNpGtJZUXN9SvFoUcqAsM9.9Ssso1ea5NfkbMASdiseDSjTN1fSWda5SgEVjdSN4CvV@GJqd3ZxpXWSNxTfx7B1pPtswpetH4LnJdFeLeuY5KUuN").unwrap(), reply_to_hops: None, reply_to_avg_mix_delays: None, buffer_timeout: None, }, - ) + ), }; - assert_eq!(connect.to_bytes().unwrap().len(), 108); + assert_eq!(connect.to_bytes().unwrap().len(), 123); } #[test] diff --git a/common/ip-packet-requests/src/response.rs b/common/ip-packet-requests/src/response.rs index 29553fa3e9..91972a35a3 100644 --- a/common/ip-packet-requests/src/response.rs +++ b/common/ip-packet-requests/src/response.rs @@ -1,9 +1,7 @@ -use std::net::IpAddr; - use nym_sphinx::addressing::clients::Recipient; use serde::{Deserialize, Serialize}; -use crate::{make_bincode_serializer, CURRENT_VERSION}; +use crate::{make_bincode_serializer, IPPair, CURRENT_VERSION}; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct IpPacketResponse { @@ -38,13 +36,13 @@ impl IpPacketResponse { } } - pub fn new_dynamic_connect_success(request_id: u64, reply_to: Recipient, ip: IpAddr) -> Self { + pub fn new_dynamic_connect_success(request_id: u64, reply_to: Recipient, ips: IPPair) -> Self { Self { version: CURRENT_VERSION, data: IpPacketResponseData::DynamicConnect(DynamicConnectResponse { request_id, reply_to, - reply: DynamicConnectResponseReply::Success(DynamicConnectSuccess { ip }), + reply: DynamicConnectResponseReply::Success(DynamicConnectSuccess { ips }), }), } } @@ -263,7 +261,7 @@ impl DynamicConnectResponseReply { #[derive(Clone, Debug, Serialize, Deserialize)] pub struct DynamicConnectSuccess { - pub ip: IpAddr, + pub ips: IPPair, } #[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)] diff --git a/common/tun/src/linux/tun_device.rs b/common/tun/src/linux/tun_device.rs index 20f514f4a9..94aea718de 100644 --- a/common/tun/src/linux/tun_device.rs +++ b/common/tun/src/linux/tun_device.rs @@ -1,3 +1,4 @@ +use std::net::Ipv6Addr; use std::{ collections::HashMap, net::{IpAddr, Ipv4Addr}, @@ -19,6 +20,12 @@ const TUN_WRITE_TIMEOUT_MS: u64 = 1000; #[derive(thiserror::Error, Debug)] pub enum TunDeviceError { + #[error("{0}")] + IO(#[from] std::io::Error), + + #[error("{0}")] + TokioTun(#[from] tokio_tun::Error), + #[error("timeout writing to tun device, dropping packet")] TunWriteTimeout, @@ -44,14 +51,18 @@ pub enum TunDeviceError { FailedToLockPeer, } -fn setup_tokio_tun_device(name: &str, address: Ipv4Addr, netmask: Ipv4Addr) -> tokio_tun::Tun { +fn setup_tokio_tun_device( + name: &str, + address: Ipv4Addr, + netmask: Ipv4Addr, +) -> Result { log::info!("Creating TUN device with: address={address}, netmask={netmask}"); // Read MTU size from env variable NYM_MTU_SIZE, else default to 1420. let mtu = std::env::var("NYM_MTU_SIZE") .map(|mtu| mtu.parse().expect("NYM_MTU_SIZE must be a valid integer")) .unwrap_or(1420); log::info!("Using MTU size: {mtu}"); - tokio_tun::Tun::builder() + Ok(tokio_tun::Tun::builder() .name(name) .tap(false) .packet_info(false) @@ -59,8 +70,7 @@ fn setup_tokio_tun_device(name: &str, address: Ipv4Addr, netmask: Ipv4Addr) -> t .up() .address(address) .netmask(netmask) - .try_build() - .expect("Failed to setup tun device, do you have permission?") + .try_build()?) } pub struct TunDevice { @@ -103,16 +113,18 @@ pub struct NatInner { pub struct TunDeviceConfig { pub base_name: String, - pub ip: Ipv4Addr, - pub netmask: Ipv4Addr, + pub ipv4: Ipv4Addr, + pub netmaskv4: Ipv4Addr, + pub ipv6: Ipv6Addr, + pub netmaskv6: String, } impl TunDevice { pub fn new( routing_mode: RoutingMode, config: TunDeviceConfig, - ) -> (Self, TunTaskTx, TunTaskResponseRx) { - let tun = Self::new_device_only(config); + ) -> Result<(Self, TunTaskTx, TunTaskResponseRx), TunDeviceError> { + let tun = Self::new_device_only(config)?; // Channels to communicate with the other tasks let (tun_task_tx, tun_task_rx) = tun_task_channel(); @@ -125,20 +137,32 @@ impl TunDevice { routing_mode, }; - (tun_device, tun_task_tx, tun_task_response_rx) + Ok((tun_device, tun_task_tx, tun_task_response_rx)) } - pub fn new_device_only(config: TunDeviceConfig) -> tokio_tun::Tun { + pub fn new_device_only(config: TunDeviceConfig) -> Result { let TunDeviceConfig { base_name, - ip, - netmask, + ipv4, + netmaskv4, + ipv6, + netmaskv6, } = config; let name = format!("{base_name}%d"); - let tun = setup_tokio_tun_device(&name, ip, netmask); + let tun = setup_tokio_tun_device(&name, ipv4, netmaskv4)?; log::info!("Created TUN device: {}", tun.name()); - tun + std::process::Command::new("ip") + .args([ + "-6", + "addr", + "add", + &format!("{}/{}", ipv6, netmaskv6), + "dev", + &tun.name(), + ]) + .output()?; + Ok(tun) } // Send outbound packets out on the wild internet diff --git a/service-providers/ip-packet-router/src/constants.rs b/service-providers/ip-packet-router/src/constants.rs index 04e3bd1a6b..eb7874f902 100644 --- a/service-providers/ip-packet-router/src/constants.rs +++ b/service-providers/ip-packet-router/src/constants.rs @@ -1,9 +1,13 @@ +use std::net::{Ipv4Addr, Ipv6Addr}; use std::time::Duration; // The interface used to route traffic pub const TUN_BASE_NAME: &str = "nymtun"; -pub const TUN_DEVICE_ADDRESS: &str = "10.0.0.1"; -pub const TUN_DEVICE_NETMASK: &str = "255.255.255.0"; +pub const TUN_DEVICE_ADDRESS_V4: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1); +pub const TUN_DEVICE_NETMASK_V4: Ipv4Addr = Ipv4Addr::new(255, 255, 255, 0); +pub const TUN_DEVICE_ADDRESS_V6: Ipv6Addr = Ipv6Addr::new(0x2001, 0xdb8, 0xa160, 0, 0, 0, 0, 0x1); // 2001:db8:a160::1 + +pub const TUN_DEVICE_NETMASK_V6: &str = "120"; // We routinely check if any clients needs to be disconnected at this interval pub(crate) const DISCONNECT_TIMER_INTERVAL: Duration = Duration::from_secs(10); diff --git a/service-providers/ip-packet-router/src/error.rs b/service-providers/ip-packet-router/src/error.rs index d4e0940b48..4732177191 100644 --- a/service-providers/ip-packet-router/src/error.rs +++ b/service-providers/ip-packet-router/src/error.rs @@ -11,6 +11,10 @@ pub enum IpPacketRouterError { #[error("client-core error: {0}")] ClientCoreError(#[from] ClientCoreError), + #[cfg(target_os = "linux")] + #[error("tun device error: {0}")] + TunDeviceError(#[from] nym_tun::tun_device::TunDeviceError), + #[error("failed to load configuration file: {0}")] FailedToLoadConfig(String), 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 c6cd08e59d..85f2ce2f3f 100644 --- a/service-providers/ip-packet-router/src/ip_packet_router.rs +++ b/service-providers/ip-packet-router/src/ip_packet_router.rs @@ -133,11 +133,13 @@ impl IpPacketRouter { // Create the TUN device that we interact with the rest of the world with let config = nym_tun::tun_device::TunDeviceConfig { base_name: crate::constants::TUN_BASE_NAME.to_string(), - ip: crate::constants::TUN_DEVICE_ADDRESS.parse().unwrap(), - netmask: crate::constants::TUN_DEVICE_NETMASK.parse().unwrap(), + ipv4: crate::constants::TUN_DEVICE_ADDRESS_V4, + netmaskv4: crate::constants::TUN_DEVICE_NETMASK_V4, + ipv6: crate::constants::TUN_DEVICE_ADDRESS_V6, + netmaskv6: crate::constants::TUN_DEVICE_NETMASK_V6.to_string(), }; let (tun_reader, tun_writer) = - tokio::io::split(nym_tun::tun_device::TunDevice::new_device_only(config)); + 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 diff --git a/service-providers/ip-packet-router/src/mixnet_listener.rs b/service-providers/ip-packet-router/src/mixnet_listener.rs index 1f14c18218..813c4b2e52 100644 --- a/service-providers/ip-packet-router/src/mixnet_listener.rs +++ b/service-providers/ip-packet-router/src/mixnet_listener.rs @@ -1,7 +1,6 @@ -use std::{ - collections::HashMap, - net::{IpAddr, SocketAddr}, -}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use std::sync::Arc; +use std::{collections::HashMap, net::SocketAddr}; use bytes::{Bytes, BytesMut}; use futures::StreamExt; @@ -12,6 +11,7 @@ use nym_ip_packet_requests::{ DynamicConnectFailureReason, ErrorResponseReply, IpPacketResponse, StaticConnectFailureReason, }, + IPPair, }; use nym_sdk::mixnet::{MixnetMessageSender, Recipient}; use nym_sphinx::receiver::ReconstructedMessage; @@ -19,6 +19,7 @@ use nym_task::TaskHandle; use tap::TapFallible; #[cfg(target_os = "linux")] use tokio::io::AsyncWriteExt; +use tokio::sync::RwLock; use tokio_util::codec::Decoder; use crate::{ @@ -37,7 +38,8 @@ use crate::{ pub(crate) struct ConnectedClients { // The set of connected clients - clients: HashMap, + clients_ipv4_mapping: HashMap, + clients_ipv6_mapping: HashMap, // Notify the tun listener when a new client connects or disconnects tun_listener_connected_client_tx: tokio::sync::mpsc::UnboundedSender, @@ -48,46 +50,62 @@ impl ConnectedClients { let (connected_client_tx, connected_client_rx) = tokio::sync::mpsc::unbounded_channel(); ( Self { - clients: Default::default(), + clients_ipv4_mapping: Default::default(), + clients_ipv6_mapping: Default::default(), tun_listener_connected_client_tx: connected_client_tx, }, tun_listener::ConnectedClientsListener::new(connected_client_rx), ) } - fn is_ip_connected(&self, ip: &IpAddr) -> bool { - self.clients.contains_key(ip) + fn is_ip_connected(&self, ips: &IPPair) -> bool { + self.clients_ipv4_mapping.contains_key(&ips.ipv4) + || self.clients_ipv6_mapping.contains_key(&ips.ipv6) } fn get_client_from_ip_mut(&mut self, ip: &IpAddr) -> Option<&mut ConnectedClient> { - self.clients.get_mut(ip) + match ip { + IpAddr::V4(ip) => self.clients_ipv4_mapping.get_mut(ip), + IpAddr::V6(ip) => self.clients_ipv6_mapping.get_mut(ip), + } } fn is_nym_address_connected(&self, nym_address: &Recipient) -> bool { - self.clients + self.clients_ipv4_mapping .values() .any(|client| client.nym_address == *nym_address) } - fn lookup_ip_from_nym_address(&self, nym_address: &Recipient) -> Option { - self.clients.iter().find_map(|(ip, client)| { - if client.nym_address == *nym_address { - Some(*ip) - } else { - None - } - }) + fn lookup_ip_from_nym_address(&self, nym_address: &Recipient) -> Option { + self.clients_ipv4_mapping + .iter() + .find_map(|(ipv4, connected_client)| { + if connected_client.nym_address == *nym_address { + Some(IPPair { + ipv4: *ipv4, + ipv6: connected_client.ipv6, + }) + } else { + None + } + }) } fn lookup_client_from_nym_address(&self, nym_address: &Recipient) -> Option<&ConnectedClient> { - self.clients - .values() - .find(|client| client.nym_address == *nym_address) + self.clients_ipv4_mapping + .iter() + .find_map(|(_, connected_client)| { + if connected_client.nym_address == *nym_address { + Some(connected_client) + } else { + None + } + }) } fn connect( &mut self, - ip: IpAddr, + ips: IPPair, nym_address: Recipient, mix_hops: Option, forward_from_tun_tx: tokio::sync::mpsc::UnboundedSender>, @@ -96,21 +114,25 @@ impl ConnectedClients { ) { // 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. - self.clients.insert( - ip, - ConnectedClient { + let client = ConnectedClient { + nym_address, + ipv6: ips.ipv6, + mix_hops, + last_activity: Arc::new(RwLock::new(std::time::Instant::now())), + _close_tx: Arc::new(CloseTx { nym_address, - mix_hops, - last_activity: std::time::Instant::now(), - close_tx: Some(close_tx), - handle, - }, - ); + inner: Some(close_tx), + }), + handle: Arc::new(handle), + }; + log::info!("Inserting {} and {}", ips.ipv4, ips.ipv6); + self.clients_ipv4_mapping.insert(ips.ipv4, client.clone()); + self.clients_ipv6_mapping.insert(ips.ipv6, client); // Send the connected client info to the tun listener, which will use it to forward packets // to the connected client handler. self.tun_listener_connected_client_tx .send(ConnectedClientEvent::Connect(Box::new(ConnectEvent { - ip, + ips, forward_from_tun_tx, }))) .tap_err(|err| { @@ -119,9 +141,9 @@ impl ConnectedClients { .ok(); } - fn update_activity(&mut self, ip: &IpAddr) -> Result<()> { - if let Some(client) = self.clients.get_mut(ip) { - client.last_activity = std::time::Instant::now(); + async fn update_activity(&mut self, ips: &IPPair) -> Result<()> { + if let Some(client) = self.clients_ipv4_mapping.get(&ips.ipv4) { + *client.last_activity.write().await = std::time::Instant::now(); Ok(()) } else { Err(IpPacketRouterError::FailedToUpdateClientActivity) @@ -129,12 +151,18 @@ impl ConnectedClients { } // Identify connected client handlers that have stopped without being told to stop - fn get_finished_client_handlers(&mut self) -> Vec<(IpAddr, Recipient)> { - self.clients + fn get_finished_client_handlers(&mut self) -> Vec<(IPPair, Recipient)> { + self.clients_ipv4_mapping .iter_mut() - .filter_map(|(ip, client)| { - if client.handle.is_finished() { - Some((*ip, client.nym_address)) + .filter_map(|(ip, connected_client)| { + if connected_client.handle.is_finished() { + Some(( + IPPair { + ipv4: *ip, + ipv6: connected_client.ipv6, + }, + connected_client.nym_address, + )) } else { None } @@ -142,26 +170,32 @@ impl ConnectedClients { .collect() } - fn get_inactive_clients(&mut self) -> Vec<(IpAddr, Recipient)> { + async fn get_inactive_clients(&mut self) -> Vec<(IPPair, Recipient)> { let now = std::time::Instant::now(); - self.clients - .iter() - .filter_map(|(ip, client)| { - if now.duration_since(client.last_activity) > CLIENT_MIXNET_INACTIVITY_TIMEOUT { - Some((*ip, client.nym_address)) - } else { - None - } - }) - .collect() + let mut ret = vec![]; + for (ip, connected_client) in self.clients_ipv4_mapping.iter() { + if now.duration_since(*connected_client.last_activity.read().await) + > CLIENT_MIXNET_INACTIVITY_TIMEOUT + { + ret.push(( + IPPair { + ipv4: *ip, + ipv6: connected_client.ipv6, + }, + connected_client.nym_address, + )) + } + } + ret } - fn disconnect_stopped_client_handlers(&mut self, stopped_clients: Vec<(IpAddr, Recipient)>) { - for (ip, _) in &stopped_clients { - log::info!("Disconnect stopped client: {ip}"); - self.clients.remove(ip); + fn disconnect_stopped_client_handlers(&mut self, stopped_clients: Vec<(IPPair, Recipient)>) { + for (ips, _) in &stopped_clients { + log::info!("Disconnect stopped client: {ips}"); + self.clients_ipv4_mapping.remove(&ips.ipv4); + self.clients_ipv6_mapping.remove(&ips.ipv6); self.tun_listener_connected_client_tx - .send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ip))) + .send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ips))) .tap_err(|err| { log::error!("Failed to send disconnect event: {err}"); }) @@ -169,12 +203,13 @@ impl ConnectedClients { } } - fn disconnect_inactive_clients(&mut self, inactive_clients: Vec<(IpAddr, Recipient)>) { - for (ip, _) in &inactive_clients { - log::info!("Disconnect inactive client: {ip}"); - self.clients.remove(ip); + fn disconnect_inactive_clients(&mut self, inactive_clients: Vec<(IPPair, Recipient)>) { + for (ips, _) in &inactive_clients { + log::info!("Disconnect inactive client: {ips}"); + self.clients_ipv4_mapping.remove(&ips.ipv4); + self.clients_ipv6_mapping.remove(&ips.ipv6); self.tun_listener_connected_client_tx - .send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ip))) + .send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ips))) .tap_err(|err| { log::error!("Failed to send disconnect event: {err}"); }) @@ -182,40 +217,49 @@ impl ConnectedClients { } } - fn find_new_ip(&self) -> Option { - generate_new_ip::find_new_ip(&self.clients) + fn find_new_ip(&self) -> Option { + generate_new_ip::find_new_ips(&self.clients_ipv4_mapping, &self.clients_ipv6_mapping) } } +pub(crate) struct CloseTx { + pub(crate) nym_address: Recipient, + // 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) inner: Option>, +} + +#[derive(Clone)] 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, + // The assigned IPv6 address of this client + pub(crate) ipv6: Ipv6Addr, + // 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, + pub(crate) last_activity: Arc>, - // 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>, + pub(crate) _close_tx: Arc, // Handle for the connected client handler - pub(crate) handle: tokio::task::JoinHandle<()>, + pub(crate) handle: Arc>, } impl ConnectedClient { - fn update_activity(&mut self) { - self.last_activity = std::time::Instant::now(); + async fn update_activity(&self) { + *self.last_activity.write().await = std::time::Instant::now(); } } -impl Drop for ConnectedClient { +impl Drop for CloseTx { fn drop(&mut self) { log::debug!("signal to close client: {}", self.nym_address); - if let Some(close_tx) = self.close_tx.take() { + if let Some(close_tx) = self.inner.take() { close_tx.send(()).ok(); } } @@ -259,7 +303,7 @@ impl MixnetListener { ); let request_id = connect_request.request_id; - let requested_ip = connect_request.ip; + let requested_ips = connect_request.ips; let reply_to = connect_request.reply_to; let reply_to_hops = connect_request.reply_to_hops; // TODO: add to connect request @@ -267,7 +311,7 @@ impl MixnetListener { // 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); + let is_ip_taken = self.connected_clients.is_ip_connected(&requested_ips); // Check that the nym address isn't already registered let is_nym_address_taken = self.connected_clients.is_nym_address_connected(&reply_to); @@ -277,7 +321,8 @@ impl MixnetListener { log::info!("Connecting an already connected client"); if self .connected_clients - .update_activity(&requested_ip) + .update_activity(&requested_ips) + .await .is_err() { log::error!("Failed to update activity for client"); @@ -300,7 +345,7 @@ impl MixnetListener { // Register the new client in the set of connected clients self.connected_clients.connect( - requested_ip, + requested_ips, reply_to, reply_to_hops, forward_from_tun_tx, @@ -350,11 +395,12 @@ impl MixnetListener { // 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.lookup_ip_from_nym_address(&reply_to) { + if let Some(existing_ips) = self.connected_clients.lookup_ip_from_nym_address(&reply_to) { log::info!("Found existing client for nym address"); if self .connected_clients - .update_activity(&existing_ip) + .update_activity(&existing_ips) + .await .is_err() { log::error!("Failed to update activity for client"); @@ -362,11 +408,11 @@ impl MixnetListener { return Ok(Some(IpPacketResponse::new_dynamic_connect_success( request_id, reply_to, - existing_ip, + existing_ips, ))); } - let Some(new_ip) = self.connected_clients.find_new_ip() else { + let Some(new_ips) = self.connected_clients.find_new_ip() else { log::info!("No available IP address"); return Ok(Some(IpPacketResponse::new_dynamic_connect_failure( request_id, @@ -386,7 +432,7 @@ impl MixnetListener { // Register the new client in the set of connected clients self.connected_clients.connect( - new_ip, + new_ips, reply_to, reply_to_hops, forward_from_tun_tx, @@ -394,7 +440,7 @@ impl MixnetListener { handle, ); Ok(Some(IpPacketResponse::new_dynamic_connect_success( - request_id, reply_to, new_ip, + request_id, reply_to, new_ips, ))) } @@ -428,7 +474,7 @@ impl MixnetListener { if let Some(connected_client) = self.connected_clients.get_client_from_ip_mut(&src_addr) { // Keep track of activity so we can disconnect inactive clients - connected_client.update_activity(); + connected_client.update_activity().await; // For packets without a port, use 0. let dst = dst.unwrap_or_else(|| SocketAddr::new(dst_addr, 0)); @@ -539,9 +585,9 @@ impl MixnetListener { } } - fn handle_disconnect_timer(&mut self) { + async fn handle_disconnect_timer(&mut self) { let stopped_clients = self.connected_clients.get_finished_client_handlers(); - let inactive_clients = self.connected_clients.get_inactive_clients(); + let inactive_clients = self.connected_clients.get_inactive_clients().await; // TODO: Send disconnect responses to all disconnected clients //for (ip, nym_address) in stopped_clients.iter().chain(disconnected_clients.iter()) { @@ -571,10 +617,14 @@ impl MixnetListener { })?; // We could avoid this lookup if we check this when we create the response. - let mix_hops = self + let mix_hops = if let Some(c) = self .connected_clients .lookup_client_from_nym_address(recipient) - .and_then(|c| c.mix_hops); + { + c.mix_hops + } else { + None + }; let input_message = create_input_message(*recipient, response_packet, mix_hops); self.mixnet_client @@ -613,7 +663,7 @@ impl MixnetListener { log::debug!("IpPacketRouter [main loop]: received shutdown"); }, _ = disconnect_timer.tick() => { - self.handle_disconnect_timer(); + self.handle_disconnect_timer().await; }, msg = self.mixnet_client.next() => { if let Some(msg) = msg { @@ -642,9 +692,9 @@ pub(crate) enum ConnectedClientEvent { Connect(Box), } -pub(crate) struct DisconnectEvent(pub(crate) IpAddr); +pub(crate) struct DisconnectEvent(pub(crate) IPPair); pub(crate) struct ConnectEvent { - pub(crate) ip: IpAddr, + pub(crate) ips: IPPair, pub(crate) forward_from_tun_tx: tokio::sync::mpsc::UnboundedSender>, } diff --git a/service-providers/ip-packet-router/src/tun_listener.rs b/service-providers/ip-packet-router/src/tun_listener.rs index 34fb12526e..8d2ba41371 100644 --- a/service-providers/ip-packet-router/src/tun_listener.rs +++ b/service-providers/ip-packet-router/src/tun_listener.rs @@ -1,5 +1,7 @@ -use std::{collections::HashMap, net::IpAddr}; +use std::collections::HashMap; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use nym_ip_packet_requests::IPPair; use nym_task::TaskClient; #[cfg(target_os = "linux")] use tokio::io::AsyncReadExt; @@ -15,10 +17,12 @@ use crate::{ // It's even ok if this is slightly out of date pub(crate) struct ConnectedClientMirror { pub(crate) forward_from_tun_tx: tokio::sync::mpsc::UnboundedSender>, + pub(crate) ips: IPPair, } pub(crate) struct ConnectedClientsListener { - clients: HashMap, + clients_ipv4: HashMap, + clients_ipv6: HashMap, connected_client_rx: tokio::sync::mpsc::UnboundedReceiver, } @@ -30,35 +34,48 @@ impl ConnectedClientsListener { >, ) -> Self { ConnectedClientsListener { - clients: HashMap::new(), + clients_ipv4: HashMap::new(), + clients_ipv6: HashMap::new(), connected_client_rx, } } pub(crate) fn get(&self, ip: &IpAddr) -> Option<&ConnectedClientMirror> { - self.clients.get(ip) + match ip { + IpAddr::V4(ip) => self.clients_ipv4.get(ip), + IpAddr::V6(ip) => self.clients_ipv6.get(ip), + } } pub(crate) fn update(&mut self, event: mixnet_listener::ConnectedClientEvent) { match event { mixnet_listener::ConnectedClientEvent::Connect(connected_event) => { let mixnet_listener::ConnectEvent { - ip, + ips, forward_from_tun_tx, } = *connected_event; - log::trace!("Connect client: {ip}"); - self.clients.insert( - ip, + log::trace!("Connect client: {ips}"); + self.clients_ipv4.insert( + ips.ipv4, + ConnectedClientMirror { + forward_from_tun_tx: forward_from_tun_tx.clone(), + ips, + }, + ); + self.clients_ipv6.insert( + ips.ipv6, ConnectedClientMirror { forward_from_tun_tx, + ips, }, ); } mixnet_listener::ConnectedClientEvent::Disconnect( - mixnet_listener::DisconnectEvent(ip), + mixnet_listener::DisconnectEvent(ips), ) => { - log::trace!("Disconnect client: {ip}"); - self.clients.remove(&ip); + log::trace!("Disconnect client: {ips}"); + self.clients_ipv4.remove(&ips.ipv4); + self.clients_ipv6.remove(&ips.ipv6); } } } @@ -82,6 +99,7 @@ impl TunListener { if let Some(ConnectedClientMirror { forward_from_tun_tx, + ips, }) = self.connected_clients.get(&dst_addr) { let packet = buf[..len].to_vec(); @@ -89,7 +107,7 @@ impl TunListener { log::warn!("Failed to forward packet to connected client {dst_addr}: disconnecting it from tun listener"); self.connected_clients .update(mixnet_listener::ConnectedClientEvent::Disconnect( - mixnet_listener::DisconnectEvent(dst_addr), + mixnet_listener::DisconnectEvent(*ips), )); } } else { 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 5824ce7bba..8e7a688853 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 @@ -1,38 +1,55 @@ -use std::{ - collections::HashMap, - net::{IpAddr, Ipv4Addr}, -}; +use nym_ip_packet_requests::IPPair; +use std::net::Ipv6Addr; +use std::{collections::HashMap, net::Ipv4Addr}; -use crate::{constants::TUN_DEVICE_ADDRESS, mixnet_listener::ConnectedClient}; +use crate::constants::{TUN_DEVICE_ADDRESS_V4, TUN_DEVICE_ADDRESS_V6}; // Find an available IP address in self.connected_clients // TODO: make this nicer -fn generate_random_ip_within_subnet() -> Ipv4Addr { +fn generate_random_ips_within_subnet() -> IPPair { let mut rng = rand::thread_rng(); // Generate a random number in the range 1-254 let last_octet = rand::Rng::gen_range(&mut rng, 1..=254); - Ipv4Addr::new(10, 0, 0, last_octet) + let ipv4 = Ipv4Addr::new(10, 0, 0, last_octet); + let ipv6 = Ipv6Addr::new(0x2001, 0x0db8, 0xa160, 0, 0, 0, 0, last_octet as u16); + IPPair { ipv4, ipv6 } } -fn is_ip_taken( - connected_clients: &HashMap, - tun_ip: Ipv4Addr, - ip: Ipv4Addr, +fn is_ip_taken( + connected_clients_ipv4: &HashMap, + connected_clients_ipv6: &HashMap, + tun_ips: IPPair, + ips: IPPair, ) -> bool { - connected_clients.contains_key(&ip.into()) || ip == tun_ip + connected_clients_ipv4.contains_key(&ips.ipv4) + || connected_clients_ipv6.contains_key(&ips.ipv6) + || ips.ipv4 == tun_ips.ipv4 + || ips.ipv6 == tun_ips.ipv6 } // TODO: brute force approach. We could consider using a more efficient algorithm -pub(crate) fn find_new_ip(connected_clients: &HashMap) -> Option { - let mut new_ip = generate_random_ip_within_subnet(); +pub(crate) fn find_new_ips( + connected_clients_ipv4: &HashMap, + connected_clients_ipv6: &HashMap, +) -> Option { + let mut new_ips = generate_random_ips_within_subnet(); let mut tries = 0; - let tun_ip = TUN_DEVICE_ADDRESS.parse::().unwrap(); - while is_ip_taken(connected_clients, tun_ip, new_ip) { - new_ip = generate_random_ip_within_subnet(); + let tun_ips = IPPair { + ipv4: TUN_DEVICE_ADDRESS_V4, + ipv6: TUN_DEVICE_ADDRESS_V6, + }; + + while is_ip_taken( + connected_clients_ipv4, + connected_clients_ipv6, + tun_ips, + new_ips, + ) { + new_ips = generate_random_ips_within_subnet(); tries += 1; if tries > 100 { return None; } } - Some(new_ip.into()) + Some(new_ips) }