From dccdde108c474277449533b44dee0f7d8e78ab8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 30 Jan 2026 11:38:17 +0000 Subject: [PATCH] Lp/bugfix/share ip allocation (#6395) * feat: use shared PeerManager between Authenticator and LpHandlerState * feat: share IpPool * clippy and test fixes * PR suggestions --- Cargo.lock | 2 +- .../src/v6/registration.rs | 3 +- .../tests/src/v2/peer_controller.rs | 14 +- common/wireguard/Cargo.toml | 2 +- common/wireguard/src/ip_pool/compat.rs | 60 +++++++++ .../src/{ip_pool.rs => ip_pool/mod.rs} | 68 ++++++++-- common/wireguard/src/lib.rs | 51 +++---- common/wireguard/src/peer_controller/mock.rs | 32 ++--- common/wireguard/src/peer_controller/mod.rs | 119 ++++++++++------ gateway/src/error.rs | 4 + .../authenticator/error.rs | 7 +- .../authenticator/mixnet_listener.rs | 61 ++++----- .../authenticator/mod.rs | 27 +--- gateway/src/node/lp_listener/handler.rs | 2 +- gateway/src/node/lp_listener/mod.rs | 44 +++--- gateway/src/node/lp_listener/peer_manager.rs | 127 ------------------ gateway/src/node/lp_listener/registration.rs | 16 ++- gateway/src/node/mod.rs | 19 +-- gateway/src/node/wireguard/error.rs | 19 +++ gateway/src/node/wireguard/mod.rs | 8 ++ .../peer_manager.rs | 104 +++++++++----- integration-tests/src/lp_registration.rs | 14 +- 22 files changed, 429 insertions(+), 374 deletions(-) create mode 100644 common/wireguard/src/ip_pool/compat.rs rename common/wireguard/src/{ip_pool.rs => ip_pool/mod.rs} (76%) delete mode 100644 gateway/src/node/lp_listener/peer_manager.rs create mode 100644 gateway/src/node/wireguard/error.rs create mode 100644 gateway/src/node/wireguard/mod.rs rename gateway/src/node/{internal_service_providers/authenticator => wireguard}/peer_manager.rs (84%) diff --git a/Cargo.lock b/Cargo.lock index f68e8f7c03..2e041700b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8375,12 +8375,12 @@ dependencies = [ "futures", "ip_network", "ipnetwork", + "nym-authenticator-requests", "nym-credential-verification", "nym-credentials-interface", "nym-crypto", "nym-gateway-requests", "nym-gateway-storage", - "nym-ip-packet-requests", "nym-metrics", "nym-network-defaults", "nym-node-metrics", diff --git a/common/authenticator-requests/src/v6/registration.rs b/common/authenticator-requests/src/v6/registration.rs index 11fcf34116..14e7462c05 100644 --- a/common/authenticator-requests/src/v6/registration.rs +++ b/common/authenticator-requests/src/v6/registration.rs @@ -20,13 +20,12 @@ use nym_crypto::asymmetric::x25519::{PrivateKey, PublicKey}; use sha2::Sha256; pub type PendingRegistrations = HashMap; -pub type PrivateIPs = HashMap; +pub type PrivateIPs = HashMap; #[cfg(feature = "verify")] pub type HmacSha256 = Hmac; pub type Nonce = u64; -pub type Taken = Option; #[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct IpPair { diff --git a/common/wireguard-private-metadata/tests/src/v2/peer_controller.rs b/common/wireguard-private-metadata/tests/src/v2/peer_controller.rs index 68fd5cdeed..15312a1f52 100644 --- a/common/wireguard-private-metadata/tests/src/v2/peer_controller.rs +++ b/common/wireguard-private-metadata/tests/src/v2/peer_controller.rs @@ -26,7 +26,7 @@ impl From<&PeerControlRequest> for PeerControlRequestTypeV2 { fn from(req: &PeerControlRequest) -> Self { match req { PeerControlRequest::AddPeer { .. } => PeerControlRequestTypeV2::AddPeer, - PeerControlRequest::RegisterPeer { .. } => PeerControlRequestTypeV2::AddPeer, + PeerControlRequest::AllocatePeerIpPair { .. } => PeerControlRequestTypeV2::AddPeer, PeerControlRequest::RemovePeer { .. } => PeerControlRequestTypeV2::RemovePeer, PeerControlRequest::QueryPeer { .. } => PeerControlRequestTypeV2::QueryPeer, PeerControlRequest::GetClientBandwidthByKey { .. } => { @@ -41,6 +41,7 @@ impl From<&PeerControlRequest> for PeerControlRequestTypeV2 { PeerControlRequest::GetVerifierByIp { ip, .. } => { PeerControlRequestTypeV2::GetVerifierByIp { ip: *ip } } + PeerControlRequest::ReleaseIpPair { .. } => unreachable!(), } } } @@ -113,7 +114,7 @@ impl MockPeerControllerV2 { ) .unwrap(); } - PeerControlRequest::RegisterPeer { response_tx, .. } => { + PeerControlRequest::AllocatePeerIpPair { response_tx, .. } => { response_tx .send( *response @@ -176,6 +177,15 @@ impl MockPeerControllerV2 { ) .ok(); } + PeerControlRequest::ReleaseIpPair { response_tx, .. } => { + response_tx + .send( + *response + .downcast() + .expect("registered response has mismatched type"), + ) + .ok(); + } } } diff --git a/common/wireguard/Cargo.toml b/common/wireguard/Cargo.toml index 16b2536f84..3857d1165a 100644 --- a/common/wireguard/Cargo.toml +++ b/common/wireguard/Cargo.toml @@ -28,7 +28,7 @@ nym-credential-verification = { workspace = true } nym-crypto = { workspace = true, features = ["asymmetric"] } nym-gateway-storage = { workspace = true } nym-gateway-requests = { workspace = true } -nym-ip-packet-requests = { workspace = true } +nym-authenticator-requests = { workspace = true } nym-metrics = { workspace = true } nym-network-defaults = { workspace = true } nym-task = { workspace = true } diff --git a/common/wireguard/src/ip_pool/compat.rs b/common/wireguard/src/ip_pool/compat.rs new file mode 100644 index 0000000000..95ae7da686 --- /dev/null +++ b/common/wireguard/src/ip_pool/compat.rs @@ -0,0 +1,60 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::ip_pool::IpPair; + +impl From for nym_authenticator_requests::v6::registration::IpPair { + fn from(ip_pair: IpPair) -> Self { + nym_authenticator_requests::v6::registration::IpPair { + ipv4: ip_pair.ipv4, + ipv6: ip_pair.ipv6, + } + } +} + +impl From for nym_authenticator_requests::v5::registration::IpPair { + fn from(ip_pair: IpPair) -> Self { + nym_authenticator_requests::v5::registration::IpPair { + ipv4: ip_pair.ipv4, + ipv6: ip_pair.ipv6, + } + } +} + +impl From for nym_authenticator_requests::v4::registration::IpPair { + fn from(ip_pair: IpPair) -> Self { + nym_authenticator_requests::v4::registration::IpPair { + ipv4: ip_pair.ipv4, + ipv6: ip_pair.ipv6, + } + } +} + +// + +impl From for IpPair { + fn from(ip_pair: nym_authenticator_requests::v6::registration::IpPair) -> Self { + IpPair { + ipv4: ip_pair.ipv4, + ipv6: ip_pair.ipv6, + } + } +} + +impl From for IpPair { + fn from(ip_pair: nym_authenticator_requests::v5::registration::IpPair) -> Self { + IpPair { + ipv4: ip_pair.ipv4, + ipv6: ip_pair.ipv6, + } + } +} + +impl From for IpPair { + fn from(ip_pair: nym_authenticator_requests::v4::registration::IpPair) -> Self { + IpPair { + ipv4: ip_pair.ipv4, + ipv6: ip_pair.ipv6, + } + } +} diff --git a/common/wireguard/src/ip_pool.rs b/common/wireguard/src/ip_pool/mod.rs similarity index 76% rename from common/wireguard/src/ip_pool.rs rename to common/wireguard/src/ip_pool/mod.rs index e1c2b0453f..b34a7134c2 100644 --- a/common/wireguard/src/ip_pool.rs +++ b/common/wireguard/src/ip_pool/mod.rs @@ -1,14 +1,56 @@ // Copyright 2024 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +mod compat; + +use defguard_wireguard_rs::host::Peer; use ipnetwork::IpNetwork; -use nym_ip_packet_requests::IpPair; use rand::seq::IteratorRandom; use std::collections::HashMap; +use std::fmt::{Display, Formatter}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use std::sync::Arc; -use std::time::SystemTime; +use std::time::{Duration, SystemTime}; use tokio::sync::RwLock; +use tokio::time::Instant; +use tracing::{trace, warn}; + +// helper to convert peer's allocation into an `IpPair` +pub fn allocated_ip_pair(peer: &Peer) -> Option { + for allowed_ip in &peer.allowed_ips { + // Extract IPv4 and IPv6 from peer's allowed_ips + if let IpAddr::V4(ipv4) = allowed_ip.address { + // Find corresponding IPv6 + if let Some(ipv6_mask) = peer + .allowed_ips + .iter() + .find(|ip| matches!(ip.address, IpAddr::V6(_))) + && let IpAddr::V6(ipv6) = ipv6_mask.address + { + return Some(IpPair::new(ipv4, ipv6)); + } + } + } + None +} + +#[derive(Copy, Clone, Debug, PartialEq, PartialOrd, Eq, Hash)] +pub struct IpPair { + pub ipv4: Ipv4Addr, + pub ipv6: Ipv6Addr, +} + +impl IpPair { + pub fn new(ipv4: Ipv4Addr, ipv6: Ipv6Addr) -> Self { + IpPair { ipv4, ipv6 } + } +} + +impl Display for IpPair { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "IPv4: {}, IPv6: {}", self.ipv4, self.ipv6) + } +} /// Represents the state of an IP allocation #[derive(Debug, Clone, Copy)] @@ -83,12 +125,8 @@ impl IpPool { } tracing::info!( - "Initialized IP pool with {} address pairs from {}/{} and {}/{}", + "Initialized IP pool with {} address pairs from {ipv4_network}/{ipv4_prefix} and {ipv6_network}/{ipv6_prefix}", allocations.len(), - ipv4_network, - ipv4_prefix, - ipv6_network, - ipv6_prefix ); Ok(IpPool { @@ -98,24 +136,28 @@ impl IpPool { /// Allocate a free IP pair from the pool /// - /// Randomly selects an available IP pair and marks it as allocated. - /// /// # Errors /// Returns `IpPoolError::NoFreeIp` if no IPs are available pub async fn allocate(&self) -> Result { let mut pool = self.allocations.write().await; // Find a free IP and allocate it + let assignment_start = Instant::now(); let free_ip = pool .iter_mut() .filter(|(_, state)| matches!(state, AllocationState::Free)) .choose(&mut rand::thread_rng()) .ok_or(IpPoolError::NoFreeIp)?; + let taken = assignment_start.elapsed(); + trace!("assigning free ip pair took {taken:?}"); + if taken > Duration::from_millis(500) { + warn!("assigning free ip pair took {taken:?}"); + } let ip_pair = *free_ip.0; *free_ip.1 = AllocationState::Allocated(SystemTime::now()); - tracing::debug!("Allocated IP pair: {}", ip_pair); + tracing::debug!("Allocated IP pair: {ip_pair}"); Ok(ip_pair) } @@ -126,7 +168,7 @@ impl IpPool { let mut pool = self.allocations.write().await; if let Some(state) = pool.get_mut(&ip_pair) { *state = AllocationState::Free; - tracing::debug!("Released IP pair: {}", ip_pair); + tracing::debug!("Released IP pair: {ip_pair}"); } } @@ -137,9 +179,9 @@ impl IpPool { let mut pool = self.allocations.write().await; if let Some(state) = pool.get_mut(&ip_pair) { *state = AllocationState::Allocated(SystemTime::now()); - tracing::debug!("Marked IP pair as used: {}", ip_pair); + tracing::debug!("Marked IP pair as used: {ip_pair}"); } else { - tracing::warn!("Attempted to mark unknown IP pair as used: {}", ip_pair); + tracing::warn!("Attempted to mark unknown IP pair as used: {ip_pair}"); } } diff --git a/common/wireguard/src/lib.rs b/common/wireguard/src/lib.rs index 459677b492..e73da973a2 100644 --- a/common/wireguard/src/lib.rs +++ b/common/wireguard/src/lib.rs @@ -2,9 +2,6 @@ // SPDX-License-Identifier: Apache-2.0 #![cfg_attr(not(target_os = "linux"), allow(dead_code))] -// #![warn(clippy::pedantic)] -// #![warn(clippy::expect_used)] -// #![warn(clippy::unwrap_used)] use defguard_wireguard_rs::{ WGApi, WireguardInterfaceApi, error::WireguardInterfaceError, host::Peer, key::Key, @@ -16,9 +13,6 @@ use std::sync::Arc; use tokio::sync::mpsc::{self, Receiver, Sender}; use tracing::error; -#[cfg(target_os = "linux")] -use nym_ip_packet_requests::IpPair; - #[cfg(target_os = "linux")] use nym_network_defaults::constants::WG_TUN_BASE_NAME; @@ -288,6 +282,22 @@ pub async fn start_wireguard( peer_bandwidth_managers.insert(peer.public_key.clone(), (bandwidth_manager, peer.clone())); } + // Initialize IP pool from configuration + info!("Initializing IP pool for WireGuard peer allocation"); + let ip_pool = IpPool::new( + wireguard_data.inner.config().private_ipv4, + wireguard_data.inner.config().private_network_prefix_v4, + wireguard_data.inner.config().private_ipv6, + wireguard_data.inner.config().private_network_prefix_v6, + )?; + + // Mark existing peer IPs as used in the pool + for peer in &peers { + if let Some(ip_pair) = crate::ip_pool::allocated_ip_pair(peer) { + ip_pool.mark_used(ip_pair).await; + } + } + wg_api.create_interface()?; let interface_config = InterfaceConfiguration { name: ifname.clone(), @@ -296,7 +306,7 @@ pub async fn start_wireguard( wireguard_data.inner.config().private_ipv4, ))], port: wireguard_data.inner.config().announced_tunnel_port, - peers: peers.clone(), // Clone since we need to use peers later to mark IPs as used + peers, mtu: None, }; info!( @@ -350,33 +360,6 @@ pub async fn start_wireguard( let host = wg_api.read_interface_data()?; - // Initialize IP pool from configuration - info!("Initializing IP pool for WireGuard peer allocation"); - let ip_pool = IpPool::new( - wireguard_data.inner.config().private_ipv4, - wireguard_data.inner.config().private_network_prefix_v4, - wireguard_data.inner.config().private_ipv6, - wireguard_data.inner.config().private_network_prefix_v6, - )?; - - // Mark existing peer IPs as used in the pool - for peer in &peers { - for allowed_ip in &peer.allowed_ips { - // Extract IPv4 and IPv6 from peer's allowed_ips - if let IpAddr::V4(ipv4) = allowed_ip.address { - // Find corresponding IPv6 - if let Some(ipv6_mask) = peer - .allowed_ips - .iter() - .find(|ip| matches!(ip.address, IpAddr::V6(_))) - && let IpAddr::V6(ipv6) = ipv6_mask.address - { - ip_pool.mark_used(IpPair::new(ipv4, ipv6)).await; - } - } - } - } - let wg_api = std::sync::Arc::new(wg_api); let mut controller = PeerController::new( ecash_manager, diff --git a/common/wireguard/src/peer_controller/mock.rs b/common/wireguard/src/peer_controller/mock.rs index 5398183127..89f2821b31 100644 --- a/common/wireguard/src/peer_controller/mock.rs +++ b/common/wireguard/src/peer_controller/mock.rs @@ -19,6 +19,7 @@ use std::sync::Arc; use tokio::sync::RwLock; use tokio::sync::mpsc::Receiver; +use crate::ip_pool::IpPair; pub use defguard_wireguard_rs::key::Key; pub fn mock_peer_controller( @@ -70,7 +71,8 @@ impl From<&Key> for KeyWrapper { #[derive(Hash, PartialOrd, PartialEq, Clone, Debug, Eq)] pub enum PeerControlRequestType { AddPeer { public_key: KeyWrapper }, - RegisterPeer { public_key: KeyWrapper }, + AllocatePeerIpPair {}, + ReleaseIpPair { ip_pair: IpPair }, RemovePeer { key: KeyWrapper }, QueryPeer { key: KeyWrapper }, GetClientBandwidthByKey { key: KeyWrapper }, @@ -83,7 +85,8 @@ impl PeerControlRequestType { pub fn peer_key(&self) -> Option { match self { PeerControlRequestType::AddPeer { public_key } => Some(public_key.clone()), - PeerControlRequestType::RegisterPeer { public_key } => Some(public_key.clone()), + PeerControlRequestType::AllocatePeerIpPair {} => None, + PeerControlRequestType::ReleaseIpPair { .. } => None, PeerControlRequestType::RemovePeer { key } => Some(key.clone()), PeerControlRequestType::QueryPeer { key } => Some(key.clone()), PeerControlRequestType::GetClientBandwidthByKey { key } => Some(key.clone()), @@ -104,11 +107,12 @@ impl From<&PeerControlRequest> for PeerControlRequestType { PeerControlRequest::AddPeer { peer, .. } => PeerControlRequestType::AddPeer { public_key: (&peer.public_key).into(), }, - PeerControlRequest::RegisterPeer { - registration_data, .. - } => PeerControlRequestType::RegisterPeer { - public_key: (®istration_data.public_key).into(), - }, + PeerControlRequest::AllocatePeerIpPair { .. } => { + PeerControlRequestType::AllocatePeerIpPair {} + } + PeerControlRequest::ReleaseIpPair { ip_pair, .. } => { + PeerControlRequestType::ReleaseIpPair { ip_pair: *ip_pair } + } PeerControlRequest::RemovePeer { key, .. } => { PeerControlRequestType::RemovePeer { key: key.into() } } @@ -166,9 +170,6 @@ pub struct MockPeerControllerState { #[derive(Clone, Default)] pub struct PeerState { - /// Has IpPair been allocated to the peer? - pub register_success: bool, - // in the future maybe we could extend it with `ClientBandwidth` information /// Has the client handle been spawned pub add_success: bool, @@ -265,14 +266,13 @@ impl MockPeerController { } response_tx.send_downcasted(response.content) } - PeerControlRequest::RegisterPeer { response_tx, .. } => { - let key = typ.peer_key_unchecked(); - let peer = peers_guard.peers.entry(key).or_default(); - if response.success { - peer.register_success = true; - } + PeerControlRequest::AllocatePeerIpPair { response_tx, .. } => { response_tx.send_downcasted(response.content) } + PeerControlRequest::ReleaseIpPair { + response_tx, + ip_pair: _, + } => response_tx.send_downcasted(response.content), PeerControlRequest::RemovePeer { response_tx, .. } => { let key = typ.peer_key_unchecked(); if response.success { diff --git a/common/wireguard/src/peer_controller/mod.rs b/common/wireguard/src/peer_controller/mod.rs index 60d7c98ee2..e8ba3c4078 100644 --- a/common/wireguard/src/peer_controller/mod.rs +++ b/common/wireguard/src/peer_controller/mod.rs @@ -1,6 +1,7 @@ // Copyright 2024 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::ip_pool::allocated_ip_pair; use crate::{ IpPool, error::{Error, Result}, @@ -32,9 +33,9 @@ use std::{ }; use tokio::sync::{RwLock, mpsc}; use tokio_stream::{StreamExt, wrappers::IntervalStream}; -use tracing::{debug, error, info, trace, warn}; +use tracing::{debug, error, info, trace}; -pub use nym_ip_packet_requests::IpPair; +pub use crate::ip_pool::IpPair; #[cfg(feature = "mock")] pub mod mock; @@ -75,10 +76,14 @@ pub enum PeerControlRequest { peer: Peer, response_tx: oneshot::Sender, }, - /// Register a new peer and allocate IPs from the pool - RegisterPeer { - registration_data: PeerRegistrationData, - response_tx: oneshot::Sender, + /// Attempt to allocate an IP pair from the pool + AllocatePeerIpPair { + response_tx: oneshot::Sender, + }, + /// Attempt to return an IP pair back to the pool + ReleaseIpPair { + response_tx: oneshot::Sender, + ip_pair: IpPair, }, RemovePeer { key: Key, @@ -109,7 +114,8 @@ pub enum PeerControlRequest { } pub type AddPeerControlResponse = Result<()>; -pub type RegisterPeerControlResponse = Result; +pub type AllocatePeerControlResponse = Result; +pub type ReleaseIpPairControlResponse = Result<()>; pub type RemovePeerControlResponse = Result<()>; pub type QueryPeerControlResponse = Result>; pub type GetClientBandwidthControlResponse = Result; @@ -207,9 +213,11 @@ impl PeerController { .await?; self.bw_storage_managers.remove(key); - warn!("MISSING CALL TO IP POOL RELEASE"); - // need to figure out what addresses to release - // self.ip_pool.release() + if let Ok(Some(peer)) = self.handle_query_peer_by_key(key).await + && let Some(ip_pair) = allocated_ip_pair(&peer) + { + self.ip_pool.release(ip_pair).await + } let ret = self.wg_api.remove_peer(key); if ret.is_err() { @@ -294,10 +302,7 @@ impl PeerController { /// /// This only allocates IPs - the caller must handle database storage and /// then call AddPeer with a complete Peer struct. - async fn handle_register_request( - &mut self, - _registration_data: PeerRegistrationData, - ) -> Result { + async fn handle_ip_allocation_request(&mut self) -> Result { nym_metrics::inc!("wg_ip_allocation_attempts"); // Allocate IP pair from pool @@ -313,6 +318,11 @@ impl PeerController { Ok(ip_pair) } + /// Return IP pair back to the pool + async fn handle_ip_release(&mut self, ip_pair: IpPair) { + self.ip_pool.release(ip_pair).await + } + async fn ip_to_key(&self, ip: IpAddr) -> Result> { Ok(self .bw_storage_managers @@ -477,6 +487,62 @@ impl PeerController { ); } + async fn handle_peer_control_request(&mut self, msg: PeerControlRequest) { + match msg { + PeerControlRequest::AddPeer { peer, response_tx } => { + response_tx.send(self.handle_add_request(&peer).await).ok(); + } + PeerControlRequest::AllocatePeerIpPair { response_tx } => { + response_tx + .send(self.handle_ip_allocation_request().await) + .ok(); + } + PeerControlRequest::ReleaseIpPair { + response_tx, + ip_pair, + } => { + self.handle_ip_release(ip_pair).await; + response_tx.send(Ok(())).ok(); + } + PeerControlRequest::RemovePeer { key, response_tx } => { + response_tx.send(self.remove_peer(&key).await).ok(); + } + PeerControlRequest::QueryPeer { key, response_tx } => { + response_tx + .send(self.handle_query_peer_by_key(&key).await) + .ok(); + } + PeerControlRequest::GetClientBandwidthByKey { key, response_tx } => { + response_tx + .send(self.handle_get_client_bandwidth_by_key(&key).await) + .ok(); + } + PeerControlRequest::GetClientBandwidthByIp { ip, response_tx } => { + response_tx + .send(self.handle_get_client_bandwidth_by_ip(ip).await) + .ok(); + } + PeerControlRequest::GetVerifierByKey { + key, + credential, + response_tx, + } => { + response_tx + .send(self.handle_query_verifier_by_key(&key, *credential).await) + .ok(); + } + PeerControlRequest::GetVerifierByIp { + ip, + credential, + response_tx, + } => { + response_tx + .send(self.handle_query_verifier_by_ip(ip, *credential).await) + .ok(); + } + } + } + pub async fn run(&mut self) { info!("started wireguard peer controller"); loop { @@ -504,30 +570,7 @@ impl PeerController { } msg = self.request_rx.recv() => { match msg { - Some(PeerControlRequest::AddPeer { peer, response_tx }) => { - response_tx.send(self.handle_add_request(&peer).await).ok(); - } - Some(PeerControlRequest::RegisterPeer { registration_data, response_tx }) => { - response_tx.send(self.handle_register_request(registration_data).await).ok(); - } - Some(PeerControlRequest::RemovePeer { key, response_tx }) => { - response_tx.send(self.remove_peer(&key).await).ok(); - } - Some(PeerControlRequest::QueryPeer { key, response_tx }) => { - response_tx.send(self.handle_query_peer_by_key(&key).await).ok(); - } - Some(PeerControlRequest::GetClientBandwidthByKey { key, response_tx }) => { - response_tx.send(self.handle_get_client_bandwidth_by_key(&key).await).ok(); - } - Some(PeerControlRequest::GetClientBandwidthByIp { ip, response_tx }) => { - response_tx.send(self.handle_get_client_bandwidth_by_ip(ip).await).ok(); - } - Some(PeerControlRequest::GetVerifierByKey { key, credential, response_tx }) => { - response_tx.send(self.handle_query_verifier_by_key(&key, *credential).await).ok(); - } - Some(PeerControlRequest::GetVerifierByIp { ip, credential, response_tx }) => { - response_tx.send(self.handle_query_verifier_by_ip(ip, *credential).await).ok(); - } + Some(request) => self.handle_peer_control_request(request).await, None => { trace!("PeerController [main loop]: stopping since channel closed"); break; diff --git a/gateway/src/error.rs b/gateway/src/error.rs index c620e019d8..1b9a39b6ae 100644 --- a/gateway/src/error.rs +++ b/gateway/src/error.rs @@ -3,6 +3,7 @@ pub use crate::node::client_handling::websocket::connection_handler::authenticated::RequestHandlingError; use crate::node::internal_service_providers::authenticator::error::AuthenticatorError; +use crate::node::wireguard::GatewayWireguardError; use nym_credential_verification::upgrade_mode::UpgradeModeEnableError; use nym_gateway_stats_storage::error::StatsStorageError; use nym_gateway_storage::error::GatewayStorageError; @@ -158,6 +159,9 @@ pub enum GatewayError { #[error(transparent)] UpgradeModeEnable(#[from] UpgradeModeEnableError), + + #[error(transparent)] + WireguardFailure(#[from] GatewayWireguardError), } impl From for GatewayError { diff --git a/gateway/src/node/internal_service_providers/authenticator/error.rs b/gateway/src/node/internal_service_providers/authenticator/error.rs index 5bdde15919..15a7edf1e0 100644 --- a/gateway/src/node/internal_service_providers/authenticator/error.rs +++ b/gateway/src/node/internal_service_providers/authenticator/error.rs @@ -1,6 +1,7 @@ // Copyright 2024 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::node::wireguard::GatewayWireguardError; use ipnetwork::IpNetworkError; use nym_client_core::error::ClientCoreError; use nym_credential_verification::upgrade_mode::UpgradeModeEnableError; @@ -82,9 +83,6 @@ pub enum AuthenticatorError { #[error("internal data corruption: {0}")] InternalDataCorruption(String), - #[error("peers can't be interacted with anymore")] - PeerInteractionStopped, - #[error("unknown version number")] UnknownVersion, @@ -108,6 +106,9 @@ pub enum AuthenticatorError { #[error(transparent)] UpgradeModeEnable(#[from] UpgradeModeEnableError), + + #[error(transparent)] + WireguardFailure(#[from] GatewayWireguardError), } impl AuthenticatorError { diff --git a/gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs b/gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs index 7b13c5cc42..3607ff7926 100644 --- a/gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs +++ b/gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs @@ -2,9 +2,9 @@ // SPDX-License-Identifier: Apache-2.0 use crate::node::internal_service_providers::authenticator::{ - config::Config, error::AuthenticatorError, peer_manager::PeerManager, - seen_credential_cache::SeenCredentialCache, + config::Config, error::AuthenticatorError, seen_credential_cache::SeenCredentialCache, }; +use crate::node::wireguard::PeerManager; use defguard_wireguard_rs::net::IpAddrMask; use defguard_wireguard_rs::{host::Peer, key::Key}; use futures::StreamExt; @@ -35,7 +35,6 @@ use nym_sphinx::receiver::ReconstructedMessage; use nym_task::ShutdownToken; use nym_wireguard::WireguardGatewayData; use nym_wireguard_types::PeerPublicKey; -use rand::{prelude::IteratorRandom, thread_rng}; use std::cmp::max; use std::{ net::IpAddr, @@ -54,14 +53,14 @@ const DEFAULT_WG_CLIENT_BANDWIDTH_THRESHOLD: i64 = 1024 * 1024 * 1024; pub(crate) struct RegisteredAndFree { registration_in_progress: PendingRegistrations, - free_private_network_ips: PrivateIPs, + taken_private_network_ips: PrivateIPs, } impl RegisteredAndFree { - pub(crate) fn new(free_private_network_ips: PrivateIPs) -> Self { + pub(crate) fn new() -> Self { RegisteredAndFree { registration_in_progress: Default::default(), - free_private_network_ips, + taken_private_network_ips: Default::default(), } } } @@ -90,7 +89,6 @@ pub(crate) struct MixnetListener { impl MixnetListener { pub fn new( config: Config, - free_private_network_ips: PrivateIPs, wireguard_gateway_data: WireguardGatewayData, mixnet_client: nym_sdk::mixnet::MixnetClient, upgrade_mode: UpgradeModeDetails, @@ -101,7 +99,7 @@ impl MixnetListener { MixnetListener { config, mixnet_client, - registered_and_free: RwLock::new(RegisteredAndFree::new(free_private_network_ips)), + registered_and_free: RwLock::new(RegisteredAndFree::new()), peer_manager: PeerManager::new(wireguard_gateway_data), upgrade_mode, ecash_verifier, @@ -139,34 +137,30 @@ impl MixnetListener { .cloned() .collect(); for reg in registered_values { - let ip = registered_and_free - .free_private_network_ips + let timestamp = registered_and_free + .taken_private_network_ips .get_mut(®.gateway_data.private_ips) .ok_or(AuthenticatorError::InternalDataCorruption(format!( "IPs {} should be present", reg.gateway_data.private_ips )))?; - let Some(timestamp) = ip else { - registered_and_free - .registration_in_progress - .remove(®.gateway_data.pub_key()); - tracing::debug!( - "Removed stale registration of {}", - reg.gateway_data.pub_key() - ); - continue; - }; let duration = SystemTime::now().duration_since(*timestamp).map_err(|_| { AuthenticatorError::InternalDataCorruption( "set timestamp shouldn't have been set in the future".to_string(), ) })?; if duration > DEFAULT_REGISTRATION_TIMEOUT_CHECK { - *ip = None; registered_and_free .registration_in_progress .remove(®.gateway_data.pub_key()); + registered_and_free + .taken_private_network_ips + .remove(®.gateway_data.private_ips); + self.peer_manager + .release_ip_pair(reg.gateway_data.private_ips.into()) + .await?; + tracing::debug!( "Removed stale registration of {}", reg.gateway_data.pub_key() @@ -383,19 +377,18 @@ impl MixnetListener { return Ok((bytes, reply_to)); } - let private_ip_ref = registered_and_free - .free_private_network_ips - .iter_mut() - .filter(|r| r.1.is_none()) - .choose(&mut thread_rng()) - .ok_or(AuthenticatorError::NoFreeIp)?; - let private_ips = *private_ip_ref.0; // mark it as used, even though it's not final - *private_ip_ref.1 = Some(SystemTime::now()); + let ip_allocation = self.peer_manager.allocate_peer_ip_pair().await?; + self.registered_and_free + .write() + .await + .taken_private_network_ips + .insert(ip_allocation.into(), SystemTime::now()); + let gateway_data = GatewayClient::new( self.keypair().private_key(), remote_public.inner(), - *private_ip_ref.0, + ip_allocation.into(), nonce, ); let registration_data = latest::registration::RegistrationData { @@ -414,7 +407,7 @@ impl MixnetListener { gateway_data: v1::registration::GatewayClient::new( self.keypair().private_key(), remote_public.inner(), - private_ips.ipv4.into(), + ip_allocation.ipv4.into(), nonce, ), wg_port: registration_data.wg_port, @@ -432,7 +425,7 @@ impl MixnetListener { gateway_data: v2::registration::GatewayClient::new( self.keypair().private_key(), remote_public.inner(), - private_ips.ipv4.into(), + ip_allocation.ipv4.into(), nonce, ), wg_port: registration_data.wg_port, @@ -450,7 +443,7 @@ impl MixnetListener { gateway_data: v3::registration::GatewayClient::new( self.keypair().private_key(), remote_public.inner(), - private_ips.ipv4.into(), + ip_allocation.ipv4.into(), nonce, ), wg_port: registration_data.wg_port, @@ -592,7 +585,7 @@ impl MixnetListener { .storage() .remove_wireguard_peer(&public_key) .await?; - return Err(e); + return Err(e.into()); } registered_and_free diff --git a/gateway/src/node/internal_service_providers/authenticator/mod.rs b/gateway/src/node/internal_service_providers/authenticator/mod.rs index cee2c9041b..95423a66e7 100644 --- a/gateway/src/node/internal_service_providers/authenticator/mod.rs +++ b/gateway/src/node/internal_service_providers/authenticator/mod.rs @@ -3,21 +3,19 @@ use crate::node::internal_service_providers::authenticator::error::AuthenticatorError; use futures::channel::oneshot; -use ipnetwork::IpNetwork; use nym_client_core::{HardcodedTopologyProvider, TopologyProvider}; +use nym_credential_verification::upgrade_mode::UpgradeModeDetails; use nym_sdk::{mixnet::Recipient, GatewayTransceiver}; use nym_task::ShutdownTracker; use nym_wireguard::WireguardGatewayData; -use std::{net::IpAddr, path::Path, sync::Arc, time::SystemTime}; +use std::{path::Path, sync::Arc}; pub use config::Config; -use nym_credential_verification::upgrade_mode::UpgradeModeDetails; pub mod config; pub mod error; pub mod mixnet_client; pub mod mixnet_listener; -mod peer_manager; mod seen_credential_cache; pub struct OnStartData { @@ -40,7 +38,6 @@ pub struct Authenticator { custom_gateway_transceiver: Option>, wireguard_gateway_data: WireguardGatewayData, ecash_verifier: Arc, - used_private_network_ips: Vec, shutdown: ShutdownTracker, on_start: Option>, } @@ -50,7 +47,6 @@ impl Authenticator { config: Config, upgrade_mode_state: UpgradeModeDetails, wireguard_gateway_data: WireguardGatewayData, - used_private_network_ips: Vec, ecash_verifier: Arc< dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync, >, @@ -64,7 +60,6 @@ impl Authenticator { custom_gateway_transceiver: None, ecash_verifier, wireguard_gateway_data, - used_private_network_ips, shutdown, on_start: None, } @@ -135,26 +130,8 @@ impl Authenticator { let self_address = *mixnet_client.nym_address(); - let used_private_network_ips = - std::collections::BTreeSet::from_iter(self.used_private_network_ips.iter()); - let private_ip_network = IpNetwork::new( - self.config.authenticator.private_ipv4.into(), - self.config.authenticator.private_network_prefix_v4, - )?; - let now = SystemTime::now(); - let free_private_network_ips = private_ip_network - .iter() - .map(|ip: IpAddr| { - if used_private_network_ips.contains(&ip) { - (ip.into(), Some(now)) - } else { - (ip.into(), None) - } - }) - .collect(); let mixnet_listener = mixnet_listener::MixnetListener::new( self.config, - free_private_network_ips, self.wireguard_gateway_data, mixnet_client, self.upgrade_mode_state, diff --git a/gateway/src/node/lp_listener/handler.rs b/gateway/src/node/lp_listener/handler.rs index 35e94d4b66..9f348bb422 100644 --- a/gateway/src/node/lp_listener/handler.rs +++ b/gateway/src/node/lp_listener/handler.rs @@ -1228,8 +1228,8 @@ where #[cfg(test)] mod tests { use super::*; - use crate::node::lp_listener::peer_manager::PeerManager; use crate::node::lp_listener::{LpConfig, LpDebug}; + use crate::node::wireguard::PeerManager; use crate::node::ActiveClientsStore; use bytes::BytesMut; use nym_credential_verification::upgrade_mode::{ diff --git a/gateway/src/node/lp_listener/mod.rs b/gateway/src/node/lp_listener/mod.rs index d20b83205d..0114e28d95 100644 --- a/gateway/src/node/lp_listener/mod.rs +++ b/gateway/src/node/lp_listener/mod.rs @@ -68,7 +68,6 @@ // They can be exported via Prometheus format using the metrics endpoint. use crate::error::GatewayError; -use crate::node::lp_listener::peer_manager::PeerManager; use crate::node::lp_listener::registration::RegistrationsInProgress; use crate::node::ActiveClientsStore; use dashmap::DashMap; @@ -86,6 +85,7 @@ use tokio::net::TcpListener; use tokio::sync::Semaphore; use tracing::*; +use crate::node::wireguard::PeerManager; pub use nym_lp::peer::LpLocalPeer; pub use nym_mixnet_client::forwarder::{ mix_forwarding_channels, MixForwardingReceiver, MixForwardingSender, @@ -94,7 +94,6 @@ pub use nym_wireguard::{PeerControlRequest, WireguardGatewayData}; mod data_handler; pub mod handler; -pub mod peer_manager; mod registration; pub type ReceiverIndex = u32; @@ -578,6 +577,7 @@ impl LpListener { /// /// The task automatically stops when the shutdown signal is received. fn spawn_state_cleanup_task(&self) -> tokio::task::JoinHandle<()> { + let peer_manager = Arc::clone(&self.handler_state.peer_manager); let handshake_states = Arc::clone(&self.handler_state.handshake_states); let session_states = Arc::clone(&self.handler_state.session_states); let pending_registrations = self.handler_state.registrations_in_progress.clone(); @@ -598,6 +598,7 @@ impl LpListener { self.shutdown.try_spawn_named( cleanup_task::cleanup_loop( + peer_manager, handshake_states, session_states, pending_registrations, @@ -620,15 +621,17 @@ impl LpListener { pub(crate) mod cleanup_task { use crate::node::lp_listener::registration::RegistrationsInProgress; use crate::node::lp_listener::{LpDebug, TimestampedState}; + use crate::node::wireguard::PeerManager; use dashmap::DashMap; use nym_lp::state_machine::LpStateBare; use nym_lp::LpStateMachine; use nym_metrics::inc_by; use nym_node_metrics::NymNodeMetrics; use std::sync::Arc; - use tracing::{debug, info}; + use tracing::{debug, error, info}; async fn perform_cleanup( + peer_manager: &PeerManager, handshake_states: &Arc>>, session_states: &Arc>>, registrations_in_progress: &RegistrationsInProgress, @@ -678,17 +681,27 @@ pub(crate) mod cleanup_task { }); // Remove stale registrations (based on time since last activity) - registrations_in_progress - .lock() - .await - .retain(|_, timestamped| { - if timestamped.age() > pending_registration_ttl { - pending_reg_removed += 1; - false - } else { - true - } - }); + let mut reg_guard = registrations_in_progress.lock().await; + let mut stale_registrations = Vec::new(); + for (k, timestamped) in reg_guard.iter() { + if timestamped.age() > pending_registration_ttl { + stale_registrations.push(*k) + } + } + + for to_remove in stale_registrations { + pending_reg_removed += 1; + + // SAFETY: we never dropped the guard and the entry existed + #[allow(clippy::unwrap_used)] + let entry = reg_guard.remove(&to_remove).unwrap(); + if let Err(err) = peer_manager + .release_ip_pair(entry.state.allocated_ip_pair()) + .await + { + error!("failed to release allocated ip pair: {err}") + } + } if hs_removed > 0 || ss_removed > 0 || demoted_removed > 0 || pending_reg_removed > 0 { let duration = start.elapsed(); @@ -724,6 +737,7 @@ pub(crate) mod cleanup_task { /// Demoted sessions (ReadOnlyTransport) use shorter TTL since they /// only need to drain in-flight packets after subsession promotion. pub(crate) async fn cleanup_loop( + peer_manager: Arc, handshake_states: Arc>>, session_states: Arc>>, registrations_in_progress: RegistrationsInProgress, @@ -743,7 +757,7 @@ pub(crate) mod cleanup_task { break; } _ = cleanup_interval.tick() => { - perform_cleanup(&handshake_states, &session_states, ®istrations_in_progress, cfg).await; + perform_cleanup(&peer_manager, &handshake_states, &session_states, ®istrations_in_progress, cfg).await; } } } diff --git a/gateway/src/node/lp_listener/peer_manager.rs b/gateway/src/node/lp_listener/peer_manager.rs deleted file mode 100644 index 027fe9221f..0000000000 --- a/gateway/src/node/lp_listener/peer_manager.rs +++ /dev/null @@ -1,127 +0,0 @@ -// Copyright 2026 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::GatewayError; -use defguard_wireguard_rs::host::Peer; -use defguard_wireguard_rs::key::Key; -use futures::channel::oneshot; -use nym_credential_verification::ClientBandwidth; -use nym_wireguard::peer_controller::IpPair; -use nym_wireguard::{PeerControlRequest, PeerRegistrationData, WireguardGatewayData}; -use nym_wireguard_types::PeerPublicKey; -use tracing::error; - -/// attempts to replicate [`crate::node::internal_service_providers::authenticator::peer_manager::PeerManager`] -// TODO: put those in the shared crate -pub struct PeerManager { - pub(crate) wireguard_gateway_data: WireguardGatewayData, -} - -impl PeerManager { - pub fn new(wireguard_gateway_data: WireguardGatewayData) -> Self { - PeerManager { - wireguard_gateway_data, - } - } - - pub async fn register_peer( - &self, - registration_data: PeerRegistrationData, - ) -> Result { - let (response_tx, response_rx) = oneshot::channel(); - let msg = PeerControlRequest::RegisterPeer { - registration_data, - response_tx, - }; - self.wireguard_gateway_data - .peer_tx() - .send(msg) - .await - .map_err(|e| { - GatewayError::InternalError(format!("Failed to send IP allocation request: {e}")) - })?; - - response_rx - .await - .map_err(|e| { - GatewayError::InternalError(format!("Failed to receive IP allocation: {e}")) - })? - .map_err(|e| { - error!("Failed to allocate IPs from pool: {e}"); - GatewayError::InternalError(format!("Failed to allocate IPs: {e}")) - }) - } - - pub async fn add_peer(&self, peer: Peer) -> Result<(), GatewayError> { - let (response_tx, response_rx) = oneshot::channel(); - let msg = PeerControlRequest::AddPeer { peer, response_tx }; - self.wireguard_gateway_data - .peer_tx() - .send(msg) - .await - .map_err(|e| { - GatewayError::InternalError(format!("Failed to send peer request: {e}")) - })?; - - response_rx - .await - .map_err(|_| GatewayError::InternalError("no response for add peer".to_string()))? - .map_err(|err| { - GatewayError::InternalError(format!("adding peer could not be performed: {err:?}")) - }) - } - - pub async fn query_peer( - &self, - public_key: PeerPublicKey, - ) -> Result, GatewayError> { - let key = Key::new(public_key.to_bytes()); - let (response_tx, response_rx) = oneshot::channel(); - let msg = PeerControlRequest::QueryPeer { key, response_tx }; - self.wireguard_gateway_data - .peer_tx() - .send(msg) - .await - .map_err(|_| { - GatewayError::InternalError("Failed to send peer query request".to_string()) - })?; - - response_rx - .await - .map_err(|_| GatewayError::InternalError("no response for query peer".to_string()))? - .map_err(|err| { - GatewayError::InternalError(format!( - "querying peer could not be performed: {err:?}" - )) - }) - } - - pub async fn query_client_bandwidth( - &self, - key: PeerPublicKey, - ) -> Result { - let key = Key::new(key.to_bytes()); - let (response_tx, response_rx) = oneshot::channel(); - let msg = PeerControlRequest::GetClientBandwidthByKey { key, response_tx }; - self.wireguard_gateway_data - .peer_tx() - .send(msg) - .await - .map_err(|_| { - GatewayError::InternalError( - "Failed to send peer bandwidth query request".to_string(), - ) - })?; - - response_rx - .await - .map_err(|_| { - GatewayError::InternalError("no response for query peer bandwidth".to_string()) - })? - .map_err(|err| { - GatewayError::InternalError(format!( - "querying client bandwidth could not be performed: {err:?}" - )) - }) - } -} diff --git a/gateway/src/node/lp_listener/registration.rs b/gateway/src/node/lp_listener/registration.rs index 5df0c609d3..577ab9a460 100644 --- a/gateway/src/node/lp_listener/registration.rs +++ b/gateway/src/node/lp_listener/registration.rs @@ -26,6 +26,7 @@ use nym_registration_common::{ LpRegistrationRequest, LpRegistrationRequestData, LpRegistrationResponse, RegistrationMode, RegistrationStatus, WireguardConfiguration, }; +use nym_wireguard::peer_controller::IpPair; use nym_wireguard::WireguardConfig; use nym_wireguard_types::PeerPublicKey; use std::collections::HashMap; @@ -71,6 +72,15 @@ pub struct PendingRegistrationState { wireguard_config: WireguardConfiguration, } +impl PendingRegistrationState { + pub(crate) fn allocated_ip_pair(&self) -> IpPair { + IpPair::new( + self.wireguard_config.private_ipv4, + self.wireguard_config.private_ipv6, + ) + } +} + #[derive(Clone, Default)] pub struct RegistrationsInProgress { /// Wrapped in TimestampedState for TTL-based cleanup of stale data. @@ -416,12 +426,8 @@ impl LpHandlerState { // Allocate IPs from centralized pool managed by PeerController let defguard_key = Key::new(peer_key.to_bytes()); - let registration_data = nym_wireguard::PeerRegistrationData::new(defguard_key.clone(), psk); - - let psk = registration_data.preshared_key.clone(); - // Request IP allocation from PeerController - let ip_pair = self.peer_manager.register_peer(registration_data).await?; + let ip_pair = self.peer_manager.allocate_peer_ip_pair().await?; let client_ipv4 = ip_pair.ipv4; let client_ipv6 = ip_pair.ipv6; diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 62263d521c..a8a613f2ec 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -18,6 +18,7 @@ use nym_credential_verification::upgrade_mode::{ }; use nym_crypto::asymmetric::{ed25519, x25519}; use nym_ip_packet_router::IpPacketRouter; +use nym_lp::peer::LpLocalPeer; use nym_mixnet_client::forwarder::MixForwardingSender; use nym_network_defaults::NymNetworkDetails; use nym_network_requester::NRServiceProviderBuilder; @@ -36,8 +37,8 @@ use tokio::sync::Semaphore; use tracing::*; use zeroize::Zeroizing; -use crate::node::lp_listener::peer_manager::PeerManager; pub use crate::node::upgrade_mode::watcher::UpgradeModeWatcher; +use crate::node::wireguard::PeerManager; pub use client_handling::active_clients::ActiveClientsStore; pub use lp_listener::LpConfig; pub use nym_credential_verification::upgrade_mode::UpgradeModeCheckRequestSender; @@ -47,7 +48,6 @@ pub use nym_gateway_storage::{ traits::{BandwidthGatewayStorage, InboxGatewayStorage}, GatewayStorage, }; -use nym_lp::peer::LpLocalPeer; pub use nym_sdk::{NymApiTopologyProvider, NymApiTopologyProviderConfig, UserAgent}; pub(crate) mod client_handling; @@ -55,6 +55,7 @@ pub(crate) mod internal_service_providers; pub mod lp_listener; mod stale_data_cleaner; pub mod upgrade_mode; +pub mod wireguard; #[derive(Debug, Clone)] pub struct LocalNetworkRequesterOpts { @@ -492,25 +493,12 @@ impl GatewayTasksBuilder { Ok(peers) } - async fn get_wireguard_networks(&mut self) -> Result, GatewayError> { - if let Some(cached) = self.wireguard_networks.take() { - return Ok(cached); - } - - let (peers, used_private_network_ips) = self.build_wireguard_peers_and_networks().await?; - // cache peers for the other task - - self.wireguard_peers = Some(peers); - Ok(used_private_network_ips) - } - pub async fn build_wireguard_authenticator( &mut self, upgrade_mode_common: UpgradeModeDetails, topology_provider: Box, ) -> Result, GatewayError> { let ecash_manager = self.ecash_manager().await?; - let used_private_network_ips = self.get_wireguard_networks().await?; let Some(opts) = &self.authenticator_opts else { return Err(GatewayError::UnspecifiedAuthenticatorConfig); @@ -533,7 +521,6 @@ impl GatewayTasksBuilder { opts.config.clone(), upgrade_mode_common, wireguard_data.inner.clone(), - used_private_network_ips, ecash_manager, self.shutdown_tracker.clone(), ) diff --git a/gateway/src/node/wireguard/error.rs b/gateway/src/node/wireguard/error.rs new file mode 100644 index 0000000000..de0a6aaa9f --- /dev/null +++ b/gateway/src/node/wireguard/error.rs @@ -0,0 +1,19 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum GatewayWireguardError { + #[error("internal error: {0}")] + InternalError(String), + + #[error("peers can't be interacted with anymore")] + PeerInteractionStopped, +} + +impl GatewayWireguardError { + pub fn internal(message: impl Into) -> Self { + GatewayWireguardError::InternalError(message.into()) + } +} diff --git a/gateway/src/node/wireguard/mod.rs b/gateway/src/node/wireguard/mod.rs new file mode 100644 index 0000000000..7c9a02b8e1 --- /dev/null +++ b/gateway/src/node/wireguard/mod.rs @@ -0,0 +1,8 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +pub mod error; +pub mod peer_manager; + +pub use error::GatewayWireguardError; +pub use peer_manager::PeerManager; diff --git a/gateway/src/node/internal_service_providers/authenticator/peer_manager.rs b/gateway/src/node/wireguard/peer_manager.rs similarity index 84% rename from gateway/src/node/internal_service_providers/authenticator/peer_manager.rs rename to gateway/src/node/wireguard/peer_manager.rs index c057dfa57c..0921b758ca 100644 --- a/gateway/src/node/internal_service_providers/authenticator/peer_manager.rs +++ b/gateway/src/node/wireguard/peer_manager.rs @@ -1,14 +1,17 @@ -// Copyright 2024 - Nym Technologies SA +// Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::node::internal_service_providers::authenticator::error::AuthenticatorError; +use crate::node::wireguard::GatewayWireguardError; use defguard_wireguard_rs::{host::Peer, key::Key}; use futures::channel::oneshot; use nym_credential_verification::{ClientBandwidth, TicketVerifier}; use nym_credentials_interface::CredentialSpendingData; +use nym_wireguard::peer_controller::IpPair; use nym_wireguard::{peer_controller::PeerControlRequest, WireguardGatewayData}; use nym_wireguard_types::PeerPublicKey; +use tracing::error; +#[derive(Clone)] pub struct PeerManager { pub(crate) wireguard_gateway_data: WireguardGatewayData, } @@ -19,26 +22,73 @@ impl PeerManager { wireguard_gateway_data, } } - pub async fn add_peer(&self, peer: Peer) -> Result<(), AuthenticatorError> { + + pub async fn allocate_peer_ip_pair(&self) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + let msg = PeerControlRequest::AllocatePeerIpPair { response_tx }; + self.wireguard_gateway_data + .peer_tx() + .send(msg) + .await + .map_err(|_| GatewayWireguardError::PeerInteractionStopped)?; + + response_rx + .await + .map_err(|e| { + GatewayWireguardError::InternalError(format!( + "Failed to receive IP allocation: {e}" + )) + })? + .map_err(|e| { + error!("Failed to allocate IPs from pool: {e}"); + GatewayWireguardError::InternalError(format!("Failed to allocate IPs: {e}")) + }) + } + + pub async fn release_ip_pair(&self, ip_pair: IpPair) -> Result<(), GatewayWireguardError> { + let (response_tx, response_rx) = oneshot::channel(); + let msg = PeerControlRequest::ReleaseIpPair { + response_tx, + ip_pair, + }; + self.wireguard_gateway_data + .peer_tx() + .send(msg) + .await + .map_err(|_| GatewayWireguardError::PeerInteractionStopped)?; + + response_rx + .await + .map_err(|_| GatewayWireguardError::internal("no response for release ip allocation"))? + .map_err(|err| { + GatewayWireguardError::InternalError(format!( + "releasing ip pair not be performed: {err:?}" + )) + })?; + + Ok(()) + } + + pub async fn add_peer(&self, peer: Peer) -> Result<(), GatewayWireguardError> { let (response_tx, response_rx) = oneshot::channel(); let msg = PeerControlRequest::AddPeer { peer, response_tx }; self.wireguard_gateway_data .peer_tx() .send(msg) .await - .map_err(|_| AuthenticatorError::PeerInteractionStopped)?; + .map_err(|_| GatewayWireguardError::PeerInteractionStopped)?; response_rx .await - .map_err(|_| AuthenticatorError::InternalError("no response for add peer".to_string()))? + .map_err(|_| GatewayWireguardError::internal("no response for add peer".to_string()))? .map_err(|err| { - AuthenticatorError::InternalError(format!( + GatewayWireguardError::internal(format!( "adding peer could not be performed: {err:?}" )) }) } - pub async fn _remove_peer(&self, pub_key: PeerPublicKey) -> Result<(), AuthenticatorError> { + pub async fn _remove_peer(&self, pub_key: PeerPublicKey) -> Result<(), GatewayWireguardError> { let key = Key::new(pub_key.to_bytes()); let (response_tx, response_rx) = oneshot::channel(); let msg = PeerControlRequest::RemovePeer { key, response_tx }; @@ -46,15 +96,13 @@ impl PeerManager { .peer_tx() .send(msg) .await - .map_err(|_| AuthenticatorError::PeerInteractionStopped)?; + .map_err(|_| GatewayWireguardError::PeerInteractionStopped)?; response_rx .await - .map_err(|_| { - AuthenticatorError::InternalError("no response for remove peer".to_string()) - })? + .map_err(|_| GatewayWireguardError::internal("no response for remove peer"))? .map_err(|err| { - AuthenticatorError::InternalError(format!( + GatewayWireguardError::InternalError(format!( "removing peer could not be performed: {err:?}" )) }) @@ -63,7 +111,7 @@ impl PeerManager { pub async fn query_peer( &self, public_key: PeerPublicKey, - ) -> Result, AuthenticatorError> { + ) -> Result, GatewayWireguardError> { let key = Key::new(public_key.to_bytes()); let (response_tx, response_rx) = oneshot::channel(); let msg = PeerControlRequest::QueryPeer { key, response_tx }; @@ -71,15 +119,13 @@ impl PeerManager { .peer_tx() .send(msg) .await - .map_err(|_| AuthenticatorError::PeerInteractionStopped)?; + .map_err(|_| GatewayWireguardError::PeerInteractionStopped)?; response_rx .await - .map_err(|_| { - AuthenticatorError::InternalError("no response for query peer".to_string()) - })? + .map_err(|_| GatewayWireguardError::internal("no response for query peer".to_string()))? .map_err(|err| { - AuthenticatorError::InternalError(format!( + GatewayWireguardError::internal(format!( "querying peer could not be performed: {err:?}" )) }) @@ -88,7 +134,7 @@ impl PeerManager { pub async fn query_bandwidth( &self, public_key: PeerPublicKey, - ) -> Result { + ) -> Result { let client_bandwidth = self.query_client_bandwidth(public_key).await?; Ok(client_bandwidth.available().await) } @@ -96,7 +142,7 @@ impl PeerManager { pub async fn query_client_bandwidth( &self, key: PeerPublicKey, - ) -> Result { + ) -> Result { let key = Key::new(key.to_bytes()); let (response_tx, response_rx) = oneshot::channel(); let msg = PeerControlRequest::GetClientBandwidthByKey { key, response_tx }; @@ -104,17 +150,13 @@ impl PeerManager { .peer_tx() .send(msg) .await - .map_err(|_| AuthenticatorError::PeerInteractionStopped)?; + .map_err(|_| GatewayWireguardError::PeerInteractionStopped)?; response_rx .await - .map_err(|_| { - AuthenticatorError::InternalError( - "no response for query client bandwidth".to_string(), - ) - })? + .map_err(|_| GatewayWireguardError::internal("no response for query client bandwidth"))? .map_err(|err| { - AuthenticatorError::InternalError(format!( + GatewayWireguardError::internal(format!( "querying client bandwidth could not be performed: {err:?}" )) }) @@ -124,7 +166,7 @@ impl PeerManager { &self, key: PeerPublicKey, credential: CredentialSpendingData, - ) -> Result, AuthenticatorError> { + ) -> Result, GatewayWireguardError> { let key = Key::new(key.to_bytes()); let (response_tx, response_rx) = oneshot::channel(); let msg = PeerControlRequest::GetVerifierByKey { @@ -136,15 +178,15 @@ impl PeerManager { .peer_tx() .send(msg) .await - .map_err(|_| AuthenticatorError::PeerInteractionStopped)?; + .map_err(|_| GatewayWireguardError::PeerInteractionStopped)?; response_rx .await .map_err(|_| { - AuthenticatorError::InternalError("no response for query verifier".to_string()) + GatewayWireguardError::internal("no response for query verifier".to_string()) })? .map_err(|err| { - AuthenticatorError::InternalError(format!( + GatewayWireguardError::internal(format!( "querying verifier could not be performed: {err:?}" )) }) diff --git a/integration-tests/src/lp_registration.rs b/integration-tests/src/lp_registration.rs index d292a2a441..2e171962c4 100644 --- a/integration-tests/src/lp_registration.rs +++ b/integration-tests/src/lp_registration.rs @@ -14,11 +14,11 @@ mod tests { use nym_crypto::asymmetric::{ed25519, x25519}; use nym_gateway::GatewayError; use nym_gateway::node::lp_listener::handler::LpConnectionHandler; - use nym_gateway::node::lp_listener::peer_manager::PeerManager; use nym_gateway::node::lp_listener::{ LpDebug, LpHandlerState, LpLocalPeer, MixForwardingReceiver, PeerControlRequest, WireguardGatewayData, mix_forwarding_channels, }; + use nym_gateway::node::wireguard::PeerManager; use nym_gateway::node::{ActiveClientsStore, GatewayStorage, LpConfig}; use nym_registration_client::{LpClientError, LpRegistrationClient}; use nym_test_utils::helpers::{CryptoRng, RngCore, u64_seeded_rng}; @@ -443,11 +443,10 @@ mod tests { // 1) peer registration - ip pair allocation let ip_pair = entry.allocate_ip_pair().await; let reg_res = Ok::<_, nym_wireguard::Error>(ip_pair); - let public_key = client_key.to_wg_key(); entry .register_peer_controller_response( - PeerControlRequestType::RegisterPeer { public_key }, + PeerControlRequestType::AllocatePeerIpPair {}, reg_res, ) .await; @@ -497,7 +496,6 @@ mod tests { let peers_guard = entry.mock_peer_controller_state.peers.read().await; let peer = peers_guard.get_by_x25519_key(&client_key).unwrap().clone(); drop(peers_guard); - assert!(peer.register_success); assert!(peer.add_success); assert_eq!(registration_result.private_ipv4, ip_pair.ipv4); @@ -616,11 +614,10 @@ mod tests { // 1) peer registration - ip pair allocation let entry_ip_pair = entry.allocate_ip_pair().await; let reg_res = Ok::<_, nym_wireguard::Error>(entry_ip_pair); - let public_key = client_key.to_wg_key(); entry .register_peer_controller_response( - PeerControlRequestType::RegisterPeer { public_key }, + PeerControlRequestType::AllocatePeerIpPair {}, reg_res, ) .await; @@ -668,10 +665,9 @@ mod tests { // 1) peer registration - ip pair allocation let exit_ip_pair = exit.allocate_ip_pair().await; let reg_res = Ok::<_, nym_wireguard::Error>(exit_ip_pair); - let public_key = client_key.to_wg_key(); exit.register_peer_controller_response( - PeerControlRequestType::RegisterPeer { public_key }, + PeerControlRequestType::AllocatePeerIpPair {}, reg_res, ) .await; @@ -738,13 +734,11 @@ mod tests { let peers_guard = entry.mock_peer_controller_state.peers.read().await; let entry_peer = peers_guard.get_by_x25519_key(&client_key).unwrap().clone(); drop(peers_guard); - assert!(entry_peer.register_success); assert!(entry_peer.add_success); let peers_guard = exit.mock_peer_controller_state.peers.read().await; let exit_peer = peers_guard.get_by_x25519_key(&client_key).unwrap().clone(); drop(peers_guard); - assert!(exit_peer.register_success); assert!(exit_peer.add_success); assert_eq!(entry_registration_result.private_ipv4, entry_ip_pair.ipv4);