From 5fa1a3fc17e5f15aba395be3bcb7a459e3bc8d33 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 8 Jan 2026 11:06:52 +0000 Subject: [PATCH] Integration test for nested LP registration - 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 --- Cargo.lock | 13 +- Cargo.toml | 2 +- common/credentials-interface/src/lib.rs | 17 +- common/nym-lp-common/src/lib.rs | 3 + common/nym-lp-transport/Cargo.toml | 21 ++ common/nym-lp-transport/src/lib.rs | 4 + .../nym-lp-transport/src}/traits.rs | 20 +- common/nymsphinx/framing/src/processing.rs | 4 +- .../test-utils/src/mocks/async_read_write.rs | 58 +++- gateway/Cargo.toml | 1 + gateway/src/error.rs | 2 +- gateway/src/node/lp_listener/handler.rs | 125 ++++--- integration-tests/Cargo.toml | 4 +- integration-tests/src/lp_registration.rs | 319 +++++++++++++++--- nym-registration-client/Cargo.toml | 6 +- nym-registration-client/src/lib.rs | 20 +- .../src/lp_client/client.rs | 8 +- nym-registration-client/src/lp_client/mod.rs | 1 - .../src/lp_client/nested_session.rs | 54 ++- 19 files changed, 521 insertions(+), 161 deletions(-) create mode 100644 common/nym-lp-transport/Cargo.toml create mode 100644 common/nym-lp-transport/src/lib.rs rename {nym-registration-client/src/lp_client => common/nym-lp-transport/src}/traits.rs (65%) diff --git a/Cargo.lock b/Cargo.lock index 83062d7382..fc8a7953f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/Cargo.toml b/Cargo.toml index 647ba0debd..3e9be3b06b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -174,7 +174,7 @@ members = [ "wasm/node-tester", "wasm/zknym-lib", "nym-gateway-probe", - "integration-tests", + "integration-tests", "common/nym-lp-transport", ] default-members = [ diff --git a/common/credentials-interface/src/lib.rs b/common/credentials-interface/src/lib.rs index 807016792b..3c19e3908d 100644 --- a/common/credentials-interface/src/lib.rs +++ b/common/credentials-interface/src/lib.rs @@ -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( diff --git a/common/nym-lp-common/src/lib.rs b/common/nym-lp-common/src/lib.rs index 4b628789e0..98da6cebd3 100644 --- a/common/nym-lp-common/src/lib.rs +++ b/common/nym-lp-common/src/lib.rs @@ -1,3 +1,6 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + use std::fmt; use std::fmt::Write; diff --git a/common/nym-lp-transport/Cargo.toml b/common/nym-lp-transport/Cargo.toml new file mode 100644 index 0000000000..c2aba1690d --- /dev/null +++ b/common/nym-lp-transport/Cargo.toml @@ -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 diff --git a/common/nym-lp-transport/src/lib.rs b/common/nym-lp-transport/src/lib.rs new file mode 100644 index 0000000000..3f62a665ed --- /dev/null +++ b/common/nym-lp-transport/src/lib.rs @@ -0,0 +1,4 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +pub mod traits; diff --git a/nym-registration-client/src/lp_client/traits.rs b/common/nym-lp-transport/src/traits.rs similarity index 65% rename from nym-registration-client/src/lp_client/traits.rs rename to common/nym-lp-transport/src/traits.rs index 8fd091283d..5b3d300884 100644 --- a/nym-registration-client/src/lp_client/traits.rs +++ b/common/nym-lp-transport/src/traits.rs @@ -1,40 +1,38 @@ // Copyright 2026 - Nym Technologies SA // 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; - 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 { 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 { Ok(MockIOStream::default()) } - fn configure(&mut self, _config: &LpConfig) -> std::io::Result<()> { + fn set_no_delay(&mut self, _nodelay: bool) -> std::io::Result<()> { Ok(()) } } diff --git a/common/nymsphinx/framing/src/processing.rs b/common/nymsphinx/framing/src/processing.rs index 817419c537..cb6990895b 100644 --- a/common/nymsphinx/framing/src/processing.rs +++ b/common/nymsphinx/framing/src/processing.rs @@ -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()); } }; diff --git a/common/test-utils/src/mocks/async_read_write.rs b/common/test-utils/src/mocks/async_read_write.rs index 1004ce6dd0..1ee5a6e57d 100644 --- a/common/test-utils/src/mocks/async_read_write.rs +++ b/common/test-utils/src/mocks/async_read_write.rs @@ -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, + + // side of the stream to use for logging purposes + side: Side, // messages to send tx: InnerWrapper>, @@ -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 { 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)) } diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index b96bda54fa..2eccd66525 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -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 } diff --git a/gateway/src/error.rs b/gateway/src/error.rs index 0e6d4f73bc..7801ef01d2 100644 --- a/gateway/src/error.rs +++ b/gateway/src/error.rs @@ -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), diff --git a/gateway/src/node/lp_listener/handler.rs b/gateway/src/node/lp_listener/handler.rs index 6d802dd611..8fe7891401 100644 --- a/gateway/src/node/lp_listener/handler.rs +++ b/gateway/src/node/lp_listener/handler.rs @@ -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 { /// 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, + /// 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 LpConnectionHandler 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, ) -> 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::(&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::(&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 ==================== diff --git a/integration-tests/Cargo.toml b/integration-tests/Cargo.toml index fe02d83dcb..2023980518 100644 --- a/integration-tests/Cargo.toml +++ b/integration-tests/Cargo.toml @@ -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] diff --git a/integration-tests/src/lp_registration.rs b/integration-tests/src/lp_registration.rs index 1720888b8b..79350fb3ac 100644 --- a/integration-tests/src/lp_registration.rs +++ b/integration-tests/src/lp_registration.rs @@ -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, + }, + Running { + handle: JoinHandle>>, + }, + 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>>>, + 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 { + 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::::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::::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::::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(()) + } } } diff --git a/nym-registration-client/Cargo.toml b/nym-registration-client/Cargo.toml index a90ede9da1..efa0b5c5a6 100644 --- a/nym-registration-client/Cargo.toml +++ b/nym-registration-client/Cargo.toml @@ -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"] \ No newline at end of file diff --git a/nym-registration-client/src/lib.rs b/nym-registration-client/src/lib.rs index 0813af6627..516f7c6245 100644 --- a/nym-registration-client/src/lib.rs +++ b/nym-registration-client/src/lib.rs @@ -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::::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::( &mut entry_client, &self.config.exit.keys, &self.config.exit.node.identity, diff --git a/nym-registration-client/src/lp_client/client.rs b/nym-registration-client/src/lp_client/client.rs index d35d98a474..a7b482e4c5 100644 --- a/nym-registration-client/src/lp_client/client.rs +++ b/nym-registration-client/src/lp_client/client.rs @@ -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 { impl LpRegistrationClient 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, diff --git a/nym-registration-client/src/lp_client/mod.rs b/nym-registration-client/src/lp_client/mod.rs index 36f1783ef3..d4085bbbaa 100644 --- a/nym-registration-client/src/lp_client/mod.rs +++ b/nym-registration-client/src/lp_client/mod.rs @@ -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; diff --git a/nym-registration-client/src/lp_client/nested_session.rs b/nym-registration-client/src/lp_client/nested_session.rs index c1c834cd50..d4f35898ee 100644 --- a/nym-registration-client/src/lp_client/nested_session.rs +++ b/nym-registration-client/src/lp_client/nested_session.rs @@ -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( + &mut self, + outer_client: &mut LpRegistrationClient, + ) -> 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( &mut self, - outer_client: &mut LpRegistrationClient, + outer_client: &mut LpRegistrationClient, wg_keypair: &x25519::KeyPair, credential: nym_credentials_interface::CredentialSpendingData, ticket_type: TicketType, client_ip: IpAddr, - ) -> Result { + ) -> Result + 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( &mut self, - outer_client: &mut LpRegistrationClient, + outer_client: &mut LpRegistrationClient, wg_keypair: &x25519::KeyPair, gateway_identity: &ed25519::PublicKey, bandwidth_controller: &dyn BandwidthTicketProvider, ticket_type: TicketType, client_ip: IpAddr, - ) -> Result { + ) -> Result + 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( &mut self, - outer_client: &mut LpRegistrationClient, + outer_client: &mut LpRegistrationClient, wg_keypair: &x25519::KeyPair, gateway_identity: &ed25519::PublicKey, bandwidth_controller: &dyn BandwidthTicketProvider, ticket_type: TicketType, client_ip: IpAddr, max_retries: u32, - ) -> Result { + ) -> Result + 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( &self, - outer_client: &mut LpRegistrationClient, + outer_client: &mut LpRegistrationClient, state_machine: &LpStateMachine, packet: &LpPacket, - ) -> Result { + ) -> Result + 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