From 4a8c09f47693128b035c62b07ffa46f56d66e8fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 27 Nov 2023 15:31:35 +0100 Subject: [PATCH] Handle dynamic ip allocation in ip packet router (#4186) * Add dynamic connect support to ip packet router * Disconnect inactive clients * Don't generate ip same as tun device * clippy * Extract a few functions to separate mod * clippy --- Cargo.lock | 1 + common/ip-packet-requests/Cargo.toml | 1 + common/ip-packet-requests/src/lib.rs | 52 +++++- .../ip-packet-router/src/generate_new_ip.rs | 38 +++++ service-providers/ip-packet-router/src/lib.rs | 152 ++++++++++++++---- 5 files changed, 208 insertions(+), 36 deletions(-) create mode 100644 service-providers/ip-packet-router/src/generate_new_ip.rs diff --git a/Cargo.lock b/Cargo.lock index 5c0711c668..7819de6ae9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6610,6 +6610,7 @@ dependencies = [ "nym-sphinx", "rand 0.8.5", "serde", + "thiserror", ] [[package]] diff --git a/common/ip-packet-requests/Cargo.toml b/common/ip-packet-requests/Cargo.toml index 969152baf7..2a92715f82 100644 --- a/common/ip-packet-requests/Cargo.toml +++ b/common/ip-packet-requests/Cargo.toml @@ -16,3 +16,4 @@ bytes = "1.5.0" nym-sphinx = { path = "../nymsphinx" } rand = "0.8.5" serde = { workspace = true, features = ["derive"] } +thiserror = { workspace = true } diff --git a/common/ip-packet-requests/src/lib.rs b/common/ip-packet-requests/src/lib.rs index 95dc67076d..290deec697 100644 --- a/common/ip-packet-requests/src/lib.rs +++ b/common/ip-packet-requests/src/lib.rs @@ -146,13 +146,17 @@ impl IpPacketResponse { } } - pub fn new_static_connect_failure(request_id: u64, reply_to: Recipient) -> Self { + pub fn new_static_connect_failure( + request_id: u64, + reply_to: Recipient, + reason: StaticConnectFailureReason, + ) -> Self { Self { version: CURRENT_VERSION, data: IpPacketResponseData::StaticConnect(StaticConnectResponse { request_id, reply_to, - reply: StaticConnectResponseReply::Failure, + reply: StaticConnectResponseReply::Failure(reason), }), } } @@ -168,13 +172,17 @@ impl IpPacketResponse { } } - pub fn new_dynamic_connect_failure(request_id: u64, reply_to: Recipient) -> Self { + pub fn new_dynamic_connect_failure( + request_id: u64, + reply_to: Recipient, + reason: DynamicConnectFailureReason, + ) -> Self { Self { version: CURRENT_VERSION, data: IpPacketResponseData::DynamicConnect(DynamicConnectResponse { request_id, reply_to, - reply: DynamicConnectResponseReply::Failure, + reply: DynamicConnectResponseReply::Failure(reason), }), } } @@ -233,17 +241,28 @@ pub struct StaticConnectResponse { #[derive(Clone, Debug, Serialize, Deserialize)] pub enum StaticConnectResponseReply { Success, - Failure, + Failure(StaticConnectFailureReason), } + impl StaticConnectResponseReply { pub fn is_success(&self) -> bool { match self { StaticConnectResponseReply::Success => true, - StaticConnectResponseReply::Failure => false, + StaticConnectResponseReply::Failure(_) => false, } } } +#[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)] +pub enum StaticConnectFailureReason { + #[error("requested ip address is already in use")] + RequestedIpAlreadyInUse, + #[error("requested nym-address is already in use")] + RequestedNymAddressAlreadyInUse, + #[error("{0}")] + Other(String), +} + #[derive(Clone, Debug, Serialize, Deserialize)] pub struct DynamicConnectResponse { pub request_id: u64, @@ -254,7 +273,16 @@ pub struct DynamicConnectResponse { #[derive(Clone, Debug, Serialize, Deserialize)] pub enum DynamicConnectResponseReply { Success(DynamicConnectSuccess), - Failure, + Failure(DynamicConnectFailureReason), +} + +impl DynamicConnectResponseReply { + pub fn is_success(&self) -> bool { + match self { + DynamicConnectResponseReply::Success(_) => true, + DynamicConnectResponseReply::Failure(_) => false, + } + } } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -262,6 +290,16 @@ pub struct DynamicConnectSuccess { pub ip: IpAddr, } +#[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)] +pub enum DynamicConnectFailureReason { + #[error("requested nym-address is already in use")] + RequestedNymAddressAlreadyInUse, + #[error("no available ip address")] + NoAvailableIp, + #[error("{0}")] + Other(String), +} + #[derive(Clone, Debug, Serialize, Deserialize)] pub struct DataResponse { pub ip_packet: bytes::Bytes, diff --git a/service-providers/ip-packet-router/src/generate_new_ip.rs b/service-providers/ip-packet-router/src/generate_new_ip.rs new file mode 100644 index 0000000000..6a86ce664f --- /dev/null +++ b/service-providers/ip-packet-router/src/generate_new_ip.rs @@ -0,0 +1,38 @@ +use std::{ + collections::HashMap, + net::{IpAddr, Ipv4Addr}, +}; + +use crate::{ConnectedClient, TUN_DEVICE_ADDRESS}; + +// Find an available IP address in self.connected_clients +// TODO: make this nicer +fn generate_random_ip_within_subnet() -> Ipv4Addr { + 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) +} + +fn is_ip_taken( + connected_clients: &HashMap, + tun_ip: Ipv4Addr, + ip: Ipv4Addr, +) -> bool { + connected_clients.contains_key(&ip.into()) || ip == tun_ip +} + +// 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(); + 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(); + tries += 1; + if tries > 100 { + return None; + } + } + Some(new_ip.into()) +} diff --git a/service-providers/ip-packet-router/src/lib.rs b/service-providers/ip-packet-router/src/lib.rs index 528b575198..4cb4b0a580 100644 --- a/service-providers/ip-packet-router/src/lib.rs +++ b/service-providers/ip-packet-router/src/lib.rs @@ -4,6 +4,7 @@ use std::{ collections::HashMap, net::{IpAddr, SocketAddr}, path::Path, + time::Duration, }; use error::IpPacketRouterError; @@ -12,7 +13,10 @@ use nym_client_core::{ client::mix_traffic::transceiver::GatewayTransceiver, config::disk_persistence::CommonClientPaths, HardcodedTopologyProvider, TopologyProvider, }; -use nym_ip_packet_requests::{IpPacketRequest, IpPacketRequestData, IpPacketResponse}; +use nym_ip_packet_requests::{ + DynamicConnectFailureReason, IpPacketRequest, IpPacketRequestData, IpPacketResponse, + StaticConnectFailureReason, +}; use nym_sdk::{ mixnet::{InputMessage, MixnetMessageSender, Recipient}, NymNetworkDetails, @@ -27,6 +31,7 @@ pub use crate::config::Config; pub mod config; pub mod error; +mod generate_new_ip; mod request_filter; // The interface used to route traffic @@ -34,6 +39,9 @@ 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"; +const DISCONNECT_TIMER_INTERVAL: Duration = Duration::from_secs(10); +const CLIENT_INACTIVITY_TIMEOUT: Duration = Duration::from_secs(5 * 60); + pub struct OnStartData { // to add more fields as required pub address: Recipient, @@ -216,36 +224,57 @@ impl IpPacketRouter { // 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 - if self.connected_clients.contains_key(&requested_ip) { - log::info!("Requested IP is not available"); - return Ok(Some(IpPacketResponse::new_static_connect_failure( - request_id, reply_to, - ))); - } + let is_ip_taken = self.connected_clients.contains_key(&requested_ip); // Check that the nym address isn't already registered - if self + let is_nym_address_taken = self .connected_clients .values() - .any(|client| client.nym_address == reply_to) - { - log::info!("Nym address is already registered"); - return Ok(Some(IpPacketResponse::new_static_connect_failure( - request_id, reply_to, - ))); + .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(), + }, + ); + 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, + ))) + } } - - // Insert the new connected client - let connected_client = ConnectedClient { - nym_address: reply_to, - last_activity: std::time::Instant::now(), - }; - self.connected_clients - .insert(requested_ip, connected_client); - - Ok(Some(IpPacketResponse::new_static_connect_success( - request_id, reply_to, - ))) } async fn on_dynamic_connect_request( @@ -256,8 +285,56 @@ impl IpPacketRouter { "Received dynamic connect request from {sender_address}", sender_address = connect_request.reply_to ); - log::info!("Dropping request: dynamic connect requests are not yet supported"); - Ok(None) + + 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(), + }, + ); + Ok(Some(IpPacketResponse::new_dynamic_connect_success( + request_id, reply_to, new_ip, + ))) } async fn on_data_request( @@ -348,12 +425,29 @@ impl IpPacketRouter { 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); + } + }, msg = self.mixnet_client.next() => { if let Some(msg) = msg { match self.on_reconstructed_message(msg).await { @@ -389,7 +483,7 @@ impl IpPacketRouter { }, packet = self.tun_task_response_rx.recv() => { if let Some((_tag, packet)) = packet { - // TODO: skip full parsing we we only need dst_addr + // TODO: skip full parsing since we only need dst_addr let Ok(ParsedPacket { packet_type: _, src_addr: _,