GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Lp/bugfix/share ip allocation (#6395)

Browse Source 
* feat: use shared PeerManager between Authenticator and LpHandlerState

* feat: share IpPool

* clippy and test fixes

* PR suggestions
This commit is contained in:
Jędrzej Stuczyński
2026-01-30 11:38:17 +00:00
committed by GitHub
parent 9d661e7a7b
commit dccdde108c
22 changed files with 429 additions and 374 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -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",
common/authenticator-requests/src/v6/registration.rs
+1 -2
View File
@@ -20,13 +20,12 @@ use nym_crypto::asymmetric::x25519::{PrivateKey, PublicKey};
use sha2::Sha256;
pub type PendingRegistrations = HashMap<PeerPublicKey, RegistrationData>;
pub type PrivateIPs = HashMap<IpPair, Taken>;
pub type PrivateIPs = HashMap<IpPair, SystemTime>;
#[cfg(feature = "verify")]
pub type HmacSha256 = Hmac<Sha256>;
pub type Nonce = u64;
pub type Taken = Option<SystemTime>;
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct IpPair {
common/wireguard-private-metadata/tests/src/v2/peer_controller.rs
+12 -2
View File
@@ -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();
}
}
}
common/wireguard/Cargo.toml
+1 -1
View File
@@ -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 }
common/wireguard/src/ip_pool/compat.rs
+60
View File
@@ -0,0 +1,60 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::ip_pool::IpPair;
impl From<IpPair> 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<IpPair> 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<IpPair> 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<nym_authenticator_requests::v6::registration::IpPair> for IpPair {
fn from(ip_pair: nym_authenticator_requests::v6::registration::IpPair) -> Self {
IpPair {
ipv4: ip_pair.ipv4,
ipv6: ip_pair.ipv6,
}
}
}
impl From<nym_authenticator_requests::v5::registration::IpPair> for IpPair {
fn from(ip_pair: nym_authenticator_requests::v5::registration::IpPair) -> Self {
IpPair {
ipv4: ip_pair.ipv4,
ipv6: ip_pair.ipv6,
}
}
}
impl From<nym_authenticator_requests::v4::registration::IpPair> for IpPair {
fn from(ip_pair: nym_authenticator_requests::v4::registration::IpPair) -> Self {
IpPair {
ipv4: ip_pair.ipv4,
ipv6: ip_pair.ipv6,
}
}
}
common/wireguard/src/ip_pool.rs → common/wireguard/src/ip_pool/mod.rs
+55 -13
View File
@@ -1,14 +1,56 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// 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<IpPair> {
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<IpPair, IpPoolError> {
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}");
}
}
common/wireguard/src/lib.rs
+17 -34
View File
@@ -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,
common/wireguard/src/peer_controller/mock.rs
+16 -16
View File
@@ -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<KeyWrapper> {
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: (&registration_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 {
common/wireguard/src/peer_controller/mod.rs
+81 -38
View File
@@ -1,6 +1,7 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// 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<AddPeerControlResponse>,
},
/// Register a new peer and allocate IPs from the pool
RegisterPeer {
registration_data: PeerRegistrationData,
response_tx: oneshot::Sender<RegisterPeerControlResponse>,
/// Attempt to allocate an IP pair from the pool
AllocatePeerIpPair {
response_tx: oneshot::Sender<AllocatePeerControlResponse>,
},
/// Attempt to return an IP pair back to the pool
ReleaseIpPair {
response_tx: oneshot::Sender<ReleaseIpPairControlResponse>,
ip_pair: IpPair,
},
RemovePeer {
key: Key,
@@ -109,7 +114,8 @@ pub enum PeerControlRequest {
}
pub type AddPeerControlResponse = Result<()>;
pub type RegisterPeerControlResponse = Result<IpPair>;
pub type AllocatePeerControlResponse = Result<IpPair>;
pub type ReleaseIpPairControlResponse = Result<()>;
pub type RemovePeerControlResponse = Result<()>;
pub type QueryPeerControlResponse = Result<Option<Peer>>;
pub type GetClientBandwidthControlResponse = Result<ClientBandwidth>;
@@ -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<IpPair> {
async fn handle_ip_allocation_request(&mut self) -> Result<IpPair> {
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<Option<Key>> {
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;
gateway/src/error.rs
+4
View File
@@ -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<ClientCoreError> for GatewayError {
gateway/src/node/internal_service_providers/authenticator/error.rs
+4 -3
View File
@@ -1,6 +1,7 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// 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 {
gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs
+27 -34
View File
@@ -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(&reg.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(&reg.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(&reg.gateway_data.pub_key());
registered_and_free
.taken_private_network_ips
.remove(&reg.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
gateway/src/node/internal_service_providers/authenticator/mod.rs
+2 -25
View File
@@ -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<Box<dyn GatewayTransceiver + Send + Sync>>,
wireguard_gateway_data: WireguardGatewayData,
ecash_verifier: Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
used_private_network_ips: Vec<IpAddr>,
shutdown: ShutdownTracker,
on_start: Option<oneshot::Sender<OnStartData>>,
}
@@ -50,7 +47,6 @@ impl Authenticator {
config: Config,
upgrade_mode_state: UpgradeModeDetails,
wireguard_gateway_data: WireguardGatewayData,
used_private_network_ips: Vec<IpAddr>,
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,
gateway/src/node/lp_listener/handler.rs
+1 -1
View File
@@ -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::{
gateway/src/node/lp_listener/mod.rs
+29 -15
View File
@@ -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<DashMap<u32, TimestampedState<LpStateMachine>>>,
session_states: &Arc<DashMap<u32, TimestampedState<LpStateMachine>>>,
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<PeerManager>,
handshake_states: Arc<DashMap<u32, TimestampedState<LpStateMachine>>>,
session_states: Arc<DashMap<u32, TimestampedState<LpStateMachine>>>,
registrations_in_progress: RegistrationsInProgress,
@@ -743,7 +757,7 @@ pub(crate) mod cleanup_task {
break;
}
_ = cleanup_interval.tick() => {
perform_cleanup(&handshake_states, &session_states, &registrations_in_progress, cfg).await;
perform_cleanup(&peer_manager, &handshake_states, &session_states, &registrations_in_progress, cfg).await;
}
}
}
gateway/src/node/lp_listener/peer_manager.rs
-127
View File
@@ -1,127 +0,0 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// 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<IpPair, GatewayError> {
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<Option<Peer>, 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<ClientBandwidth, GatewayError> {
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:?}"
))
})
}
}
gateway/src/node/lp_listener/registration.rs
+11 -5
View File
@@ -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;
gateway/src/node/mod.rs
+3 -16
View File
@@ -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<Vec<IpAddr>, 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<dyn TopologyProvider + Send + Sync>,
) -> Result<ServiceProviderBeingBuilt<Authenticator>, 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(),
)
gateway/src/node/wireguard/error.rs
+19
View File
@@ -0,0 +1,19 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// 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<String>) -> Self {
GatewayWireguardError::InternalError(message.into())
}
}
gateway/src/node/wireguard/mod.rs
+8
View File
@@ -0,0 +1,8 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod error;
pub mod peer_manager;
pub use error::GatewayWireguardError;
pub use peer_manager::PeerManager;
gateway/src/node/internal_service_providers/authenticator/peer_manager.rs → gateway/src/node/wireguard/peer_manager.rs
+73 -31
View File
@@ -1,14 +1,17 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// 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<IpPair, GatewayWireguardError> {
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<Option<Peer>, AuthenticatorError> {
) -> Result<Option<Peer>, 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<i64, AuthenticatorError> {
) -> Result<i64, GatewayWireguardError> {
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<ClientBandwidth, AuthenticatorError> {
) -> Result<ClientBandwidth, GatewayWireguardError> {
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<Box<dyn TicketVerifier + Send + Sync>, AuthenticatorError> {
) -> Result<Box<dyn TicketVerifier + Send + Sync>, 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:?}"
))
})
integration-tests/src/lp_registration.rs
+4 -10
View File
@@ -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);