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

Integration test for nested LP registration

Browse Source 
- move `LpTransport` trait definition to shared `nym-lp-transport` crate
- make transport layer within `LpConnectionHandler` generic with respect to the forwarding target. it must, however, use the same type as the incoming client connection
- extracted explicit `LpConnectionHandler::establish_exit_stream` to more easily modify it in the future to fully protect the channel and disallow using untrusted egress points
- fix additional log-string interpolation nits
This commit is contained in:
Jędrzej Stuczyński
2026-01-08 11:06:52 +00:00
parent 9782ca1452
commit 5fa1a3fc17
19 changed files with 521 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+12 -1
View File
@@ -4016,12 +4016,14 @@ dependencies = [
"nym-credentials-interface",
"nym-crypto",
"nym-gateway",
"nym-lp-transport",
"nym-registration-client",
"nym-test-utils",
"nym-wireguard",
"sqlx",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
@@ -6240,6 +6242,7 @@ dependencies = [
"nym-ip-packet-router",
"nym-kcp",
"nym-lp",
"nym-lp-transport",
"nym-metrics",
"nym-mixnet-client",
"nym-network-defaults",
@@ -6746,6 +6749,14 @@ dependencies = [
name = "nym-lp-common"
version = "0.1.0"
[[package]]
name = "nym-lp-transport"
version = "0.1.0"
dependencies = [
"nym-test-utils",
"tokio",
]
[[package]]
name = "nym-metrics"
version = "0.1.0"
@@ -7331,9 +7342,9 @@ dependencies = [
"nym-crypto",
"nym-ip-packet-client",
"nym-lp",
"nym-lp-transport",
"nym-registration-common",
"nym-sdk",
"nym-test-utils",
"nym-validator-client",
"nym-wireguard-types",
"rand 0.8.5",
Cargo.toml
+1 -1
View File
@@ -174,7 +174,7 @@ members = [
"wasm/node-tester",
"wasm/zknym-lib",
"nym-gateway-probe",
"integration-tests",
"integration-tests", "common/nym-lp-transport",
]
default-members = [
common/credentials-interface/src/lib.rs
+16 -1
View File
@@ -3,6 +3,7 @@
use rand::Rng;
use serde::{Deserialize, Serialize};
use std::fmt::Debug;
use thiserror::Error;
use time::{Date, OffsetDateTime};
@@ -73,7 +74,7 @@ pub struct CredentialSigningData {
pub ticketbook_type: TicketType,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Clone)]
#[derive(Serialize, Deserialize, PartialEq, Clone)]
pub struct CredentialSpendingData {
pub payment: Payment,
@@ -86,6 +87,20 @@ pub struct CredentialSpendingData {
pub epoch_id: u64,
}
impl Debug for CredentialSpendingData {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// we're redacting the payment not since it contains secret,
// but because it's producing a lot of noise in the output and
// we are not really interested in coordinates of each of the attached curve points
f.debug_struct("CredentialSpendingData")
.field("payment", &"[REDACTED]")
.field("pay_info", &self.pay_info)
.field("spend_date", &self.spend_date)
.field("epoch_id", &self.epoch_id)
.finish()
}
}
impl CredentialSpendingData {
pub fn verify(&self, verification_key: &VerificationKeyAuth) -> Result<(), CompactEcashError> {
self.payment.spend_verify(
common/nym-lp-common/src/lib.rs
+3
View File
@@ -1,3 +1,6 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use std::fmt;
use std::fmt::Write;
common/nym-lp-transport/Cargo.toml
+21
View File
@@ -0,0 +1,21 @@
[package]
name = "nym-lp-transport"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license.workspace = true
rust-version.workspace = true
readme.workspace = true
[dependencies]
tokio = { workspace = true, features = ["net"] }
nym-test-utils = { path = "../test-utils", optional = true }
[features]
io-mocks = ["nym-test-utils"]
[lints]
workspace = true
common/nym-lp-transport/src/lib.rs
+4
View File
@@ -0,0 +1,4 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod traits;
nym-registration-client/src/lp_client/traits.rs → common/nym-lp-transport/src/traits.rs
+9 -11
View File
@@ -1,40 +1,38 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::LpConfig;
#[cfg(feature = "io-mocks")]
use nym_test_utils::mocks::async_read_write::MockIOStream;
use std::net::SocketAddr;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::net::TcpStream;
#[cfg(feature = "io-mocks")]
use nym_test_utils::mocks::async_read_write::MockIOStream;
// only used in internal code (and tests)
#[allow(async_fn_in_trait)]
pub trait LpTransportLayer: AsyncRead + AsyncWrite + Sized {
pub trait LpTransport: AsyncRead + AsyncWrite + Sized {
async fn connect(endpoint: SocketAddr) -> std::io::Result<Self>;
fn configure(&mut self, config: &LpConfig) -> std::io::Result<()>;
fn set_no_delay(&mut self, nodelay: bool) -> std::io::Result<()>;
}
impl LpTransportLayer for TcpStream {
impl LpTransport for TcpStream {
async fn connect(endpoint: SocketAddr) -> std::io::Result<Self> {
TcpStream::connect(endpoint).await
}
fn configure(&mut self, config: &LpConfig) -> std::io::Result<()> {
fn set_no_delay(&mut self, nodelay: bool) -> std::io::Result<()> {
// Set TCP_NODELAY for low latency
self.set_nodelay(config.tcp_nodelay)
self.set_nodelay(nodelay)
}
}
#[cfg(feature = "io-mocks")]
impl LpTransportLayer for MockIOStream {
impl LpTransport for MockIOStream {
async fn connect(_endpoint: SocketAddr) -> std::io::Result<Self> {
Ok(MockIOStream::default())
}
fn configure(&mut self, _config: &LpConfig) -> std::io::Result<()> {
fn set_no_delay(&mut self, _nodelay: bool) -> std::io::Result<()> {
Ok(())
}
}
common/nymsphinx/framing/src/processing.rs
+2 -2
View File
@@ -14,7 +14,7 @@ use nym_sphinx_types::{
};
use std::fmt::Display;
use thiserror::Error;
use tracing::{debug, info, trace};
use tracing::{debug, trace};
#[derive(Debug)]
pub enum MixProcessingResultData {
@@ -375,7 +375,7 @@ fn split_into_ack_and_message(
match SurbAck::try_recover_first_hop_packet(&ack_data, packet_type) {
Ok((first_hop, packet)) => (first_hop, packet),
Err(err) => {
info!("Failed to recover first hop from ack data: {err}");
tracing::info!("Failed to recover first hop from ack data: {err}");
return Err(err.into());
}
};
common/test-utils/src/mocks/async_read_write.rs
+52 -6
View File
@@ -3,8 +3,11 @@
use crate::mocks::shared::InnerWrapper;
use futures::ready;
use std::fmt::{Display, Formatter};
use std::io;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicU8, Ordering};
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tracing::trace;
@@ -12,6 +15,21 @@ use tracing::trace;
const INIT_ID: &str = "initialiser";
const RECV_ID: &str = "recipient";
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum Side {
Initialiser,
Recipient,
}
impl Display for Side {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
Side::Initialiser => INIT_ID.fmt(f),
Side::Recipient => RECV_ID.fmt(f),
}
}
}
// sending buffer of the first stream is the receiving buffer of the second stream
// and vice versa
pub fn mock_io_streams() -> (MockIOStream, MockIOStream) {
@@ -23,7 +41,10 @@ pub fn mock_io_streams() -> (MockIOStream, MockIOStream) {
pub struct MockIOStream {
// identifier to use for logging purposes
id: &'static str,
id: Arc<AtomicU8>,
// side of the stream to use for logging purposes
side: Side,
// messages to send
tx: InnerWrapper<Vec<u8>>,
@@ -35,7 +56,8 @@ pub struct MockIOStream {
impl Default for MockIOStream {
fn default() -> Self {
MockIOStream {
id: INIT_ID,
id: Arc::new(AtomicU8::new(0)),
side: Side::Initialiser,
tx: Default::default(),
rx: Default::default(),
}
@@ -45,16 +67,21 @@ impl Default for MockIOStream {
impl MockIOStream {
#[allow(clippy::panic)]
fn make_connection(&self) -> Self {
if self.id != INIT_ID {
if self.side != Side::Initialiser {
panic!("attempted to make invalid connection")
}
MockIOStream {
id: RECV_ID,
id: self.id.clone(),
side: Side::Recipient,
tx: self.rx.cloned_buffer(),
rx: self.tx.cloned_buffer(),
}
}
pub fn set_id(&self, id: u8) {
self.id.store(id, Ordering::Relaxed)
}
// the prefix `try_` is due to the fact that if the mock is cloned at an invalid state,
// `assert!` will fail causing panic (which is fine in **test** code)
pub fn try_get_remote_handle(&self) -> Self {
@@ -72,6 +99,25 @@ impl MockIOStream {
pub fn unchecked_rx_data(&self) -> Vec<u8> {
self.rx.buffer.try_lock().unwrap().content.clone()
}
fn log_read(&self, bytes: usize) {
let id = self.id.load(Ordering::Relaxed);
if id == 0 {
trace!("[{}] read {bytes} bytes from mock stream", self.side)
} else {
trace!("[{}-{id}] read {bytes} bytes from mock stream", self.side)
}
}
fn log_write(&self, bytes: usize) {
let id = self.id.load(Ordering::Relaxed);
if id == 0 {
trace!("[{}] wrote {bytes} bytes to mock stream", self.side)
} else {
trace!("[{}-{id}] wrote {bytes} bytes to mock stream", self.side)
}
}
}
impl AsyncRead for MockIOStream {
@@ -98,7 +144,7 @@ impl AsyncRead for MockIOStream {
return Poll::Pending;
}
trace!("{}: read {} bytes from mock stream", self.id, data.len());
self.log_read(data.len());
// if let Some(waker) = guard.waker.take() {
// waker.wake();
// }
@@ -135,7 +181,7 @@ impl AsyncWrite for MockIOStream {
// return Poll::Pending;
// }
trace!("{}: wrote {len} bytes to mock stream", self.id);
self.log_write(buf.len());
Poll::Ready(Ok(len))
}
gateway/Cargo.toml
+1
View File
@@ -78,6 +78,7 @@ nym-service-provider-requests-common = { path = "../common/service-provider-requ
# LP dependencies
nym-lp = { path = "../common/nym-lp" }
nym-lp-transport = { path = "../common/nym-lp-transport" }
nym-kcp = { path = "../common/nym-kcp" }
nym-registration-common = { path = "../common/registration" }
bytes = { workspace = true }
gateway/src/error.rs
+1 -1
View File
@@ -124,7 +124,7 @@ pub enum GatewayError {
},
#[error("{0}")]
CredentialVefiricationError(#[from] nym_credential_verification::Error),
CredentialVerificationError(#[from] nym_credential_verification::Error),
#[error("LP connection error: {0}")]
LpConnectionError(String),
gateway/src/node/lp_listener/handler.rs
+71 -54
View File
@@ -10,10 +10,13 @@ use nym_lp::{
codec::OuterAeadKey, keypair::PublicKey, message::ForwardPacketData, packet::LpHeader,
LpMessage, LpPacket, OuterHeader,
};
use nym_lp_transport::traits::LpTransport;
use nym_metrics::{add_histogram_obs, inc};
use std::net::SocketAddr;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::time::timeout;
use tracing::*;
// Histogram buckets for LP operation duration (legacy - used by unused forwarding methods)
@@ -80,15 +83,16 @@ pub struct LpConnectionHandler<S = TcpStream> {
/// All subsequent packets on this connection must use this receiver_idx.
/// Set from ClientHello's proposed receiver_index, or from header for non-bootstrap packets.
bound_receiver_idx: Option<u32>,
/// Persistent connection to exit gateway for forwarding.
/// Opened on first forward, reused for subsequent forwards, closed when client disconnects.
/// Tuple contains (stream, target_address) to verify subsequent forwards go to same exit.
exit_stream: Option<(TcpStream, SocketAddr)>,
exit_stream: Option<(S, SocketAddr)>,
}
impl<S> LpConnectionHandler<S>
where
S: AsyncRead + AsyncWrite + Unpin,
S: LpTransport + Unpin,
{
pub fn new(stream: S, remote_addr: SocketAddr, state: LpHandlerState) -> Self {
Self {
@@ -622,13 +626,15 @@ where
receiver_idx: u32,
decrypted_bytes: Vec<u8>,
) -> Result<(), GatewayError> {
let remote = self.remote_addr;
// Try to deserialize as LpRegistrationRequest first (most common case after handshake)
if let Ok(request) =
lp_bincode_serializer().deserialize::<LpRegistrationRequest>(&decrypted_bytes)
{
debug!(
"LP registration request from {} (receiver_idx={}): mode={:?}",
self.remote_addr, receiver_idx, request.mode
"LP registration request from {remote} (receiver_idx={receiver_idx}): mode={:?}",
request.mode
);
return self
.handle_registration_request(receiver_idx, request)
@@ -640,8 +646,8 @@ where
lp_bincode_serializer().deserialize::<ForwardPacketData>(&decrypted_bytes)
{
debug!(
"LP forward request from {} (receiver_idx={}) to {}",
self.remote_addr, receiver_idx, forward_data.target_lp_address
"LP forward request from {remote} (receiver_idx={receiver_idx}) to {}",
forward_data.target_lp_address
);
return self
.handle_forwarding_request(receiver_idx, forward_data)
@@ -649,10 +655,7 @@ where
}
// Neither registration nor forwarding - unknown payload type
warn!(
"Unknown transport payload type from {} (receiver_idx={})",
self.remote_addr, receiver_idx
);
warn!("Unknown transport payload type from {remote} (receiver_idx={receiver_idx})");
inc!("lp_errors_unknown_payload_type");
Err(GatewayError::LpProtocolError(
"Unknown transport payload type (not registration or forwarding)".to_string(),
@@ -790,7 +793,6 @@ where
///
/// Entry gateway receives ForwardPacketData from client, forwards inner packet
/// to exit gateway, receives response, encrypts it, and sends back to client.
/// Connection closes after response is sent (single-packet model).
async fn handle_forwarding_request(
&mut self,
receiver_idx: u32,
@@ -955,15 +957,63 @@ where
}
}
/// Forward an LP packet to another gateway (single-packet model)
/// Returns reference to the established forwarding channel to the exit.
pub fn forwarding_channel(&self) -> &Option<(S, SocketAddr)> {
&self.exit_stream
}
/// This method establishes connection to the target gateway in order to
/// forward received packets and retrieve any responses
//
// In the future it will also perform identity validation to make sure
// the target node is a valid gateway present in the network
//
// Do not manually call this function. It is only exposed for the purposes of integration tests
#[doc(hidden)]
pub async fn establish_exit_stream(
&mut self,
target_addr: SocketAddr,
) -> Result<(), GatewayError> {
// Acquire semaphore permit to limit concurrent connection opens (FD exhaustion protection)
// Permit is scoped to this block - only protects the connect() call, not stream reuse
let _permit = match self.state.forward_semaphore.try_acquire() {
Ok(permit) => permit,
Err(_) => {
inc!("lp_forward_rejected");
return Err(GatewayError::LpConnectionError(
"Gateway at forward capacity".into(),
));
}
};
// Connect to target gateway with timeout
let stream = match timeout(Duration::from_secs(5), S::connect(target_addr)).await {
Ok(Ok(stream)) => stream,
Ok(Err(e)) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(format!(
"Failed to connect to target gateway: {e}",
)));
}
Err(_) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(
"Target gateway connection timeout".to_string(),
));
}
};
debug!("Opened persistent exit connection to {target_addr} for forwarding");
self.exit_stream = Some((stream, target_addr));
Ok(())
}
/// Forward an LP packet to another gateway
///
/// This method connects to the target gateway, forwards the inner packet bytes,
/// receives the response, and returns it. Used for telescoping (hiding client IP).
///
/// Called from `handle_forwarding_request()` as part of the single-packet-per-connection
/// architecture. Each forward request arrives on a new connection, gets processed,
/// response sent, and connection closes.
///
/// # Arguments
/// * `forward_data` - ForwardPacketData containing target gateway info and inner packet
///
@@ -1015,38 +1065,7 @@ where
};
if need_new_connection {
// Acquire semaphore permit to limit concurrent connection opens (FD exhaustion protection)
// Permit is scoped to this block - only protects the connect() call, not stream reuse
let _permit = match self.state.forward_semaphore.try_acquire() {
Ok(permit) => permit,
Err(_) => {
inc!("lp_forward_rejected");
return Err(GatewayError::LpConnectionError(
"Gateway at forward capacity".into(),
));
}
};
// Connect to target gateway with timeout
let stream =
match timeout(Duration::from_secs(5), TcpStream::connect(target_addr)).await {
Ok(Ok(stream)) => stream,
Ok(Err(e)) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(format!(
"Failed to connect to target gateway: {e}",
)));
}
Err(_) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(
"Target gateway connection timeout".to_string(),
));
}
};
debug!("Opened persistent exit connection to {target_addr} for forwarding",);
self.exit_stream = Some((stream, target_addr));
self.establish_exit_stream(target_addr).await?;
}
// Get mutable reference to the exit stream
@@ -1099,8 +1118,7 @@ where
const MAX_PACKET_SIZE: usize = 65536;
if response_len > MAX_PACKET_SIZE {
return Err(GatewayError::LpProtocolError(format!(
"Response size {} exceeds maximum {}",
response_len, MAX_PACKET_SIZE
"Response size {response_len} exceeds maximum {MAX_PACKET_SIZE}",
)));
}
@@ -1110,8 +1128,7 @@ where
.await
.map_err(|e| {
GatewayError::LpConnectionError(format!(
"Failed to read response from target: {}",
e
"Failed to read response from target: {e}",
))
})?;
@@ -1276,7 +1293,7 @@ mod tests {
use nym_lp::packet::{LpHeader, LpPacket};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
// ==================== Test Helpers ====================
integration-tests/Cargo.toml
+3 -1
View File
@@ -20,9 +20,11 @@ nym-crypto = { path = "../common/crypto", features = ["asymmetric", "rand"] }
nym-credential-verification = { path = "../common/credential-verification" }
nym-credentials-interface = { path = "../common/credentials-interface" }
nym-test-utils = { path = "../common/test-utils" }
nym-registration-client = { path = "../nym-registration-client", features = ["io-mocks"] }
nym-registration-client = { path = "../nym-registration-client" }
nym-lp-transport = { path = "../common/nym-lp-transport", features = ["io-mocks"] }
nym-gateway = { path = "../gateway" }
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite"] }
tracing = { workspace = true }
[lints]
integration-tests/src/lp_registration.rs
+276 -43
View File
@@ -25,15 +25,14 @@ mod tests {
RegisteredResponse, mock_peer_controller,
};
use nym_wireguard::{IpPool, WireguardConfig};
use std::any::Any;
use std::mem;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::Semaphore;
use tokio::sync::mpsc::Receiver;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::error;
trait WgKeyConv {
fn to_wg_key(self) -> KeyWrapper;
@@ -91,19 +90,32 @@ mod tests {
Invalid,
}
struct EntryGateway {
enum SpawnedLpConnectionHandlerState {
NotCreated,
Ready {
handler: LpConnectionHandler<MockIOStream>,
},
Running {
handle: JoinHandle<Option<Result<(), GatewayError>>>,
},
Finished,
}
struct Gateway {
base: Party,
lp_state: LpHandlerState,
ip_pool: IpPool,
// might be used later for mixnet registration tests
#[allow(unused)]
mix_receiver: MixForwardingReceiver,
mock_peer_controller: SpawnedPeerController,
mock_peer_controller_state: MockPeerControllerState,
handle_cancellation: CancellationToken,
handler_handle: Option<JoinHandle<Option<Result<(), GatewayError>>>>,
tasks_cancellation: CancellationToken,
mock_peer_controller_state: MockPeerControllerState,
lp_connection_handler: SpawnedLpConnectionHandlerState,
}
impl EntryGateway {
impl Gateway {
async fn register_peer_controller_response(
&self,
request: PeerControlRequestType,
@@ -219,7 +231,7 @@ mod tests {
forward_semaphore,
};
Ok(EntryGateway {
Ok(Gateway {
base,
lp_state,
ip_pool: Self::ip_pool(),
@@ -228,26 +240,68 @@ mod tests {
controller: mock_peer_controller,
},
mock_peer_controller_state: peer_controller_state,
handle_cancellation: Default::default(),
handler_handle: None,
tasks_cancellation: Default::default(),
lp_connection_handler: SpawnedLpConnectionHandlerState::NotCreated,
})
}
fn spawn_lp_handler(
fn create_lp_handler(
&mut self,
client_connection: MockIOStream,
client_address: SocketAddr,
) {
assert!(self.handler_handle.is_none());
let mut gateway_lp_handler =
LpConnectionHandler::new(client_connection, client_address, self.lp_state.clone());
let SpawnedLpConnectionHandlerState::NotCreated = self.lp_connection_handler else {
panic!("lp connection handler in invalid state")
};
let cancellation_token = self.handle_cancellation.clone();
self.handler_handle = Some(tokio::spawn(async move {
cancellation_token
.run_until_cancelled(gateway_lp_handler.handle())
.await
}));
self.lp_connection_handler = SpawnedLpConnectionHandlerState::Ready {
handler: LpConnectionHandler::new(
client_connection,
client_address,
self.lp_state.clone(),
),
};
}
async fn establish_forwarding_channel(
&mut self,
exit_address: SocketAddr,
) -> anyhow::Result<MockIOStream> {
let SpawnedLpConnectionHandlerState::Ready { handler } =
&mut self.lp_connection_handler
else {
panic!("lp connection handler in invalid state")
};
handler.establish_exit_stream(exit_address).await?;
Ok(handler
.forwarding_channel()
.as_ref()
.context("mock connection has failed!")?
.0
.try_get_remote_handle())
}
fn spawn_lp_handler(&mut self) {
let SpawnedLpConnectionHandlerState::Ready { handler } = mem::replace(
&mut self.lp_connection_handler,
SpawnedLpConnectionHandlerState::NotCreated,
) else {
panic!("lp connection handler in invalid state")
};
let cancellation_token = self.tasks_cancellation.clone();
self.lp_connection_handler = SpawnedLpConnectionHandlerState::Running {
handle: tokio::spawn(async move {
let handler_future = async move {
handler
.handle()
.await
.inspect_err(|err| error!("lp handler has failed: {err}"))
};
cancellation_token.run_until_cancelled(handler_future).await
}),
}
}
fn spawn_peer_controller(&mut self) {
@@ -258,7 +312,7 @@ mod tests {
panic!("mock peer controller in invalid state")
};
let cancellation_token = self.handle_cancellation.clone();
let cancellation_token = self.tasks_cancellation.clone();
let join_handle = tokio::spawn(async move {
cancellation_token
.run_until_cancelled(controller.run())
@@ -271,36 +325,37 @@ mod tests {
#[allow(clippy::panic)]
async fn stop_tasks(&mut self) -> anyhow::Result<()> {
self.handle_cancellation.cancel();
self.tasks_cancellation.cancel();
if let Some(handle) = self.handler_handle.take() {
if let Some(Err(err)) = handle.timeboxed().await?.context("join failure")? {
panic!("gateway handler failure: {err}")
}
}
let SpawnedLpConnectionHandlerState::Running { handle: lp_handle } = mem::replace(
&mut self.lp_connection_handler,
SpawnedLpConnectionHandlerState::NotCreated,
) else {
panic!("lp connection handler in invalid state")
};
let SpawnedPeerController::Running { handle } = mem::replace(
let SpawnedPeerController::Running {
handle: peer_controller_handle,
} = mem::replace(
&mut self.mock_peer_controller,
SpawnedPeerController::Invalid,
) else {
)
else {
panic!("mock peer controller in invalid state")
};
handle.timeboxed().await?.context("join failure")?;
lp_handle.timeboxed().await?.context("join failure")?;
peer_controller_handle
.timeboxed()
.await?
.context("join failure")?;
self.mock_peer_controller = SpawnedPeerController::Finished;
self.lp_connection_handler = SpawnedLpConnectionHandlerState::Finished;
Ok(())
}
}
fn mock_client_address() -> SocketAddr {
SocketAddr::from(([1, 2, 3, 4], 5678))
}
fn mock_gateway_address() -> SocketAddr {
SocketAddr::from(([8, 7, 6, 5], 4321))
}
#[cfg(test)]
mod using_registration_client {
@@ -311,6 +366,7 @@ mod tests {
#[cfg(test)]
mod using_lp_registration_client {
use super::*;
use nym_registration_client::NestedLpSession;
#[tokio::test]
async fn test_basic_lp_entry_registration() -> anyhow::Result<()> {
@@ -321,7 +377,7 @@ mod tests {
let client_data = Client::mock(&mut client_rng);
let client_key = *client_data.base.x25519_wg_keys.public_key();
let mut entry = EntryGateway::mock(&mut gateway_rng).await?;
let mut entry = Gateway::mock(&mut gateway_rng).await?;
let mut client = LpRegistrationClient::<MockIOStream>::new_with_default_psk(
client_data.base.ed25519_keys,
@@ -339,7 +395,8 @@ mod tests {
.try_get_remote_handle();
// 2. create and spawn gateway handler for the client connection
entry.spawn_lp_handler(gateway_conn, client_data.base.socket_addr);
entry.create_lp_handler(gateway_conn, client_data.base.socket_addr);
entry.spawn_lp_handler();
// 3. register all needed responses for the dvpn registration that will reach the peer controller
// 1) peer registration - ip pair allocation
@@ -411,8 +468,7 @@ mod tests {
let mut gateway_rng = u64_seeded_rng(1);
let client_data = Client::mock(&mut client_rng);
let client_key = *client_data.base.x25519_wg_keys.public_key();
let mut entry = EntryGateway::mock(&mut gateway_rng).await?;
let mut entry = Gateway::mock(&mut gateway_rng).await?;
let mut client = LpRegistrationClient::<MockIOStream>::new_with_default_psk(
client_data.base.ed25519_keys,
@@ -430,7 +486,8 @@ mod tests {
.try_get_remote_handle();
// 2. create and spawn gateway handler for the client connection
entry.spawn_lp_handler(gateway_conn, client_data.base.socket_addr);
entry.create_lp_handler(gateway_conn, client_data.base.socket_addr);
entry.spawn_lp_handler();
// 3. spawn peer controller to be able to handle dvpn registration requests
// (which we shouldn't receive anyway)
@@ -460,5 +517,181 @@ mod tests {
entry.stop_tasks().await?;
Ok(())
}
#[tokio::test]
async fn test_basic_lp_exit_registration() -> anyhow::Result<()> {
// initialise random, but deterministic, keys, addresses, etc. for the parties
let mut client_rng = u64_seeded_rng(0);
let mut entry_rng = u64_seeded_rng(1);
let mut exit_rng = u64_seeded_rng(2);
let client_data = Client::mock(&mut client_rng);
let client_key = *client_data.base.x25519_wg_keys.public_key();
let mut entry = Gateway::mock(&mut entry_rng).await?;
let mut exit = Gateway::mock(&mut exit_rng).await?;
let mut entry_client = LpRegistrationClient::<MockIOStream>::new_with_default_psk(
client_data.base.ed25519_keys.clone(),
*entry.base.ed25519_keys.public_key(),
entry.base.socket_addr,
client_data.base.socket_addr.ip(),
);
// START: ENTRY SETUP
//
// 1. establish mock connection between client and gateway and retrieve gateway's handle
entry_client.ensure_connected().await?;
let entry_conn = entry_client
.connection()
.as_ref()
.context("mock connection has failed!")?
.try_get_remote_handle();
entry_conn.set_id(1);
// 2. create handler for the client connection (entry)
entry.create_lp_handler(entry_conn, client_data.base.socket_addr);
// 3. pre-establish connection between entry and exit
let exit_conn = entry
.establish_forwarding_channel(exit.base.socket_addr)
.await?;
exit_conn.set_id(255);
// 4. register all needed responses for the dvpn registration that will reach the peer controller
// 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 },
reg_res,
)
.await;
// 2) new peer inclusion - in non-mock system it would spawn handlers,
// here we'll just set a flag and say it's all fine
let public_key = client_key.to_wg_key();
let add_res = Ok::<_, nym_wireguard::Error>(());
entry
.register_peer_controller_response(
PeerControlRequestType::AddPeer { public_key },
add_res,
)
.await;
// 5. spawn peer controller to be able to handle dvpn registration requests
entry.spawn_peer_controller();
// 6. finally spawn the handler
entry.spawn_lp_handler();
// 7. perform client handshake (with the entry)
entry_client.perform_handshake().timeboxed().await??;
// END: ENTRY SETUP
//
// START: EXIT SETUP:
// 8. create handler for the forwarding channel (exit)
exit.create_lp_handler(exit_conn, client_data.base.socket_addr);
// 9. spawn the handler
exit.spawn_lp_handler();
// 10. register all needed responses for the dvpn registration that will reach the peer controller
// 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 },
reg_res,
)
.await;
// 2) new peer inclusion - in non-mock system it would spawn handlers,
// here we'll just set a flag and say it's all fine
let public_key = client_key.to_wg_key();
let add_res = Ok::<_, nym_wireguard::Error>(());
exit.register_peer_controller_response(
PeerControlRequestType::AddPeer { public_key },
add_res,
)
.await;
// 11. spawn peer controller to be able to handle dvpn registration requests
exit.spawn_peer_controller();
// END: EXIT SETUP
// 12. create nested session to register with exit via forwarding
// technically we should use different ephemeral keys than we had for the entry
// but crypto is going to work the same
let mut nested_session = NestedLpSession::new(
exit.base.ed25519_keys.public_key().to_bytes(),
exit.base.socket_addr.to_string(),
client_data.base.ed25519_keys,
*exit.base.ed25519_keys.public_key(),
);
// 13. Perform handshake and registration with exit gateway (all via entry forwarding)
let exit_registration_result = nested_session
.handshake_and_register(
&mut entry_client,
&client_data.base.x25519_wg_keys,
exit.base.ed25519_keys.public_key(),
&client_data.ticket_provider,
TicketType::V1WireguardExit,
client_data.base.socket_addr.ip(),
)
.timeboxed()
.await??;
// 14. complete registration with the entry
let entry_registration_result = entry_client
.register(
&client_data.base.x25519_wg_keys,
entry.base.ed25519_keys.public_key(),
&client_data.ticket_provider,
TicketType::V1WireguardEntry,
)
.timeboxed()
.await??;
// 15. verify all registration results
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);
assert_eq!(entry_registration_result.private_ipv6, entry_ip_pair.ipv6);
assert_eq!(
entry_registration_result.public_key,
*entry.base.x25519_wg_keys.public_key()
);
assert_eq!(exit_registration_result.private_ipv4, exit_ip_pair.ipv4);
assert_eq!(exit_registration_result.private_ipv6, exit_ip_pair.ipv6);
assert_eq!(
exit_registration_result.public_key,
*exit.base.x25519_wg_keys.public_key()
);
// 16. stop the gateway task and finish the test
entry.stop_tasks().await?;
exit.stop_tasks().await?;
Ok(())
}
}
}
nym-registration-client/Cargo.toml
+1 -5
View File
@@ -29,12 +29,8 @@ nym-credentials-interface = { path = "../common/credentials-interface" }
nym-crypto = { path = "../common/crypto" }
nym-ip-packet-client = { path = "../nym-ip-packet-client" }
nym-lp = { path = "../common/nym-lp" }
nym-lp-transport = { path = "../common/nym-lp-transport" }
nym-registration-common = { path = "../common/registration" }
nym-sdk = { path = "../sdk/rust/nym-sdk" }
nym-validator-client = { path = "../common/client-libs/validator-client" }
nym-wireguard-types = { path = "../common/wireguard-types" }
nym-test-utils = { path = "../common/test-utils", optional = true }
[features]
io-mocks = ["nym-test-utils"]
nym-registration-client/src/lib.rs
+8 -12
View File
@@ -3,17 +3,15 @@
use tokio_util::sync::CancellationToken;
use nym_authenticator_client::{
AuthClientMixnetListener, AuthClientMixnetListenerHandle, AuthenticatorClient,
};
use crate::config::RegistrationClientConfig;
use nym_authenticator_client::{AuthClientMixnetListener, AuthenticatorClient};
use nym_bandwidth_controller::BandwidthTicketProvider;
use nym_credentials_interface::TicketType;
use nym_ip_packet_client::IprClientConnect;
use nym_registration_common::AssignedAddresses;
use nym_sdk::mixnet::{EventReceiver, MixnetClient, Recipient};
use std::sync::Arc;
use crate::config::RegistrationClientConfig;
use tokio::net::TcpStream;
mod builder;
mod config;
@@ -28,9 +26,7 @@ pub use builder::config::{
};
pub use config::RegistrationMode;
pub use error::RegistrationClientError;
pub use lp_client::{
LpConfig, LpRegistrationClient, NestedLpSession, error::LpClientError, traits::LpTransportLayer,
};
pub use lp_client::{LpConfig, LpRegistrationClient, NestedLpSession, error::LpClientError};
pub use types::{
LpRegistrationResult, MixnetRegistrationResult, RegistrationResult, WireguardRegistrationResult,
};
@@ -171,8 +167,8 @@ impl RegistrationClient {
},
)?;
tracing::debug!("Entry gateway LP address: {}", entry_lp_address);
tracing::debug!("Exit gateway LP address: {}", exit_lp_address);
tracing::debug!("Entry gateway LP address: {entry_lp_address}");
tracing::debug!("Exit gateway LP address: {exit_lp_address}");
// Generate fresh Ed25519 keypairs for LP registration
// These are ephemeral and used only for the LP handshake protocol
@@ -185,7 +181,7 @@ impl RegistrationClient {
// This creates the LP session that will be used to forward packets to exit.
// Uses packet-per-connection model: each handshake packet on new TCP connection.
tracing::info!("Establishing outer session with entry gateway");
let mut entry_client = LpRegistrationClient::new_with_default_psk(
let mut entry_client = LpRegistrationClient::<TcpStream>::new_with_default_psk(
entry_lp_keypair.clone(),
self.config.entry.node.identity,
entry_lp_address,
@@ -215,7 +211,7 @@ impl RegistrationClient {
// Perform handshake and registration with exit gateway (all via entry forwarding)
let exit_gateway_data = nested_session
.handshake_and_register(
.handshake_and_register::<TcpStream>(
&mut entry_client,
&self.config.exit.keys,
&self.config.exit.node.identity,
nym-registration-client/src/lp_client/client.rs
+4 -4
View File
@@ -5,7 +5,6 @@
use super::config::LpConfig;
use super::error::{LpClientError, Result};
use crate::lp_client::traits::LpTransportLayer;
use bytes::BytesMut;
use nym_bandwidth_controller::{BandwidthTicketProvider, DEFAULT_TICKETS_TO_SPEND};
use nym_credentials_interface::{CredentialSpendingData, TicketType};
@@ -15,6 +14,7 @@ use nym_lp::codec::{OuterAeadKey, parse_lp_packet, serialize_lp_packet};
use nym_lp::message::ForwardPacketData;
use nym_lp::serialisation::{BincodeOptions, lp_bincode_serializer};
use nym_lp::state_machine::{LpAction, LpInput, LpStateMachine};
use nym_lp_transport::traits::LpTransport;
use nym_registration_common::{GatewayData, LpRegistrationRequest, LpRegistrationResponse};
use nym_wireguard_types::PeerPublicKey;
use std::net::{IpAddr, SocketAddr};
@@ -63,7 +63,7 @@ pub struct LpRegistrationClient<S = TcpStream> {
impl<S> LpRegistrationClient<S>
where
S: LpTransportLayer + Unpin,
S: LpTransport + Unpin,
{
/// Creates a new LP registration client.
///
@@ -186,7 +186,7 @@ where
// Set TCP_NODELAY for low latency
stream
.configure(&self.config)
.set_no_delay(self.config.tcp_nodelay)
.map_err(|source| LpClientError::TcpConnection {
address: self.gateway_lp_address.to_string(),
source,
@@ -563,7 +563,7 @@ where
// 2. Set TCP_NODELAY
stream
.configure(config)
.set_no_delay(config.tcp_nodelay)
.map_err(|source| LpClientError::TcpConnection {
address: address.to_string(),
source,
nym-registration-client/src/lp_client/mod.rs
-1
View File
@@ -35,7 +35,6 @@ mod client;
mod config;
pub(crate) mod error;
mod nested_session;
pub mod traits;
pub use client::LpRegistrationClient;
pub use config::LpConfig;
nym-registration-client/src/lp_client/nested_session.rs
+36 -18
View File
@@ -28,6 +28,7 @@ use nym_lp::codec::{OuterAeadKey, parse_lp_packet, serialize_lp_packet};
use nym_lp::serialisation::{BincodeOptions, lp_bincode_serializer};
use nym_lp::state_machine::{LpAction, LpInput, LpStateMachine};
use nym_lp::{LpMessage, LpPacket};
use nym_lp_transport::traits::LpTransport;
use nym_registration_common::{GatewayData, LpRegistrationRequest, LpRegistrationResponse};
use nym_wireguard_types::PeerPublicKey;
use std::net::IpAddr;
@@ -112,7 +113,13 @@ impl NestedLpSession {
/// - Forwarding through entry gateway fails
/// - Exit gateway handshake fails
/// - Cryptographic operations fail
async fn perform_handshake(&mut self, outer_client: &mut LpRegistrationClient) -> Result<()> {
async fn perform_handshake<S>(
&mut self,
outer_client: &mut LpRegistrationClient<S>,
) -> Result<()>
where
S: LpTransport + Unpin,
{
tracing::debug!(
"Starting nested LP handshake with exit gateway {}",
self.exit_address
@@ -125,7 +132,7 @@ impl NestedLpSession {
let timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("System time before UNIX epoch")
.map_err(|_| LpClientError::Other("System time before UNIX epoch".into()))?
.as_secs();
// Step 2: Generate ClientHello for exit gateway
@@ -170,8 +177,7 @@ impl NestedLpSession {
}
LpMessage::Collision => {
return Err(LpClientError::Transport(format!(
"Exit gateway returned Collision - receiver_index {} already in use",
receiver_index
"Exit gateway returned Collision - receiver_index {receiver_index} already in use",
)));
}
other => {
@@ -203,8 +209,7 @@ impl NestedLpSession {
}
other => {
return Err(LpClientError::Transport(format!(
"Unexpected action at handshake start: {:?}",
other
"Unexpected action at handshake start: {other:?}",
)));
}
}
@@ -293,14 +298,17 @@ impl NestedLpSession {
///
/// # Returns
/// * `Ok(GatewayData)` - Exit gateway configuration data on successful registration
pub async fn handshake_and_register_with_credential(
pub async fn handshake_and_register_with_credential<S>(
&mut self,
outer_client: &mut LpRegistrationClient,
outer_client: &mut LpRegistrationClient<S>,
wg_keypair: &x25519::KeyPair,
credential: nym_credentials_interface::CredentialSpendingData,
ticket_type: TicketType,
client_ip: IpAddr,
) -> Result<GatewayData> {
) -> Result<GatewayData>
where
S: LpTransport + Unpin,
{
// Step 1: Perform handshake with exit gateway via forwarding
self.perform_handshake(outer_client).await?;
@@ -454,15 +462,18 @@ impl NestedLpSession {
/// - Forwarding through entry gateway fails
/// - Response decryption/deserialization fails
/// - Gateway rejects the registration
pub async fn handshake_and_register(
pub async fn handshake_and_register<S>(
&mut self,
outer_client: &mut LpRegistrationClient,
outer_client: &mut LpRegistrationClient<S>,
wg_keypair: &x25519::KeyPair,
gateway_identity: &ed25519::PublicKey,
bandwidth_controller: &dyn BandwidthTicketProvider,
ticket_type: TicketType,
client_ip: IpAddr,
) -> Result<GatewayData> {
) -> Result<GatewayData>
where
S: LpTransport + Unpin,
{
// Step 1: Perform handshake with exit gateway via forwarding
self.perform_handshake(outer_client).await?;
@@ -628,16 +639,20 @@ impl NestedLpSession {
///
/// # Errors
/// Returns an error if all retry attempts fail.
pub async fn handshake_and_register_with_retry(
#[allow(clippy::too_many_arguments)]
pub async fn handshake_and_register_with_retry<S>(
&mut self,
outer_client: &mut LpRegistrationClient,
outer_client: &mut LpRegistrationClient<S>,
wg_keypair: &x25519::KeyPair,
gateway_identity: &ed25519::PublicKey,
bandwidth_controller: &dyn BandwidthTicketProvider,
ticket_type: TicketType,
client_ip: IpAddr,
max_retries: u32,
) -> Result<GatewayData> {
) -> Result<GatewayData>
where
S: LpTransport + Unpin,
{
tracing::debug!(
"Starting resilient exit registration (max_retries={})",
max_retries
@@ -719,12 +734,15 @@ impl NestedLpSession {
/// 2. Serializes the packet with outer encryption
/// 3. Forwards via entry gateway
/// 4. Parses and returns the response
async fn send_and_receive_via_forward(
async fn send_and_receive_via_forward<S>(
&self,
outer_client: &mut LpRegistrationClient,
outer_client: &mut LpRegistrationClient<S>,
state_machine: &LpStateMachine,
packet: &LpPacket,
) -> Result<LpPacket> {
) -> Result<LpPacket>
where
S: LpTransport + Unpin,
{
let send_key = Self::get_send_key(state_machine);
let packet_bytes = Self::serialize_packet(packet, send_key.as_ref())?;
let response_bytes = outer_client