From c6589ca92c7b8d51b0cba19e856fff1fb356ba1c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 5 Mar 2026 10:56:26 +0000 Subject: [PATCH 1/4] chore: add unit test for mutual KKT --- Cargo.lock | 1 + common/nym-kkt/Cargo.toml | 1 + common/nym-kkt/src/lib.rs | 181 +++++++++++++++++++++++++------------- 3 files changed, 124 insertions(+), 59 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 55823ce4f5..e7bdcdd1a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6833,6 +6833,7 @@ dependencies = [ "nym-kkt-ciphersuite", "nym-kkt-context", "nym-pemstore", + "nym-test-utils", "rand 0.9.2", "rand_chacha 0.9.0", "strum 0.28.0", diff --git a/common/nym-kkt/Cargo.toml b/common/nym-kkt/Cargo.toml index 38ddd37582..02c9f50d49 100644 --- a/common/nym-kkt/Cargo.toml +++ b/common/nym-kkt/Cargo.toml @@ -30,6 +30,7 @@ libcrux-ml-kem = { workspace = true } [dev-dependencies] rand_chacha = "0.9.0" anyhow = { workspace = true } +nym-test-utils = { workspace = true } [lints] diff --git a/common/nym-kkt/src/lib.rs b/common/nym-kkt/src/lib.rs index b4100602b5..b7905160a6 100644 --- a/common/nym-kkt/src/lib.rs +++ b/common/nym-kkt/src/lib.rs @@ -16,10 +16,6 @@ pub use nym_kkt_context as context; #[cfg(test)] mod test { - use nym_kkt_ciphersuite::{Ciphersuite, HashFunction, HashLength, KEM, SignatureScheme}; - use rand09::RngCore; - use std::collections::BTreeMap; - use crate::keys::KEMKeys; use crate::{ initiator::KKTInitiator, @@ -29,9 +25,13 @@ mod test { }, responder::KKTResponder, }; + use nym_kkt_ciphersuite::{Ciphersuite, HashFunction, HashLength, KEM, SignatureScheme}; + use nym_test_utils::helpers::deterministic_rng_09; + use rand09::RngCore; + use std::collections::BTreeMap; #[test] - fn test_kkt_psq_e2e_encrypted_carrier() { + fn test_kkt_psq_e2e_one_way_encrypted_carrier() { let mut rng = rand09::rng(); let mut payload: Vec = vec![0u8; 900_000]; @@ -47,7 +47,6 @@ mod test { HashFunction::Shake256, ] { // generate kem public keys - let responder_mlkem_keypair = generate_keypair_mlkem(&mut rng); let responder_mceliece_keypair = generate_keypair_mceliece(&mut rng); @@ -64,20 +63,6 @@ mod test { HashLength::Default.value(), responder_kem.mc_eliece_encapsulation_key().as_ref(), ); - let initiator_mlkem_keypair = generate_keypair_mlkem(&mut rng); - let initiator_mceliece_keypair = generate_keypair_mceliece(&mut rng); - - let _i_dir_hash_mlkem = hash_encapsulation_key( - hash_function, - HashLength::Default.value(), - initiator_mlkem_keypair.public_key().as_slice(), - ); - - let _i_dir_hash_mceliece = hash_encapsulation_key( - hash_function, - HashLength::Default.value(), - initiator_mceliece_keypair.pk.as_ref(), - ); let init_hashes = BTreeMap::new(); @@ -128,41 +113,6 @@ mod test { responder_kem.ml_kem768_encapsulation_key().as_slice(), ) } - // Mutual - MlKem - { - let ciphersuite = Ciphersuite::resolve_ciphersuite( - KEM::MlKem768, - hash_function, - SignatureScheme::Ed25519, - None, - ) - .unwrap(); - let (mut initiator, request) = KKTInitiator::generate_one_way_request( - &mut rng, - ciphersuite, - &responder_x25519_keypair.pk, - &r_dir_hash_mlkem, - 1u8, - Some(payload.clone()), - ) - .unwrap(); - - let processed_request = responder.process_request(request, payload.len()).unwrap(); - - assert_eq!(processed_request.request_payload, payload); - - // if we keep unverified keys, this should change - assert!(processed_request.remote_encapsulation_key.is_none()); - - let processed_response = initiator - .process_response(processed_request.response, 0) - .unwrap(); - - assert_eq!( - processed_response.encapsulation_key.as_bytes(), - responder_kem.ml_kem768_encapsulation_key().as_slice(), - ) - } // OneWay - McEliece { @@ -195,7 +145,110 @@ mod test { responder_kem.mc_eliece_encapsulation_key().as_ref() ) } + } + } + + #[test] + fn test_kkt_psq_e2e_mutual_encrypted_carrier() { + let mut rng = deterministic_rng_09(); + + let mut payload: Vec = vec![0u8; 50000]; + rng.fill_bytes(&mut payload); + + // generate kem public keys + let initiator_mlkem_keypair = generate_keypair_mlkem(&mut rng); + let initiator_mceliece_keypair = generate_keypair_mceliece(&mut rng); + + let responder_mlkem_keypair = generate_keypair_mlkem(&mut rng); + let responder_mceliece_keypair = generate_keypair_mceliece(&mut rng); + + let responder_x25519_keypair = generate_lp_keypair_x25519(&mut rng); + + let initiator_kem = KEMKeys::new(initiator_mceliece_keypair, initiator_mlkem_keypair); + let responder_kem = KEMKeys::new(responder_mceliece_keypair, responder_mlkem_keypair); + + let init_hashes = initiator_kem.encapsulation_keys_digests(); + + let responder = KKTResponder::new( + &responder_x25519_keypair, + &responder_kem, + &init_hashes, + &[ + HashFunction::Blake3, + HashFunction::SHA256, + HashFunction::Shake128, + HashFunction::Shake256, + ], + &[SignatureScheme::Ed25519], + &[1], + ) + .unwrap(); + + for hash_function in [ + HashFunction::Blake3, + HashFunction::SHA256, + HashFunction::Shake128, + HashFunction::Shake256, + ] { + let r_dir_hash_mlkem = hash_encapsulation_key( + hash_function, + HashLength::Default.value(), + responder_kem.ml_kem768_encapsulation_key().as_slice(), + ); + + let r_dir_hash_mceliece = hash_encapsulation_key( + hash_function, + HashLength::Default.value(), + responder_kem.mc_eliece_encapsulation_key().as_ref(), + ); + // Mutual - MlKem + { + let ciphersuite = Ciphersuite::resolve_ciphersuite( + KEM::MlKem768, + hash_function, + SignatureScheme::Ed25519, + None, + ) + .unwrap(); + let (mut initiator, request) = KKTInitiator::generate_mutual_request( + &mut rng, + ciphersuite, + initiator_kem + .encoded_encapsulation_key(KEM::MlKem768) + .unwrap(), + &responder_x25519_keypair.pk, + &r_dir_hash_mlkem, + 1u8, + Some(payload.clone()), + ) + .unwrap(); + + let processed_request = responder.process_request(request, payload.len()).unwrap(); + + assert_eq!(processed_request.request_payload, payload); + assert_eq!( + processed_request + .remote_encapsulation_key + .unwrap() + .as_bytes(), + initiator_kem + .encapsulation_key(KEM::MlKem768) + .unwrap() + .as_bytes() + ); + + let processed_response = initiator + .process_response(processed_request.response, 0) + .unwrap(); + + assert_eq!( + processed_response.encapsulation_key.as_bytes(), + responder_kem.ml_kem768_encapsulation_key().as_slice(), + ) + } + + // Mutual - McEliece is not supported due to the key being too large { let ciphersuite = Ciphersuite::resolve_ciphersuite( KEM::McEliece, @@ -204,9 +257,12 @@ mod test { None, ) .unwrap(); - let (mut initiator, request) = KKTInitiator::generate_one_way_request( + let (mut initiator, request) = KKTInitiator::generate_mutual_request( &mut rng, ciphersuite, + initiator_kem + .encoded_encapsulation_key(KEM::McEliece) + .unwrap(), &responder_x25519_keypair.pk, &r_dir_hash_mceliece, 1u8, @@ -217,9 +273,16 @@ mod test { let processed_request = responder.process_request(request, payload.len()).unwrap(); assert_eq!(processed_request.request_payload, payload); - - // if we keep unverified keys, this should change - assert!(processed_request.remote_encapsulation_key.is_none()); + assert_eq!( + processed_request + .remote_encapsulation_key + .unwrap() + .as_bytes(), + initiator_kem + .encapsulation_key(KEM::McEliece) + .unwrap() + .as_bytes() + ); let processed_response = initiator .process_response(processed_request.response, 0) From 93ac638765579d3bf1ed5338d28ee9d85a871e1b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 6 Mar 2026 16:07:35 +0000 Subject: [PATCH 2/4] importing over changes from 'lp/persistent-node-connection' --- common/nym-lp/src/peer.rs | 4 + integration-tests/src/lp_registration.rs | 2 +- nym-node/Cargo.toml | 1 + nym-node/nym-node-metrics/src/network.rs | 44 +++- nym-node/src/node/lp/cleanup.rs | 9 +- nym-node/src/node/lp/control/egress/mod.rs | 99 +++++++++ .../{handler.rs => ingress/client_handler.rs} | 139 +++---------- .../src/node/lp/control/ingress/listener.rs | 192 ++++++++++++++++++ nym-node/src/node/lp/control/ingress/mod.rs | 6 + .../node/lp/control/ingress/node_handler.rs | 149 ++++++++++++++ nym-node/src/node/lp/control/listener.rs | 130 ------------ nym-node/src/node/lp/control/mod.rs | 118 ++++++++++- nym-node/src/node/lp/control/tests.rs | 89 ++++++++ nym-node/src/node/lp/directory.rs | 75 +++++++ nym-node/src/node/lp/error.rs | 16 +- nym-node/src/node/lp/mod.rs | 29 ++- nym-node/src/node/lp/registration.rs | 4 +- nym-node/src/node/lp/state.rs | 55 ++++- nym-node/src/node/mod.rs | 8 +- nym-node/src/node/shared_network.rs | 8 + nym-wallet/Cargo.lock | 49 +++-- 21 files changed, 938 insertions(+), 288 deletions(-) create mode 100644 nym-node/src/node/lp/control/egress/mod.rs rename nym-node/src/node/lp/control/{handler.rs => ingress/client_handler.rs} (87%) create mode 100644 nym-node/src/node/lp/control/ingress/listener.rs create mode 100644 nym-node/src/node/lp/control/ingress/mod.rs create mode 100644 nym-node/src/node/lp/control/ingress/node_handler.rs delete mode 100644 nym-node/src/node/lp/control/listener.rs create mode 100644 nym-node/src/node/lp/control/tests.rs create mode 100644 nym-node/src/node/lp/directory.rs diff --git a/common/nym-lp/src/peer.rs b/common/nym-lp/src/peer.rs index 3cfe665ee8..ce37065eef 100644 --- a/common/nym-lp/src/peer.rs +++ b/common/nym-lp/src/peer.rs @@ -143,6 +143,10 @@ impl LpRemotePeer { .ok_or(LpError::NoKnownKEMKeyDigests { kem, hash_function }) .cloned() } + + pub fn kem_key_digests(&self) -> &BTreeMap { + &self.expected_kem_key_digests + } } impl From for LpRemotePeer { diff --git a/integration-tests/src/lp_registration.rs b/integration-tests/src/lp_registration.rs index 592ad2dd01..ab9ef91ebc 100644 --- a/integration-tests/src/lp_registration.rs +++ b/integration-tests/src/lp_registration.rs @@ -17,7 +17,7 @@ mod tests { use nym_lp::peer::LpLocalPeer; use nym_node::config::{LpConfig, LpDebug}; use nym_node::node::GatewayStorage; - use nym_node::node::lp::control::handler::LpConnectionHandler; + use nym_node::node::lp::control::ingress::client_handler::LpConnectionHandler; use nym_node::node::lp::error::LpHandlerError; use nym_node::node::lp::{SharedLpControlState, SharedLpState}; use nym_node::wireguard::{PeerManager, PeerRegistrator}; diff --git a/nym-node/Cargo.toml b/nym-node/Cargo.toml index 28e7904007..336b9bb49b 100644 --- a/nym-node/Cargo.toml +++ b/nym-node/Cargo.toml @@ -139,6 +139,7 @@ harness = false cargo_metadata = { workspace = true } [dev-dependencies] +nym-lp = { workspace = true, features = ["mock"] } criterion = { workspace = true, features = ["async_tokio"] } nym-test-utils = { workspace = true } diff --git a/nym-node/nym-node-metrics/src/network.rs b/nym-node/nym-node-metrics/src/network.rs index 33fed5474b..120a38135b 100644 --- a/nym-node/nym-node-metrics/src/network.rs +++ b/nym-node/nym-node-metrics/src/network.rs @@ -16,7 +16,14 @@ pub struct NetworkStats { // the call stack active_egress_mixnet_connections: Arc, - active_lp_connections: AtomicUsize, + // incoming LP control connections from clients + active_lp_ingress_client_connections: AtomicUsize, + + // incoming LP control connections from nodes + active_lp_ingress_node_connections: AtomicUsize, + + // outgoing LP control connections to nodes + active_lp_egress_node_connections: AtomicUsize, } impl NetworkStats { @@ -59,15 +66,38 @@ impl NetworkStats { .load(Ordering::Relaxed) } - pub fn new_lp_connection(&self) { - self.active_lp_connections.fetch_add(1, Ordering::Relaxed); + pub fn new_ingress_lp_client_connection(&self) { + self.active_lp_ingress_client_connections + .fetch_add(1, Ordering::Relaxed); } - pub fn lp_connection_closed(&self) { - self.active_lp_connections.fetch_sub(1, Ordering::Relaxed); + pub fn closed_ingress_lp_client_connection(&self) { + self.active_lp_ingress_client_connections + .fetch_sub(1, Ordering::Relaxed); } - pub fn active_lp_connections_count(&self) -> usize { - self.active_lp_connections.load(Ordering::Relaxed) + pub fn new_ingress_lp_node_connection(&self) { + self.active_lp_ingress_node_connections + .fetch_add(1, Ordering::Relaxed); + } + + pub fn closed_ingress_lp_node_connection(&self) { + self.active_lp_ingress_node_connections + .fetch_sub(1, Ordering::Relaxed); + } + + pub fn new_egress_lp_node_connection(&self) { + self.active_lp_egress_node_connections + .fetch_add(1, Ordering::Relaxed); + } + + pub fn closed_egress_lp_node_connection(&self) { + self.active_lp_egress_node_connections + .fetch_sub(1, Ordering::Relaxed); + } + + pub fn active_lp_client_connections_count(&self) -> usize { + self.active_lp_ingress_client_connections + .load(Ordering::Relaxed) } } diff --git a/nym-node/src/node/lp/cleanup.rs b/nym-node/src/node/lp/cleanup.rs index 4b3ff23de6..2a5e5fa0a1 100644 --- a/nym-node/src/node/lp/cleanup.rs +++ b/nym-node/src/node/lp/cleanup.rs @@ -2,11 +2,10 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::config::LpDebug; -use dashmap::DashMap; +use crate::node::lp::state::ActiveLpSessions; use nym_lp::LpTransportSession; use nym_lp::peer_config::LpReceiverIndex; use nym_metrics::inc_by; -use std::sync::Arc; use std::time::Duration; use tracing::{debug, info}; @@ -74,14 +73,14 @@ impl TimestampedState { } pub(crate) struct CleanupTask { - session_states: Arc>>, + session_states: ActiveLpSessions, cfg: LpDebug, shutdown: nym_task::ShutdownToken, } impl CleanupTask { pub fn new( - session_states: Arc>>, + session_states: ActiveLpSessions, cfg: LpDebug, shutdown: nym_task::ShutdownToken, ) -> Self { @@ -100,7 +99,7 @@ impl CleanupTask { // Remove stale sessions (based on time since last activity) // Use shorter TTL for demoted (ReadOnlyTransport) sessions - self.session_states.retain(|_, timestamped| { + self.session_states.sessions.retain(|_, timestamped| { if timestamped.since_activity() > session_ttl { ss_removed += 1; false diff --git a/nym-node/src/node/lp/control/egress/mod.rs b/nym-node/src/node/lp/control/egress/mod.rs new file mode 100644 index 0000000000..c43f964ded --- /dev/null +++ b/nym-node/src/node/lp/control/egress/mod.rs @@ -0,0 +1,99 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::node::lp::control::LpConnectionStats; +use crate::node::lp::directory::LpNodeDetails; +use crate::node::lp::error::LpHandlerError; +use crate::node::lp::state::SharedLpNodeControlState; +use nym_lp::LpTransportSession; +use nym_lp::transport::LpHandshakeChannel; +use std::net::SocketAddr; +use tracing::debug; + +/// Initial connection handler for an egress LP node before completing the KKT/PSQ handshake. +pub struct InitialLpEgressNodeConnectionHandler { + stream: S, + remote_addr: SocketAddr, + responder_details: LpNodeDetails, + + state: SharedLpNodeControlState, + stats: LpConnectionStats, +} + +impl InitialLpEgressNodeConnectionHandler +where + S: LpHandshakeChannel + LpHandshakeChannel + Unpin, +{ + pub(crate) fn new( + stream: S, + remote_addr: SocketAddr, + responder_details: LpNodeDetails, + state: SharedLpNodeControlState, + ) -> Self { + Self { + stream, + remote_addr, + responder_details, + state, + stats: LpConnectionStats::new(), + } + } + + pub(crate) async fn complete_initial_handshake( + mut self, + ) -> Option> { + let remote = self.remote_addr; + + if self.responder_details.kem_key_hashes.is_empty() { + return Some(Err(LpHandlerError::MissingNodeKEMKeyHashes { + node_ip: self.remote_addr.ip(), + node_id: self.responder_details.node_id, + })); + } + + // 1. complete KKT/PSQ handshake before doing anything else. + // bail if it takes too long + let timeout = self.state.shared.lp_config.debug.handshake_ttl; + let stream = &mut self.stream; + + let handshake_state = match LpTransportSession::psq_handshake_initiator_mutual_internode( + stream, + self.state.local_lp_peer.clone(), + self.responder_details.to_lp_peer(), + self.responder_details.supported_protocol, + ) { + Ok(handshake_state) => handshake_state, + Err(err) => { + debug!("failed to initiate mutual KTT/PSQ handshake with {remote}: {err}"); + self.stats.emit_lifecycle_node_metrics(false); + return None; + } + }; + + let session = match tokio::time::timeout(timeout, handshake_state.complete_handshake()) + .await + { + Err(_timeout) => { + debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}"); + self.stats.emit_lifecycle_node_metrics(false); + return None; + } + Ok(Err(handshake_failure)) => { + debug!( + "failed to complete mutual KKT/PSQ handshake with {remote}: {handshake_failure}" + ); + self.stats.emit_lifecycle_node_metrics(false); + return None; + } + Ok(Ok(session)) => session, + }; + + debug!( + "completed egress KKT/PSQ handshake with node {}: {remote}", + self.responder_details.node_id + ); + + // TODO: change return type into complete handler + Some(Ok(session)) + } +} diff --git a/nym-node/src/node/lp/control/handler.rs b/nym-node/src/node/lp/control/ingress/client_handler.rs similarity index 87% rename from nym-node/src/node/lp/control/handler.rs rename to nym-node/src/node/lp/control/ingress/client_handler.rs index 8d5ceb0249..b8dd26a36f 100644 --- a/nym-node/src/node/lp/control/handler.rs +++ b/nym-node/src/node/lp/control/ingress/client_handler.rs @@ -2,8 +2,9 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::node::lp::cleanup::TimestampedState; +use crate::node::lp::control::{LP_DURATION_BUCKETS, LpConnectionStats}; use crate::node::lp::error::LpHandlerError; -use crate::node::lp::state::SharedLpControlState; +use crate::node::lp::state::SharedLpClientControlState; use dashmap::mapref::one::RefMut; use nym_lp::packet::message::LpMessageType; use nym_lp::packet::{EncryptedLpPacket, ForwardPacketData, LpMessage}; @@ -13,6 +14,7 @@ use nym_lp::transport::LpHandshakeChannel; use nym_lp::transport::traits::LpTransportChannel; use nym_lp::{LpTransportSession, packet::message::ExpectedResponseSize}; use nym_metrics::{add_histogram_obs, inc, inc_by}; +use nym_node_metrics::NymNodeMetrics; use nym_registration_common::{LpRegistrationRequest, RegistrationStatus}; use std::net::SocketAddr; use std::time::Duration; @@ -20,66 +22,15 @@ use tokio::net::TcpStream; use tokio::time::timeout; use tracing::*; -// Histogram buckets for LP operation duration (legacy - used by unused forwarding methods) -const LP_DURATION_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]; - // Timeout for forward I/O operations (send + receive on exit stream) // Must be long enough to cover exit gateway processing time const FORWARD_IO_TIMEOUT_SECS: u64 = 30; -// Histogram buckets for LP connection lifecycle duration -// LP connections can be very short (registration only: ~1s) or very long (dVPN sessions: hours/days) -// Covers full range from seconds to 24 hours -const LP_CONNECTION_DURATION_BUCKETS: &[f64] = &[ - 1.0, // 1 second - 5.0, // 5 seconds - 10.0, // 10 seconds - 30.0, // 30 seconds - 60.0, // 1 minute - 300.0, // 5 minutes - 600.0, // 10 minutes - 1800.0, // 30 minutes - 3600.0, // 1 hour - 7200.0, // 2 hours - 14400.0, // 4 hours - 28800.0, // 8 hours - 43200.0, // 12 hours - 86400.0, // 24 hours -]; - -/// Connection lifecycle statistics tracking -struct ConnectionStats { - /// When the connection started - start_time: std::time::Instant, - /// Total bytes received (including protocol framing) - bytes_received: u64, - /// Total bytes sent (including protocol framing) - bytes_sent: u64, -} - -impl ConnectionStats { - fn new() -> Self { - Self { - start_time: std::time::Instant::now(), - bytes_received: 0, - bytes_sent: 0, - } - } - - fn record_bytes_received(&mut self, bytes: usize) { - self.bytes_received += bytes as u64; - } - - fn record_bytes_sent(&mut self, bytes: usize) { - self.bytes_sent += bytes as u64; - } -} - -pub struct LpConnectionHandler { +pub struct LpClientConnectionHandler { stream: S, remote_addr: SocketAddr, - state: SharedLpControlState, - stats: ConnectionStats, + state: SharedLpClientControlState, + stats: LpConnectionStats, // /// Flag indicating whether this is a connection from an entry gateway serving as a proxy // forwarded_connection: bool, @@ -94,27 +45,25 @@ pub struct LpConnectionHandler { exit_stream: Option<(S, SocketAddr)>, } -impl LpConnectionHandler +impl LpClientConnectionHandler where S: LpTransportChannel + LpHandshakeChannel + Unpin, { - pub fn new( - stream: S, - // forwarded_connection: bool, - remote_addr: SocketAddr, - state: SharedLpControlState, - ) -> Self { + pub fn new(stream: S, remote_addr: SocketAddr, state: SharedLpClientControlState) -> Self { Self { stream, remote_addr, - // forwarded_connection, state, - stats: ConnectionStats::new(), + stats: LpConnectionStats::new(), bound_receiver_idx: None, exit_stream: None, } } + pub(crate) fn metrics(&self) -> &NymNodeMetrics { + &self.state.shared.metrics + } + /// Get the mutable reference to the state machine associated with this client. /// It is vital it's never held across await points or this might lead to a deadlock. fn state_entry_mut( @@ -125,8 +74,7 @@ where self.state .shared .session_states - .get_mut(&receiver_index) - .ok_or_else(|| LpHandlerError::MissingLpSession { receiver_index }) + .get_state_entry_mut(receiver_index) } /// AIDEV-NOTE: Stream-oriented packet loop @@ -135,11 +83,12 @@ where /// First packet binds the connection to a receiver_idx (session-affine). /// Binding is set by handle_client_hello() from payload's receiver_index, /// or by validate_or_set_binding() for non-bootstrap first packets. - pub async fn handle(mut self) -> Result<(), LpHandlerError> { - debug!("Handling LP connection from {}", self.remote_addr); + pub async fn handle(&mut self) -> Result<(), LpHandlerError> { + let remote = self.remote_addr; + debug!("Handling LP connection from {remote}"); // Track total LP connections handled - inc!("lp_connections_total"); + inc!("lp_client_connections_total"); // ============================================================ // STREAM-ORIENTED PROCESSING: Loop until connection closes @@ -160,18 +109,12 @@ where .await { Err(_timeout) => { - debug!( - "timed out attempting to complete KTT/PSQ handshake with {}", - self.remote_addr - ); + debug!("timed out attempting to complete KTT/PSQ handshake with {remote}"); self.emit_lifecycle_metrics(false); return Ok(()); } Ok(Err(handshake_failure)) => { - debug!( - "failed to complete KKT/PSQ handshake with {}: {handshake_failure}", - self.remote_addr - ); + debug!("failed to complete KKT/PSQ handshake with {remote}: {handshake_failure}",); self.emit_lifecycle_metrics(false); return Ok(()); } @@ -180,10 +123,7 @@ where let receiver_idx = session.receiver_index(); // 2. insert the state machine into the shared state - self.state - .shared - .session_states - .insert(receiver_idx, TimestampedState::new(session)); + self.state.shared.session_states.insert_new_session(session); self.bound_receiver_idx = Some(receiver_idx); // 3. handle any new incoming packet @@ -194,7 +134,7 @@ where Err(err) => { if err.is_connection_closed() { // Graceful EOF - client closed connection - trace!("Connection closed by {} (EOF)", self.remote_addr); + trace!("Connection closed by {remote} (EOF)"); break; } else { inc!("lp_errors_receive_packet"); @@ -625,30 +565,7 @@ where /// Emit connection lifecycle metrics fn emit_lifecycle_metrics(&self, graceful: bool) { - // Track connection duration - let duration = self.stats.start_time.elapsed().as_secs_f64(); - add_histogram_obs!( - "lp_connection_duration_seconds", - duration, - LP_CONNECTION_DURATION_BUCKETS - ); - - // Track bytes transferred - inc_by!( - "lp_connection_bytes_received_total", - self.stats.bytes_received as i64 - ); - inc_by!( - "lp_connection_bytes_sent_total", - self.stats.bytes_sent as i64 - ); - - // Track completion type - if graceful { - inc!("lp_connections_completed_gracefully"); - } else { - inc!("lp_connections_completed_with_error"); - } + self.stats.emit_lifecycle_client_metrics(graceful); } } @@ -657,7 +574,7 @@ mod tests { use super::*; use crate::config::LpConfig; use crate::config::lp::LpDebug; - use crate::node::lp::state::SharedLpState; + use crate::node::lp::state::{ActiveLpSessions, SharedLpState}; use nym_lp::peer::{KEMKeys, LpLocalPeer, generate_keypair_mceliece, generate_keypair_mlkem}; use nym_lp::{Ciphersuite, SessionManager, sessions_for_tests}; use nym_test_utils::helpers::{deterministic_rng, deterministic_rng_09}; @@ -665,7 +582,7 @@ mod tests { // ==================== Test Helpers ==================== /// Create a minimal test state for handler tests - async fn create_minimal_test_state() -> SharedLpControlState { + async fn create_minimal_test_state() -> SharedLpClientControlState { use nym_crypto::asymmetric::ed25519; let mut rng = deterministic_rng(); @@ -690,14 +607,14 @@ mod tests { ); let lp_peer = LpLocalPeer::new(Ciphersuite::default(), x_keys).with_kem_keys(kem_keys); - SharedLpControlState { + SharedLpClientControlState { local_lp_peer: lp_peer, peer_registrator: None, forward_semaphore, shared: SharedLpState { lp_config, metrics: nym_node_metrics::NymNodeMetrics::default(), - session_states: Arc::new(dashmap::DashMap::new()), + session_states: ActiveLpSessions::new(), }, } } @@ -724,7 +641,7 @@ mod tests { let server_task = tokio::spawn(async move { let (stream, remote_addr) = listener.accept().await.unwrap(); let state = create_minimal_test_state().await; - let mut handler = LpConnectionHandler::new(stream, remote_addr, state); + let mut handler = LpClientConnectionHandler::new(stream, remote_addr, state); // Two-phase: receive raw bytes + header, then parse full packet let packet = handler.receive_raw_packet().await?; let header = packet.outer_header(); diff --git a/nym-node/src/node/lp/control/ingress/listener.rs b/nym-node/src/node/lp/control/ingress/listener.rs new file mode 100644 index 0000000000..34a09fba77 --- /dev/null +++ b/nym-node/src/node/lp/control/ingress/listener.rs @@ -0,0 +1,192 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::config::LpConfig; +use crate::error::NymNodeError; +use crate::node::lp::control::ingress::client_handler::LpClientConnectionHandler; +use crate::node::lp::control::ingress::node_handler::InitialLpIngressNodeConnectionHandler; +use crate::node::lp::directory::LpNodeDetails; +use crate::node::lp::state::{SharedLpClientControlState, SharedLpNodeControlState}; +use nym_task::ShutdownTracker; +use std::net::SocketAddr; +use tokio::net::TcpListener; +use tracing::{debug, error, info, trace, warn}; + +/// LP listener that accepts TCP connections on port 41264 +pub struct LpControlListener { + /// Address to bind to + bind_address: SocketAddr, + + /// Shared state for clients connection handlers + clients_handler_state: SharedLpClientControlState, + + /// Shared state for nodes connection handlers + nodes_handler_state: SharedLpNodeControlState, + + /// Shutdown coordination + shutdown: ShutdownTracker, +} + +impl LpControlListener { + pub fn new( + bind_address: SocketAddr, + clients_handler_state: SharedLpClientControlState, + nodes_handler_state: SharedLpNodeControlState, + shutdown: ShutdownTracker, + ) -> Self { + Self { + bind_address, + clients_handler_state, + nodes_handler_state, + shutdown, + } + } + + fn lp_config(&self) -> LpConfig { + self.clients_handler_state.shared.lp_config + } + + pub async fn run(&mut self) -> Result<(), NymNodeError> { + let bind_address = self.bind_address; + info!("Starting LP control listener on {bind_address}"); + + let listener = TcpListener::bind(bind_address).await.map_err(|source| { + error!("Failed to bind LP listener to {bind_address}: {source}",); + NymNodeError::LpBindFailure { + address: bind_address, + source, + } + })?; + + let shutdown_token = self.shutdown.clone_shutdown_token(); + + loop { + tokio::select! { + biased; + _ = shutdown_token.cancelled() => { + trace!("LP listener: received shutdown signal"); + break; + } + + result = listener.accept() => { + match result { + Ok((stream, addr)) => self.handle_connection(stream, addr), + Err(e) => warn!("Failed to accept LP connection: {e}") + } + } + } + } + + info!("LP listener shutdown complete"); + Ok(()) + } + + fn handle_node_connection( + &self, + stream: tokio::net::TcpStream, + remote_addr: SocketAddr, + initiator_details: LpNodeDetails, + ) { + warn!( + "received an incoming LP node connection - we can't support this yet beyond the handshake!" + ); + + debug!("Accepting LP node connection from {remote_addr}"); + + // Spawn handler task + let mut handler = InitialLpIngressNodeConnectionHandler::new( + stream, + remote_addr, + initiator_details, + self.nodes_handler_state.clone(), + ); + + self.shutdown.try_spawn_named_with_shutdown( + async move { + let metrics = handler.metrics().clone(); + + // Increment connection counter + metrics.network.new_ingress_lp_node_connection(); + + let result = handler.handle().await; + + // Decrement connection counter + metrics.network.closed_ingress_lp_node_connection(); + + // Handler emits lifecycle metrics internally on success + // For errors, we need to emit them here since handler is consumed + if let Err(e) = result { + warn!("LP node handler error for {remote_addr}: {e}"); + // Note: metrics are emitted in handle() for graceful path + // On error path, handle() returns early without emitting + // So we track errors here + } + }, + &format!("LP_NODE::{remote_addr}"), + ); + } + + fn handle_client_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) { + // Check connection limit (only for clients, nodes must always be allowed regardless of the limit) + let active_connections = self.active_client_connections(); + let max_connections = self.lp_config().debug.max_connections; + if active_connections >= max_connections { + warn!( + "LP connection limit exceeded ({active_connections}/{max_connections}), rejecting connection from {remote_addr}" + ); + return; + } + + debug!( + "Accepting LP client connection from {remote_addr} ({active_connections} active connections)" + ); + + // Spawn handler task + let mut handler = + LpClientConnectionHandler::new(stream, remote_addr, self.clients_handler_state.clone()); + + self.shutdown.try_spawn_named_with_shutdown( + async move { + // Increment connection counter + handler.metrics().network.new_ingress_lp_client_connection(); + + let result = handler.handle().await; + // Decrement connection counter + handler + .metrics() + .network + .closed_ingress_lp_client_connection(); + + // Handler emits lifecycle metrics internally on success + // For errors, we need to emit them here since handler is consumed + if let Err(e) = result { + warn!("LP client handler error for {remote_addr}: {e}"); + // Note: metrics are emitted in handle() for graceful path + // On error path, handle() returns early without emitting + // So we track errors here + } + }, + &format!("LP_CLIENT::{remote_addr}"), + ); + } + + fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) { + if let Some(initiator_details) = self + .nodes_handler_state + .nodes + .get_node_details(remote_addr.ip()) + { + self.handle_node_connection(stream, remote_addr, initiator_details); + } else { + self.handle_client_connection(stream, remote_addr); + } + } + + fn active_client_connections(&self) -> usize { + self.clients_handler_state + .shared + .metrics + .network + .active_lp_client_connections_count() + } +} diff --git a/nym-node/src/node/lp/control/ingress/mod.rs b/nym-node/src/node/lp/control/ingress/mod.rs new file mode 100644 index 0000000000..79a7f15682 --- /dev/null +++ b/nym-node/src/node/lp/control/ingress/mod.rs @@ -0,0 +1,6 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +pub mod client_handler; +pub(crate) mod listener; +mod node_handler; diff --git a/nym-node/src/node/lp/control/ingress/node_handler.rs b/nym-node/src/node/lp/control/ingress/node_handler.rs new file mode 100644 index 0000000000..2ade110c63 --- /dev/null +++ b/nym-node/src/node/lp/control/ingress/node_handler.rs @@ -0,0 +1,149 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use crate::node::lp::control::LpConnectionStats; +use crate::node::lp::directory::LpNodeDetails; +use crate::node::lp::error::LpHandlerError; +use crate::node::lp::state::SharedLpNodeControlState; +use nym_lp::LpTransportSession; +use nym_lp::transport::{LpHandshakeChannel, LpTransportChannel}; +use nym_metrics::inc; +use nym_node_metrics::NymNodeMetrics; +use nym_topology::NodeId; +use std::net::SocketAddr; +use tokio::net::TcpStream; +use tracing::debug; + +/// Initial connection handler for an ingress LP node before completing the KKT/PSQ handshake. +pub struct InitialLpIngressNodeConnectionHandler { + stream: S, + remote_addr: SocketAddr, + initiator_details: LpNodeDetails, + + state: SharedLpNodeControlState, + stats: LpConnectionStats, +} + +impl InitialLpIngressNodeConnectionHandler +where + S: LpHandshakeChannel + LpTransportChannel + Unpin, +{ + pub fn new( + stream: S, + remote_addr: SocketAddr, + initiator_details: LpNodeDetails, + state: SharedLpNodeControlState, + ) -> Self { + Self { + stream, + remote_addr, + initiator_details, + state, + stats: LpConnectionStats::new(), + } + } + + pub(crate) fn metrics(&self) -> &NymNodeMetrics { + &self.state.shared.metrics + } + + pub(crate) async fn complete_initial_handshake( + mut self, + ) -> Option, LpHandlerError>> { + let remote = self.remote_addr; + + if self.initiator_details.kem_key_hashes.is_empty() { + return Some(Err(LpHandlerError::MissingNodeKEMKeyHashes { + node_ip: self.remote_addr.ip(), + node_id: self.initiator_details.node_id, + })); + } + + // 1. complete KKT/PSQ handshake before doing anything else. + // bail if it takes too long + let timeout = self.state.shared.lp_config.debug.handshake_ttl; + let local_peer = self.state.local_lp_peer.clone(); + let stream = &mut self.stream; + let kem_hashes = self.initiator_details.kem_key_hashes.clone(); + + let session = match tokio::time::timeout(timeout, async move { + LpTransportSession::psq_handshake_responder_mutual(stream, local_peer, kem_hashes) + .complete_handshake() + .await + }) + .await + { + Err(_timeout) => { + debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}"); + self.stats.emit_lifecycle_node_metrics(false); + return None; + } + Ok(Err(handshake_failure)) => { + debug!( + "failed to complete mutual KKT/PSQ handshake with {remote}: {handshake_failure}" + ); + self.stats.emit_lifecycle_node_metrics(false); + return None; + } + Ok(Ok(session)) => session, + }; + + debug!( + "completed ingress KKT/PSQ handshake with node {}: {remote}", + self.initiator_details.node_id + ); + + Some(Ok(LpIngressNodeConnectionHandler { + stream: self.stream, + remote_addr: remote, + remote_node_id: self.initiator_details.node_id, + state: self.state, + stats: self.stats, + transport_session: session, + })) + } + + pub async fn handle(mut self) -> Result<(), LpHandlerError> { + // Track total LP connections handled + inc!("lp_node_connections_total"); + + // attempt to complete initial handshake + let upgraded_handler = match self.complete_initial_handshake().await { + None => return Ok(()), + Some(handler_res) => handler_res?, + }; + + // continue handling the requests with the transport session + upgraded_handler.handle().await + } +} + +/// Connection handler for an LP node after completing the KKT/PSQ handshake. +pub struct LpIngressNodeConnectionHandler { + stream: S, + remote_addr: SocketAddr, + remote_node_id: NodeId, + + state: SharedLpNodeControlState, + stats: LpConnectionStats, + transport_session: LpTransportSession, + // TODO [still WIP]: + // LOCAL receiver index to stream id + // client_streams: HashMap, +} + +impl LpIngressNodeConnectionHandler +where + S: LpHandshakeChannel + LpTransportChannel + Unpin, +{ + async fn handle(mut self) -> Result<(), LpHandlerError> { + // handle all the forwarding here + + self.stats.emit_lifecycle_node_metrics(true); + Ok(()) + } + + pub(crate) fn transport_session(&self) -> &LpTransportSession { + &self.transport_session + } +} diff --git a/nym-node/src/node/lp/control/listener.rs b/nym-node/src/node/lp/control/listener.rs deleted file mode 100644 index 742f526fe8..0000000000 --- a/nym-node/src/node/lp/control/listener.rs +++ /dev/null @@ -1,130 +0,0 @@ -// Copyright 2026 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only - -use crate::config::LpConfig; -use crate::error::NymNodeError; -use crate::node::lp::control::handler::LpConnectionHandler; -use crate::node::lp::state::SharedLpControlState; -use nym_task::ShutdownTracker; -use std::net::SocketAddr; -use tokio::net::TcpListener; -use tracing::{debug, error, info, trace, warn}; - -/// LP listener that accepts TCP connections on port 41264 -pub struct LpControlListener { - /// Address to bind to - bind_address: SocketAddr, - - /// Shared state for connection handlers - handler_state: SharedLpControlState, - - /// Shutdown coordination - shutdown: ShutdownTracker, -} - -impl LpControlListener { - pub fn new( - bind_address: SocketAddr, - handler_state: SharedLpControlState, - shutdown: ShutdownTracker, - ) -> Self { - Self { - bind_address, - handler_state, - shutdown, - } - } - - fn lp_config(&self) -> LpConfig { - self.handler_state.shared.lp_config - } - - pub async fn run(&mut self) -> Result<(), NymNodeError> { - let bind_address = self.bind_address; - info!("Starting LP control listener on {bind_address}"); - - let listener = TcpListener::bind(bind_address).await.map_err(|source| { - error!("Failed to bind LP listener to {bind_address}: {source}",); - NymNodeError::LpBindFailure { - address: bind_address, - source, - } - })?; - - let shutdown_token = self.shutdown.clone_shutdown_token(); - - loop { - tokio::select! { - biased; - _ = shutdown_token.cancelled() => { - trace!("LP listener: received shutdown signal"); - break; - } - - result = listener.accept() => { - match result { - Ok((stream, addr)) => self.handle_connection(stream, addr), - Err(e) => warn!("Failed to accept LP connection: {e}") - } - } - } - } - - info!("LP listener shutdown complete"); - Ok(()) - } - - fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) { - // Check connection limit - let active_connections = self.active_lp_connections(); - let max_connections = self.lp_config().debug.max_connections; - if active_connections >= max_connections { - warn!( - "LP connection limit exceeded ({active_connections}/{max_connections}), rejecting connection from {remote_addr}" - ); - return; - } - - debug!( - "Accepting LP connection from {remote_addr} ({active_connections} active connections)" - ); - - // Increment connection counter - self.handler_state - .shared - .metrics - .network - .new_lp_connection(); - - // Spawn handler task - let handler = LpConnectionHandler::new(stream, remote_addr, self.handler_state.clone()); - - let metrics = self.handler_state.shared.metrics.clone(); - self.shutdown.try_spawn_named_with_shutdown( - async move { - let result = handler.handle().await; - - // Handler emits lifecycle metrics internally on success - // For errors, we need to emit them here since handler is consumed - if let Err(e) = result { - warn!("LP handler error for {remote_addr}: {e}"); - // Note: metrics are emitted in handle() for graceful path - // On error path, handle() returns early without emitting - // So we track errors here - } - - // Decrement connection counter on exit - metrics.network.lp_connection_closed(); - }, - &format!("LP::{remote_addr}"), - ); - } - - fn active_lp_connections(&self) -> usize { - self.handler_state - .shared - .metrics - .network - .active_lp_connections_count() - } -} diff --git a/nym-node/src/node/lp/control/mod.rs b/nym-node/src/node/lp/control/mod.rs index 5453ecddf5..a51c6bacd9 100644 --- a/nym-node/src/node/lp/control/mod.rs +++ b/nym-node/src/node/lp/control/mod.rs @@ -1,5 +1,119 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -pub mod handler; -pub(crate) mod listener; +use nym_metrics::{add_histogram_obs, inc, inc_by}; + +pub mod egress; +pub mod ingress; +mod tests; + +// Histogram buckets for LP operation duration (legacy - used by unused forwarding methods) +const LP_DURATION_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]; + +// Histogram buckets for LP connection lifecycle duration +// LP connections can be very short (registration only: ~1s) or very long (dVPN sessions: hours/days) +// Covers full range from seconds to 24 hours +const LP_CONNECTION_DURATION_BUCKETS: &[f64] = &[ + 1.0, // 1 second + 5.0, // 5 seconds + 10.0, // 10 seconds + 30.0, // 30 seconds + 60.0, // 1 minute + 300.0, // 5 minutes + 600.0, // 10 minutes + 1800.0, // 30 minutes + 3600.0, // 1 hour + 7200.0, // 2 hours + 14400.0, // 4 hours + 28800.0, // 8 hours + 43200.0, // 12 hours + 86400.0, // 24 hours +]; + +/// Connection lifecycle statistics tracking +pub(crate) struct LpConnectionStats { + /// When the connection started + start_time: std::time::Instant, + /// Total bytes received (including protocol framing) + bytes_received: u64, + /// Total bytes sent (including protocol framing) + bytes_sent: u64, +} + +impl LpConnectionStats { + fn new() -> Self { + Self { + start_time: std::time::Instant::now(), + bytes_received: 0, + bytes_sent: 0, + } + } + + fn duration(&self) -> std::time::Duration { + self.start_time.elapsed() + } + + fn record_bytes_received(&mut self, bytes: usize) { + self.bytes_received += bytes as u64; + } + + fn record_bytes_sent(&mut self, bytes: usize) { + self.bytes_sent += bytes as u64; + } + + /// Emit connection lifecycle metrics for a client connection + fn emit_lifecycle_client_metrics(&self, graceful: bool) { + // Track connection duration + let duration = self.duration().as_secs_f64(); + add_histogram_obs!( + "lp_client_connection_duration_seconds", + duration, + LP_CONNECTION_DURATION_BUCKETS + ); + + // Track bytes transferred + inc_by!( + "lp_client_connection_bytes_received_total", + self.bytes_received as i64 + ); + inc_by!( + "lp_client_connection_bytes_sent_total", + self.bytes_sent as i64 + ); + + // Track completion type + if graceful { + inc!("lp_client_connections_completed_gracefully"); + } else { + inc!("lp_client_connections_completed_with_error"); + } + } + + /// Emit connection lifecycle metrics for a node connection + fn emit_lifecycle_node_metrics(&self, graceful: bool) { + // Track connection duration + let duration = self.duration().as_secs_f64(); + add_histogram_obs!( + "lp_node_connection_duration_seconds", + duration, + LP_CONNECTION_DURATION_BUCKETS + ); + + // Track bytes transferred + inc_by!( + "lp_node_connection_bytes_received_total", + self.bytes_received as i64 + ); + inc_by!( + "lp_node_connection_bytes_sent_total", + self.bytes_sent as i64 + ); + + // Track completion type + if graceful { + inc!("lp_node_connections_completed_gracefully"); + } else { + inc!("lp_node_connections_completed_with_error"); + } + } +} diff --git a/nym-node/src/node/lp/control/tests.rs b/nym-node/src/node/lp/control/tests.rs new file mode 100644 index 0000000000..2270e87e58 --- /dev/null +++ b/nym-node/src/node/lp/control/tests.rs @@ -0,0 +1,89 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +#[cfg(test)] +mod tests { + use crate::node::lp::SharedLpState; + use crate::node::lp::control::egress::connection::InitialLpEgressNodeConnectionHandler; + use crate::node::lp::control::ingress::node_handler::InitialLpIngressNodeConnectionHandler; + use crate::node::lp::directory::LpNodeDetails; + use crate::node::lp::state::SharedLpNodeControlState; + use anyhow::Context; + use nym_lp::packet::version; + use nym_lp::peer::{LpLocalPeer, LpRemotePeer, mock_peers}; + use nym_test_utils::helpers::seeded_rng; + use nym_test_utils::mocks::async_read_write::MockIOStream; + use nym_test_utils::traits::TimeboxedSpawnable; + use rand::RngCore; + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + + fn shared_node_state(peer: LpLocalPeer) -> SharedLpNodeControlState { + SharedLpNodeControlState { + local_lp_peer: peer, + nodes: Default::default(), + shared: SharedLpState { + metrics: Default::default(), + lp_config: Default::default(), + session_states: Default::default(), + }, + } + } + + fn lp_node_details(peer: LpRemotePeer) -> LpNodeDetails { + let key_bytes = peer.x25519().as_ref().try_into().unwrap(); + let mut rng = seeded_rng(key_bytes); + LpNodeDetails::new( + rng.next_u32(), + peer.kem_key_digests().clone(), + peer.x25519().clone(), + version::CURRENT, + ) + } + + #[tokio::test] + async fn basic_node_to_node_handshake() -> anyhow::Result<()> { + nym_test_utils::helpers::setup_test_logger(); + + let (init, resp) = mock_peers(); + let init_remote = init.as_remote(); + let resp_remote = resp.as_remote(); + + let conn_init = MockIOStream::default(); + let conn_resp = conn_init.try_get_remote_handle(); + + let init_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 1234); + let init_details = lp_node_details(init_remote); + let resp_details = lp_node_details(resp_remote); + + let init_state = shared_node_state(init); + let resp_state = shared_node_state(resp); + + let init_handler = InitialLpEgressNodeConnectionHandler::new( + conn_init, + init_addr, + resp_details, + init_state, + ); + + let resp_handler = InitialLpIngressNodeConnectionHandler::new( + conn_resp, + init_addr, + init_details, + resp_state, + ); + + let init_future = init_handler.complete_initial_handshake().spawn_timeboxed(); + let resp_future = resp_handler.complete_initial_handshake().spawn_timeboxed(); + + let (init_result, resp_result) = tokio::join!(init_future, resp_future); + let init_result = init_result??.context("handshake failure")??; + let resp_result = resp_result??.context("handshake failure")??; + + assert_eq!( + init_result.receiver_index(), + resp_result.transport_session().receiver_index() + ); + + Ok(()) + } +} diff --git a/nym-node/src/node/lp/directory.rs b/nym-node/src/node/lp/directory.rs new file mode 100644 index 0000000000..78e16357a2 --- /dev/null +++ b/nym-node/src/node/lp/directory.rs @@ -0,0 +1,75 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +use arc_swap::ArcSwap; +use nym_lp::peer::{DHPublicKey, LpRemotePeer}; +use nym_lp::{KEM, KEMKeyDigests}; +use nym_topology::NodeId; +use std::collections::{BTreeMap, HashMap}; +use std::net::IpAddr; +use std::ops::Deref; +use std::sync::Arc; + +/// Wrapper around all known LP nodes +#[derive(Clone, Default)] +pub struct LpNodes { + // map between all available ip addresses of other nodes and their details + nodes: Arc>>, +} + +impl LpNodes { + pub(crate) fn get_node_details(&self, node_ip: IpAddr) -> Option { + self.nodes.load().get(&node_ip).cloned() + } + + pub(crate) fn get_node_id(&self, node_ip: IpAddr) -> Option { + self.nodes + .load() + .get(&node_ip) + .map(|details| details.node_id) + } +} + +#[derive(Clone)] +pub(crate) struct LpNodeDetails { + inner: Arc, +} + +impl LpNodeDetails { + pub(crate) fn new( + node_id: NodeId, + kem_key_hashes: BTreeMap, + x25519: DHPublicKey, + supported_protocol: u8, + ) -> Self { + LpNodeDetails { + inner: Arc::new(LpNodeDetailsInner { + node_id, + kem_key_hashes, + x25519, + supported_protocol, + }), + } + } +} + +impl Deref for LpNodeDetails { + type Target = LpNodeDetailsInner; + + fn deref(&self) -> &Self::Target { + self.inner.deref() + } +} + +pub(crate) struct LpNodeDetailsInner { + pub(crate) node_id: NodeId, + pub(crate) kem_key_hashes: BTreeMap, + pub(crate) x25519: DHPublicKey, + pub(crate) supported_protocol: u8, +} + +impl LpNodeDetailsInner { + pub(crate) fn to_lp_peer(&self) -> LpRemotePeer { + LpRemotePeer::new(self.x25519).with_key_digests(self.kem_key_hashes.clone()) + } +} diff --git a/nym-node/src/node/lp/error.rs b/nym-node/src/node/lp/error.rs index 8952e9f680..865d4ccaab 100644 --- a/nym-node/src/node/lp/error.rs +++ b/nym-node/src/node/lp/error.rs @@ -6,7 +6,8 @@ use nym_lp::peer_config::LpReceiverIndex; use nym_lp::session::LpAction; use nym_lp::transport::LpTransportError; use nym_lp::{LpError, packet::MalformedLpPacketError}; -use std::net::SocketAddr; +use nym_topology::NodeId; +use std::net::{IpAddr, SocketAddr}; use thiserror::Error; #[derive(Debug, Error)] @@ -47,9 +48,18 @@ pub enum LpHandlerError { #[error("timed out while attempting to send to/receive from the connection")] ConnectionTimeout, + #[error("missing KEM key hashes for node {node_id} connected from {node_ip}")] + MissingNodeKEMKeyHashes { node_ip: IpAddr, node_id: NodeId }, + #[error("data channel is not yet implemented")] UnimplementedDataChannel, + #[error("{ip_addr} does not correspond to any known LP node")] + NotLpNode { ip_addr: IpAddr }, + + #[error("{0}")] + Internal(String), + #[error("{0}")] Other(String), } @@ -64,6 +74,10 @@ impl LpHandlerError { } } + pub fn internal(msg: impl Into) -> Self { + LpHandlerError::Internal(msg.into()) + } + pub fn other(msg: impl Into) -> Self { LpHandlerError::Other(msg.into()) } diff --git a/nym-node/src/node/lp/mod.rs b/nym-node/src/node/lp/mod.rs index cd550cbb19..8bc14fab6e 100644 --- a/nym-node/src/node/lp/mod.rs +++ b/nym-node/src/node/lp/mod.rs @@ -9,7 +9,7 @@ // ## Connection Metrics (via NetworkStats in nym-node-metrics) // - active_lp_connections: Gauge tracking current active LP connections (incremented on accept, decremented on close) // -// ## Handler Metrics (in handler.rs) +// ## Handler Metrics (in client_handler) // - lp_connections_total: Counter for total LP connections handled // - lp_client_hello_failed: Counter for ClientHello failures (timestamp validation, protocol errors) // - lp_handshakes_success: Counter for successful handshake completions @@ -46,7 +46,7 @@ // ## Error Categorization Metrics // - lp_errors_wg_peer_registration: Counter for WireGuard peer registration failures // -// ## Connection Lifecycle Metrics (in handler.rs) +// ## Connection Lifecycle Metrics (in client_handler) // - lp_connection_duration_seconds: Histogram of connection duration from start to end (buckets: 1s to 24h) // - lp_connection_bytes_received_total: Counter for total bytes received including protocol framing // - lp_connection_bytes_sent_total: Counter for total bytes sent including protocol framing @@ -58,7 +58,7 @@ // - lp_states_cleanup_session_removed: Counter for stale sessions removed by cleanup task // - lp_states_cleanup_demoted_removed: Counter for demoted (read-only) sessions removed by cleanup task // -// ## Subsession/Rekeying Metrics (in handler.rs) +// ## Subsession/Rekeying Metrics (in client_handler) // - lp_subsession_kk2_sent: Counter for SubsessionKK2 responses sent (indicates client initiated rekeying) // - lp_subsession_complete: Counter for successful subsession promotions // - lp_subsession_receiver_index_collision: Counter for subsession receiver_index collisions @@ -70,8 +70,8 @@ use crate::config::LpConfig; use crate::error::NymNodeError; use crate::node::lp::cleanup::CleanupTask; -use crate::node::lp::control::listener::LpControlListener; use crate::node::lp::data::listener::LpDataListener; +use control::ingress::listener::LpControlListener; use dashmap::DashMap; use nym_gateway::node::wireguard::PeerRegistrator; use nym_lp::peer::LpLocalPeer; @@ -82,12 +82,15 @@ use std::sync::Arc; use tokio::sync::Semaphore; use tracing::error; +use crate::node::lp::directory::LpNodes; +use crate::node::lp::state::{ActiveLpSessions, SharedLpNodeControlState}; pub use nym_mixnet_client::forwarder::{MixForwardingReceiver, mix_forwarding_channels}; -pub use state::{SharedLpControlState, SharedLpDataState, SharedLpState}; +pub use state::{SharedLpClientControlState, SharedLpDataState, SharedLpState}; mod cleanup; pub mod control; mod data; +pub mod directory; pub mod error; mod registration; pub mod state; @@ -107,11 +110,12 @@ impl LpSetup { lp_config: LpConfig, metrics: NymNodeMetrics, peer_registrator: Option, + network_nodes: LpNodes, mix_packet_sender: MixForwardingSender, shutdown: ShutdownTracker, ) -> Result { // TODO: this will require loading old states from disk in the future - let session_states = Arc::new(DashMap::new()); + let session_states = ActiveLpSessions::new(); let shared_lp_state = SharedLpState { metrics, @@ -119,13 +123,19 @@ impl LpSetup { session_states: session_states.clone(), }; - let control_state = SharedLpControlState { - local_lp_peer, + let client_control_state = SharedLpClientControlState { + local_lp_peer: local_lp_peer.clone(), peer_registrator, forward_semaphore: Arc::new(Semaphore::new(lp_config.debug.max_concurrent_forwards)), shared: shared_lp_state.clone(), }; + let nodes_control_state = SharedLpNodeControlState { + local_lp_peer, + nodes: network_nodes, + shared: shared_lp_state.clone(), + }; + let data_state = SharedLpDataState { outbound_mix_sender: mix_packet_sender, shared: shared_lp_state, @@ -133,7 +143,8 @@ impl LpSetup { let control_listener = LpControlListener::new( lp_config.control_bind_address, - control_state, + client_control_state, + nodes_control_state, shutdown.clone(), ); let data_listener = LpDataListener::new( diff --git a/nym-node/src/node/lp/registration.rs b/nym-node/src/node/lp/registration.rs index 678f57167f..fa51d49ac0 100644 --- a/nym-node/src/node/lp/registration.rs +++ b/nym-node/src/node/lp/registration.rs @@ -1,7 +1,7 @@ // Copyright 2025 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -use crate::node::lp::state::SharedLpControlState; +use crate::node::lp::state::SharedLpClientControlState; use nym_lp::peer_config::LpReceiverIndex; use nym_metrics::{add_histogram_obs, inc}; use nym_registration_common::dvpn::{ @@ -29,7 +29,7 @@ const LP_REGISTRATION_DURATION_BUCKETS: &[f64] = &[ 30.0, // 30s ]; -impl SharedLpControlState { +impl SharedLpClientControlState { async fn process_dvpn_initial_registration( &self, sender: LpReceiverIndex, diff --git a/nym-node/src/node/lp/state.rs b/nym-node/src/node/lp/state.rs index e003dded6a..c16dcf50ca 100644 --- a/nym-node/src/node/lp/state.rs +++ b/nym-node/src/node/lp/state.rs @@ -3,7 +3,10 @@ use crate::config::LpConfig; use crate::node::lp::cleanup::TimestampedState; +use crate::node::lp::directory::LpNodes; +use crate::node::lp::error::LpHandlerError; use dashmap::DashMap; +use dashmap::mapref::one::RefMut; use nym_gateway::node::wireguard::PeerRegistrator; use nym_lp::LpTransportSession; use nym_lp::peer::LpLocalPeer; @@ -15,7 +18,7 @@ use tokio::sync::Semaphore; /// Shared state for LP control connections #[derive(Clone)] -pub struct SharedLpControlState { +pub struct SharedLpClientControlState { /// Encapsulates all required key information of a local Lewes Protocol Peer. pub local_lp_peer: LpLocalPeer, @@ -34,6 +37,19 @@ pub struct SharedLpControlState { pub shared: SharedLpState, } +/// [Placeholder] Shared state for LP nodes control connections +#[derive(Clone)] +pub struct SharedLpNodeControlState { + /// Encapsulates all required key information of a local Lewes Protocol Peer. + pub local_lp_peer: LpLocalPeer, + + /// Information about all known LP nodes + pub nodes: LpNodes, + + /// Common shared data + pub shared: SharedLpState, +} + /// Shared state for LP data connections #[derive(Clone)] pub struct SharedLpDataState { @@ -48,6 +64,37 @@ pub struct SharedLpDataState { pub shared: SharedLpState, } +/// Established sessions keyed by the receiver index +/// +/// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions. +#[derive(Clone, Default)] +pub struct ActiveLpSessions { + // TODO: this might require split between client and node sessions. TBD + pub(crate) sessions: Arc>>, +} + +impl ActiveLpSessions { + pub fn new() -> Self { + Self::default() + } + + pub(crate) fn get_state_entry_mut( + &self, + receiver_index: LpReceiverIndex, + ) -> Result>, LpHandlerError> + { + self.sessions + .get_mut(&receiver_index) + .ok_or_else(|| LpHandlerError::MissingLpSession { receiver_index }) + } + + pub(crate) fn insert_new_session(&self, session: LpTransportSession) { + let receiver_index = session.receiver_index(); + self.sessions + .insert(receiver_index, TimestampedState::new(session)); + } +} + /// Shared state for LP connection handlers #[derive(Clone)] pub struct SharedLpState { @@ -57,8 +104,6 @@ pub struct SharedLpState { /// LP configuration (for timestamp validation, etc.) pub lp_config: LpConfig, - /// Established sessions keyed by receiver index - /// - /// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions. - pub session_states: Arc>>, + /// Currently active LP sessions + pub session_states: ActiveLpSessions, } diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index 6fdd5095ac..d755e7b12b 100644 --- a/nym-node/src/node/mod.rs +++ b/nym-node/src/node/mod.rs @@ -84,6 +84,7 @@ use tokio::sync::mpsc; use tracing::{debug, info, trace}; use zeroize::Zeroizing; +use crate::node::lp::directory::LpNodes; pub use nym_gateway::node::ActiveClientsStore; pub use nym_gateway::node::GatewayStorage; @@ -491,6 +492,7 @@ impl NymNode { &self, peer_registrator: Option, mix_packet_sender: MixForwardingSender, + network_nodes: LpNodes, ) -> Result { let lp_peer = LpLocalPeer::new(Ciphersuite::default(), self.x25519_lp_keys.clone()) .with_kem_keys(self.psq_kem_keys.clone()); @@ -500,6 +502,7 @@ impl NymNode { self.config.lp, self.metrics.clone(), peer_registrator, + network_nodes, mix_packet_sender, self.shutdown_manager.shutdown_tracker().clone(), ) @@ -683,6 +686,7 @@ impl NymNode { async fn start_gateway_tasks( &mut self, cached_network: CachedNetwork, + lp_nodes: LpNodes, metrics_sender: MetricEventsSender, active_clients_store: ActiveClientsStore, mix_packet_sender: MixForwardingSender, @@ -766,7 +770,7 @@ impl NymNode { self.config.lp.control_bind_address, self.config.lp.data_bind_address, ); let lp_tasks = self - .build_lp_tasks(wg_peer_registrator.clone(), mix_packet_sender) + .build_lp_tasks(wg_peer_registrator.clone(), mix_packet_sender, lp_nodes) .await?; lp_tasks.start_tasks(); } else { @@ -1355,6 +1359,7 @@ impl NymNode { let network_refresher = self.build_network_refresher().await?; let active_clients_store = ActiveClientsStore::new(); + let lp_nodes = network_refresher.lp_nodes(); let bloomfilters_manager = self.setup_replay_detection().await?; @@ -1381,6 +1386,7 @@ impl NymNode { self.start_gateway_tasks( network_refresher.cached_network(), + lp_nodes, metrics_sender, active_clients_store, mix_packet_sender, diff --git a/nym-node/src/node/shared_network.rs b/nym-node/src/node/shared_network.rs index 861d1a3e2b..c32fae2acc 100644 --- a/nym-node/src/node/shared_network.rs +++ b/nym-node/src/node/shared_network.rs @@ -3,6 +3,7 @@ use crate::error::NymNodeError; use crate::node::key_rotation::active_keys::ActiveSphinxKeys; +use crate::node::lp::directory::LpNodes; use crate::node::routing_filter::network_filter::NetworkRoutingFilter; use async_trait::async_trait; use nym_crypto::asymmetric::ed25519; @@ -212,6 +213,7 @@ pub struct NetworkRefresher { network: CachedNetwork, routing_filter: NetworkRoutingFilter, noise_view: NoiseNetworkView, + lp_nodes: LpNodes, } impl NetworkRefresher { @@ -240,6 +242,7 @@ impl NetworkRefresher { network: CachedNetwork::new_empty(), routing_filter: NetworkRoutingFilter::new_empty(testnet), noise_view: NoiseNetworkView::new_empty(), + lp_nodes: Default::default(), }; this.obtain_initial_network().await?; @@ -334,6 +337,7 @@ impl NetworkRefresher { }) .collect::>(); self.noise_view.swap_view(noise_nodes); + debug!("unimplemented: update LP nodes data"); let mut network_guard = self.network.inner.write().await; network_guard.topology_metadata = metadata.to_topology_metadata(); @@ -373,6 +377,10 @@ impl NetworkRefresher { self.noise_view.clone() } + pub(crate) fn lp_nodes(&self) -> LpNodes { + self.lp_nodes.clone() + } + pub(crate) async fn run(&mut self) { let mut full_refresh_interval = interval(self.full_refresh_interval); full_refresh_interval.reset(); diff --git a/nym-wallet/Cargo.lock b/nym-wallet/Cargo.lock index 7b049ffd8a..b3d10fd034 100644 --- a/nym-wallet/Cargo.lock +++ b/nym-wallet/Cargo.lock @@ -40,7 +40,7 @@ dependencies = [ "serde", "serde_json", "serde_repr", - "strum", + "strum 0.27.2", "tap", "tauri", "tauri-build", @@ -4506,8 +4506,8 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.9", - "strum", - "strum_macros", + "strum 0.28.0", + "strum_macros 0.28.0", "tendermint", "tendermint-rpc", "thiserror 2.0.12", @@ -4621,8 +4621,8 @@ dependencies = [ "nym-upgrade-mode-check", "rand 0.8.5", "serde", - "strum", - "strum_macros", + "strum 0.28.0", + "strum_macros 0.28.0", "thiserror 2.0.12", "time", "utoipa", @@ -4764,8 +4764,8 @@ version = "1.20.4" dependencies = [ "num_enum", "semver", - "strum", - "strum_macros", + "strum 0.28.0", + "strum_macros 0.28.0", "thiserror 2.0.12", ] @@ -4839,8 +4839,8 @@ dependencies = [ "schemars", "serde", "serde_json", - "strum", - "strum_macros", + "strum 0.28.0", + "strum_macros 0.28.0", "thiserror 2.0.12", "time", "url", @@ -4940,8 +4940,8 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.9", - "strum", - "strum_macros", + "strum 0.28.0", + "strum_macros 0.28.0", "thiserror 2.0.12", "ts-rs", "url", @@ -5055,8 +5055,8 @@ dependencies = [ "nym-vesting-contract-common", "serde", "serde_json", - "strum", - "strum_macros", + "strum 0.27.2", + "strum_macros 0.27.2", "ts-rs", ] @@ -7375,7 +7375,16 @@ version = "0.27.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" dependencies = [ - "strum_macros", + "strum_macros 0.27.2", +] + +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +dependencies = [ + "strum_macros 0.28.0", ] [[package]] @@ -7390,6 +7399,18 @@ dependencies = [ "syn 2.0.100", ] +[[package]] +name = "strum_macros" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" +dependencies = [ + "heck 0.5.0", + "proc-macro2", + "quote", + "syn 2.0.100", +] + [[package]] name = "subtle" version = "2.6.1" From 48dad0f16b7ebc3ff463bb3c306cd7874430f645 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 6 Mar 2026 10:18:45 +0000 Subject: [PATCH 3/4] bugfix: setting correct LpPeerConfig during handshake --- common/nym-lp/src/peer_config.rs | 5 ++ common/nym-lp/src/psq/initiator.rs | 56 ++++++++++++++----- common/nym-lp/src/psq/mod.rs | 31 +++++++--- common/nym-lp/src/session.rs | 29 +++++++++- integration-tests/src/lp_registration.rs | 23 +++++--- .../src/lp_client/client.rs | 4 +- .../src/lp_client/nested_session/mod.rs | 4 +- 7 files changed, 116 insertions(+), 36 deletions(-) diff --git a/common/nym-lp/src/peer_config.rs b/common/nym-lp/src/peer_config.rs index 02b1af34a4..d28dd6bc23 100644 --- a/common/nym-lp/src/peer_config.rs +++ b/common/nym-lp/src/peer_config.rs @@ -65,6 +65,7 @@ impl LpPeerConfig { rng.random(), ) } + /// Creates a new client to exit config. /// Inputs: /// hop_id: this value must be in the range (1..=15). This function returns an error if this is not the case. @@ -79,6 +80,7 @@ impl LpPeerConfig { { Self::new(rng, hop_id, true, false, censorship_resistance) } + /// Creates a new client to an intermediate node config. /// Inputs: /// hop_id: this value must be in the range (1..=14). This function returns an error if this is not the case. @@ -130,6 +132,7 @@ impl LpPeerConfig { rng.random(), ) } + fn build( hop_id: u8, is_exit: bool, @@ -147,6 +150,7 @@ impl LpPeerConfig { seed, } } + fn build_checked( hop_id: u8, is_exit: bool, @@ -203,6 +207,7 @@ impl LpPeerConfig { output_bytes[4..].copy_from_slice(&self.seed); output_bytes } + pub fn deserialize(bytes: &[u8]) -> Result { if bytes.len() != LP_PEER_CONFIG_SIZE { return Err(LpError::DeserializationError(format!( diff --git a/common/nym-lp/src/psq/initiator.rs b/common/nym-lp/src/psq/initiator.rs index 81fd571e01..7abc75be3b 100644 --- a/common/nym-lp/src/psq/initiator.rs +++ b/common/nym-lp/src/psq/initiator.rs @@ -24,10 +24,31 @@ use nym_kkt::message::{KKTRequest, KKTResponse}; use rand09::SeedableRng; use tracing::debug; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HandshakeMode { + // Client <> Entry + OneWayEntry, + + // Client <> Exit + OneWayExit, + + // Entry <> Exit + MutualInternode, + // in the future more variants will be supported (such as individual mix hops) +} + +impl HandshakeMode { + pub fn is_mutual(&self) -> bool { + matches!(self, HandshakeMode::MutualInternode) + } +} + pub struct PSQHandshakeStateInitiator<'a, S> { pub(super) inner_state: PSQHandshakeState<'a, S>, pub(super) initiator_data: InitiatorData, - pub(super) mutual: bool, + + /// The mode of the handshake (mutual node-node, client-entry, entry-exit) + pub(super) mode: HandshakeMode, } pub(crate) fn build_psq_principal( @@ -78,13 +99,23 @@ impl<'a, S> PSQHandshakeStateInitiator<'a, S> where S: LpHandshakeChannel + Unpin, { - pub fn set_mutual_kkt(mut self) -> Result { - if self.inner_state.local_peer.kem_keypairs.is_none() { - return Err(LpError::PSQMutualInitiatorMissingKemKey); - } + fn lp_peer_config(&self, rng: &mut R) -> Result + where + R: rand09::CryptoRng, + { + // for now we don't support censorship resistance flag + let censorship_resistance = false; - self.mutual = true; - Ok(self) + match self.mode { + HandshakeMode::OneWayEntry => Ok(LpPeerConfig::new_client_to_entry( + rng, + censorship_resistance, + )), + HandshakeMode::OneWayExit => { + LpPeerConfig::new_client_to_exit(rng, 1, censorship_resistance) + } + HandshakeMode::MutualInternode => LpPeerConfig::new_node_to_node(rng), + } } /// Attempt to send KKT request to begin the handshake @@ -132,7 +163,7 @@ where let ciphersuite = self.inner_state.local_peer.ciphersuite(); let kem = ciphersuite.kem(); - let lp_peer_config = LpPeerConfig::new_client_to_entry(rng, false); + let lp_peer_config = self.lp_peer_config(rng)?; // 1. retrieve the expected kem key hash. if we don't know it, let dir_hash = self @@ -141,7 +172,7 @@ where .expected_kem_key_hash(ciphersuite)?; // 2. prepare and send KKT request - let (mut initiator, kkt_request) = if self.mutual { + let (mut initiator, kkt_request) = if self.mode.is_mutual() { // this has been verified when setting the mutual flag let Some(local_encapsulation_key) = self.inner_state.local_peer.encoded_kem_key(kem) else { @@ -273,8 +304,8 @@ mod tests { resp.ciphersuite = ciphersuite; let initiator_data = InitiatorData::new(1, resp_remote); - let handshake_init = - PSQHandshakeState::new(conn_init, init).as_initiator(initiator_data); + let handshake_init = PSQHandshakeState::new(conn_init, init) + .as_initiator(initiator_data, HandshakeMode::OneWayEntry)?; let mut init_rng = DeterministicRng09Send::new(u64_seeded_rng_09(1)); @@ -396,8 +427,7 @@ mod tests { let initiator_data = InitiatorData::new(1, resp_remote); let handshake_init = PSQHandshakeState::new(conn_init, init) - .as_initiator(initiator_data) - .set_mutual_kkt()?; + .as_initiator(initiator_data, HandshakeMode::MutualInternode)?; let mut init_rng = DeterministicRng09Send::new(u64_seeded_rng_09(1)); diff --git a/common/nym-lp/src/psq/mod.rs b/common/nym-lp/src/psq/mod.rs index 7302e30c3a..0386e62880 100644 --- a/common/nym-lp/src/psq/mod.rs +++ b/common/nym-lp/src/psq/mod.rs @@ -12,6 +12,8 @@ mod helpers; pub mod initiator; pub mod responder; +use crate::LpError; +use crate::psq::initiator::HandshakeMode; pub use initiator::PSQHandshakeStateInitiator; pub use responder::PSQHandshakeStateResponder; @@ -107,12 +109,20 @@ where } } - pub fn as_initiator(self, initiator_data: InitiatorData) -> PSQHandshakeStateInitiator<'a, S> { - PSQHandshakeStateInitiator { + pub fn as_initiator( + self, + initiator_data: InitiatorData, + mode: HandshakeMode, + ) -> Result, LpError> { + if mode.is_mutual() && self.local_peer.kem_keypairs.is_none() { + return Err(LpError::PSQMutualInitiatorMissingKemKey); + } + + Ok(PSQHandshakeStateInitiator { initiator_data, inner_state: self, - mutual: false, - } + mode, + }) } pub fn as_responder(self, responder_data: ResponderData) -> PSQHandshakeStateResponder<'a, S> { @@ -160,8 +170,10 @@ mod tests { resp.ciphersuite = ciphersuite; let resp_remote = resp.as_remote(); - let handshake_init = PSQHandshakeState::new(conn_init, init) - .as_initiator(InitiatorData::new(1, resp_remote)); + let handshake_init = PSQHandshakeState::new(conn_init, init).as_initiator( + InitiatorData::new(1, resp_remote), + HandshakeMode::OneWayEntry, + )?; let handshake_resp = PSQHandshakeState::new(conn_resp, resp).as_responder(ResponderData::default()); @@ -232,9 +244,10 @@ mod tests { let resp_remote = resp.as_remote(); let init_remote = init.as_remote(); - let handshake_init = PSQHandshakeState::new(conn_init, init) - .as_initiator(InitiatorData::new(1, resp_remote)) - .set_mutual_kkt()?; + let handshake_init = PSQHandshakeState::new(conn_init, init).as_initiator( + InitiatorData::new(1, resp_remote), + HandshakeMode::MutualInternode, + )?; let handshake_resp = PSQHandshakeState::new(conn_resp, resp).as_responder( ResponderData::default() .with_initiator_kem_hashes(init_remote.expected_kem_key_digests), diff --git a/common/nym-lp/src/session.rs b/common/nym-lp/src/session.rs index 3d0d51efdd..0d02133374 100644 --- a/common/nym-lp/src/session.rs +++ b/common/nym-lp/src/session.rs @@ -9,6 +9,7 @@ use crate::codec::{decrypt_lp_packet, encrypt_lp_packet}; use crate::packet::{EncryptedLpPacket, LpHeader, LpMessage, LpPacket}; use crate::peer::{LpLocalPeer, LpRemotePeer}; use crate::peer_config::LpReceiverIndex; +use crate::psq::initiator::HandshakeMode; use crate::psq::{ InitiatorData, PSQHandshakeState, PSQHandshakeStateInitiator, PSQHandshakeStateResponder, ResponderData, @@ -154,12 +155,34 @@ impl LpTransportSession { local_peer: LpLocalPeer, remote_peer: LpRemotePeer, remote_protocol_version: u8, - ) -> PSQHandshakeStateInitiator<'_, S> + mode: HandshakeMode, + ) -> Result, LpError> where S: LpHandshakeChannel + Unpin, { - PSQHandshakeState::new(connection, local_peer) - .as_initiator(InitiatorData::new(remote_protocol_version, remote_peer)) + PSQHandshakeState::new(connection, local_peer).as_initiator( + InitiatorData::new(remote_protocol_version, remote_peer), + mode, + ) + } + + /// Helper function to create `PSQHandshakeState` for the handshake initiator for mutual KKT + pub fn psq_handshake_initiator_mutual_internode( + connection: &'_ mut S, + local_peer: LpLocalPeer, + remote_peer: LpRemotePeer, + remote_protocol_version: u8, + ) -> Result, LpError> + where + S: LpHandshakeChannel + Unpin, + { + Self::psq_handshake_initiator( + connection, + local_peer, + remote_peer, + remote_protocol_version, + HandshakeMode::MutualInternode, + ) } /// Helper function to create `PSQHandshakeState` for the handshake responder diff --git a/integration-tests/src/lp_registration.rs b/integration-tests/src/lp_registration.rs index ab9ef91ebc..875109dc77 100644 --- a/integration-tests/src/lp_registration.rs +++ b/integration-tests/src/lp_registration.rs @@ -17,9 +17,10 @@ mod tests { use nym_lp::peer::LpLocalPeer; use nym_node::config::{LpConfig, LpDebug}; use nym_node::node::GatewayStorage; - use nym_node::node::lp::control::ingress::client_handler::LpConnectionHandler; + use nym_node::node::lp::control::ingress::client_handler::LpClientConnectionHandler; use nym_node::node::lp::error::LpHandlerError; - use nym_node::node::lp::{SharedLpControlState, SharedLpState}; + use nym_node::node::lp::state::{ActiveLpSessions, NestedConnectionsManager}; + use nym_node::node::lp::{SharedLpClientControlState, SharedLpState}; use nym_node::wireguard::{PeerManager, PeerRegistrator}; use nym_registration_client::{LpClientError, LpRegistrationClient}; use nym_test_utils::helpers::{CryptoRng09, seeded_rng}; @@ -35,7 +36,7 @@ mod tests { use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; use tokio::sync::Semaphore; - use tokio::sync::mpsc::Receiver; + use tokio::sync::mpsc::{Receiver, channel}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::error; @@ -120,7 +121,7 @@ mod tests { enum SpawnedLpConnectionHandlerState { NotCreated, Ready { - handler: LpConnectionHandler, + handler: LpClientConnectionHandler, }, Running { handle: JoinHandle>>, @@ -130,7 +131,7 @@ mod tests { struct Gateway { base: Party, - lp_state: SharedLpControlState, + lp_state: SharedLpClientControlState, ip_pool: IpPool, mock_peer_controller: SpawnedPeerController, @@ -216,6 +217,9 @@ mod tests { let (mock_peer_controller, peer_controller_state) = mock_peer_controller(peer_request_rx); + let (connection_ctrl_sender, _connection_manager_receiver) = channel(42); + let nested_connections_manager = NestedConnectionsManager::new(connection_ctrl_sender); + // registering particular responses for peer controller is up to given test let ecash_verifier = Arc::new(ecash_verifier); @@ -225,7 +229,7 @@ mod tests { upgrade_mode_details, ); - let lp_state = SharedLpControlState { + let lp_state = SharedLpClientControlState { local_lp_peer: base.peer.clone(), forward_semaphore, @@ -235,8 +239,9 @@ mod tests { shared: SharedLpState { metrics: Default::default(), lp_config, - session_states: Arc::new(Default::default()), + session_states: ActiveLpSessions::new(), }, + nested_connections_manager, }; Ok(Gateway { @@ -262,7 +267,7 @@ mod tests { }; self.lp_connection_handler = SpawnedLpConnectionHandlerState::Ready { - handler: LpConnectionHandler::new( + handler: LpClientConnectionHandler::new( client_connection, client_address, self.lp_state.clone(), @@ -290,7 +295,7 @@ mod tests { } fn spawn_lp_handler(&mut self) { - let SpawnedLpConnectionHandlerState::Ready { handler } = mem::replace( + let SpawnedLpConnectionHandlerState::Ready { mut handler } = mem::replace( &mut self.lp_connection_handler, SpawnedLpConnectionHandlerState::NotCreated, ) else { diff --git a/nym-registration-client/src/lp_client/client.rs b/nym-registration-client/src/lp_client/client.rs index 53231ba5a3..c1de9db024 100644 --- a/nym-registration-client/src/lp_client/client.rs +++ b/nym-registration-client/src/lp_client/client.rs @@ -14,6 +14,7 @@ use nym_crypto::asymmetric::{ed25519, x25519}; use nym_lp::LpTransportSession; use nym_lp::peer::{DHKeyPair, LpLocalPeer, LpRemotePeer}; use nym_lp::peer_config::LpReceiverIndex; +use nym_lp::psq::initiator::HandshakeMode; use nym_lp::transport::traits::LpTransportChannel; use nym_lp::transport::{LpHandshakeChannel, LpTransportError}; use nym_lp::{Ciphersuite, packet::EncryptedLpPacket, packet::version}; @@ -396,7 +397,8 @@ where local_peer, remote_peer, protocol_version, - ) + HandshakeMode::OneWayEntry, + )? .complete_handshake() .await?; diff --git a/nym-registration-client/src/lp_client/nested_session/mod.rs b/nym-registration-client/src/lp_client/nested_session/mod.rs index c1a45f6fbe..5cf8e98434 100644 --- a/nym-registration-client/src/lp_client/nested_session/mod.rs +++ b/nym-registration-client/src/lp_client/nested_session/mod.rs @@ -28,6 +28,7 @@ use nym_crypto::asymmetric::{ed25519, x25519}; use nym_lp::packet::version; use nym_lp::packet::{EncryptedLpPacket, LpMessage}; use nym_lp::peer::{DHKeyPair, LpLocalPeer, LpRemotePeer}; +use nym_lp::psq::initiator::HandshakeMode; use nym_lp::transport::LpHandshakeChannel; use nym_lp::transport::traits::LpTransportChannel; use nym_lp::{Ciphersuite, KEM, LpTransportSession}; @@ -185,7 +186,8 @@ impl NestedLpSession { local_peer, remote_peer, protocol_version, - ) + HandshakeMode::OneWayExit, + )? .complete_handshake() .await?; From ef65cf4c9e172469c9d9e01f6dfe884628cf0e9f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 6 Mar 2026 16:34:42 +0000 Subject: [PATCH 4/4] additional adjustments --- common/nym-lp/src/session.rs | 15 +++ integration-tests/src/lp_registration.rs | 8 +- nym-node/src/node/lp/cleanup.rs | 2 - .../lp/control/{tests.rs => control_tests.rs} | 4 +- .../src/node/lp/control/egress/connection.rs | 102 ++++++++++++++++++ nym-node/src/node/lp/control/egress/mod.rs | 99 +---------------- .../node/lp/control/ingress/client_handler.rs | 2 +- .../src/node/lp/control/ingress/listener.rs | 4 +- nym-node/src/node/lp/control/ingress/mod.rs | 4 +- .../node/lp/control/ingress/node_handler.rs | 7 +- nym-node/src/node/lp/control/mod.rs | 2 +- nym-node/src/node/lp/directory.rs | 21 ++-- nym-node/src/node/lp/mod.rs | 1 - 13 files changed, 141 insertions(+), 130 deletions(-) rename nym-node/src/node/lp/control/{tests.rs => control_tests.rs} (97%) create mode 100644 nym-node/src/node/lp/control/egress/connection.rs diff --git a/common/nym-lp/src/session.rs b/common/nym-lp/src/session.rs index 0d02133374..b173057cec 100644 --- a/common/nym-lp/src/session.rs +++ b/common/nym-lp/src/session.rs @@ -20,6 +20,8 @@ use crate::{LpError, replay::ReceivingKeyCounterValidator}; use libcrux_psq::handshake::types::{Authenticator, DHPublicKey}; use libcrux_psq::session::{Session, SessionBinding}; use nym_kkt::keys::EncapsulationKey; +use nym_kkt_ciphersuite::{KEM, KEMKeyDigests}; +use std::collections::BTreeMap; use std::fmt::{Debug, Formatter}; /// Represents inputs that drive the state machine transitions. @@ -196,6 +198,19 @@ impl LpTransportSession { PSQHandshakeState::new(connection, local_peer).as_responder(ResponderData::default()) } + /// Helper function to create `PSQHandshakeState` for the handshake responder for mutual KKT + pub fn psq_handshake_responder_mutual( + connection: &'_ mut S, + local_peer: LpLocalPeer, + initiator_kem_hashes: BTreeMap, + ) -> PSQHandshakeStateResponder<'_, S> + where + S: LpHandshakeChannel + Unpin, + { + PSQHandshakeState::new(connection, local_peer) + .as_responder(ResponderData::default().with_initiator_kem_hashes(initiator_kem_hashes)) + } + pub fn session_binding(&self) -> &PersistentSessionBinding { &self.session_binding } diff --git a/integration-tests/src/lp_registration.rs b/integration-tests/src/lp_registration.rs index 875109dc77..34a98e06e1 100644 --- a/integration-tests/src/lp_registration.rs +++ b/integration-tests/src/lp_registration.rs @@ -19,7 +19,7 @@ mod tests { use nym_node::node::GatewayStorage; use nym_node::node::lp::control::ingress::client_handler::LpClientConnectionHandler; use nym_node::node::lp::error::LpHandlerError; - use nym_node::node::lp::state::{ActiveLpSessions, NestedConnectionsManager}; + use nym_node::node::lp::state::ActiveLpSessions; use nym_node::node::lp::{SharedLpClientControlState, SharedLpState}; use nym_node::wireguard::{PeerManager, PeerRegistrator}; use nym_registration_client::{LpClientError, LpRegistrationClient}; @@ -36,7 +36,7 @@ mod tests { use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; use tokio::sync::Semaphore; - use tokio::sync::mpsc::{Receiver, channel}; + use tokio::sync::mpsc::Receiver; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::error; @@ -217,9 +217,6 @@ mod tests { let (mock_peer_controller, peer_controller_state) = mock_peer_controller(peer_request_rx); - let (connection_ctrl_sender, _connection_manager_receiver) = channel(42); - let nested_connections_manager = NestedConnectionsManager::new(connection_ctrl_sender); - // registering particular responses for peer controller is up to given test let ecash_verifier = Arc::new(ecash_verifier); @@ -241,7 +238,6 @@ mod tests { lp_config, session_states: ActiveLpSessions::new(), }, - nested_connections_manager, }; Ok(Gateway { diff --git a/nym-node/src/node/lp/cleanup.rs b/nym-node/src/node/lp/cleanup.rs index 2a5e5fa0a1..546d8426cd 100644 --- a/nym-node/src/node/lp/cleanup.rs +++ b/nym-node/src/node/lp/cleanup.rs @@ -3,8 +3,6 @@ use crate::config::LpDebug; use crate::node::lp::state::ActiveLpSessions; -use nym_lp::LpTransportSession; -use nym_lp::peer_config::LpReceiverIndex; use nym_metrics::inc_by; use std::time::Duration; use tracing::{debug, info}; diff --git a/nym-node/src/node/lp/control/tests.rs b/nym-node/src/node/lp/control/control_tests.rs similarity index 97% rename from nym-node/src/node/lp/control/tests.rs rename to nym-node/src/node/lp/control/control_tests.rs index 2270e87e58..f83e8c76e9 100644 --- a/nym-node/src/node/lp/control/tests.rs +++ b/nym-node/src/node/lp/control/control_tests.rs @@ -35,14 +35,14 @@ mod tests { LpNodeDetails::new( rng.next_u32(), peer.kem_key_digests().clone(), - peer.x25519().clone(), + *peer.x25519(), version::CURRENT, ) } #[tokio::test] async fn basic_node_to_node_handshake() -> anyhow::Result<()> { - nym_test_utils::helpers::setup_test_logger(); + // nym_test_utils::helpers::setup_test_logger(); let (init, resp) = mock_peers(); let init_remote = init.as_remote(); diff --git a/nym-node/src/node/lp/control/egress/connection.rs b/nym-node/src/node/lp/control/egress/connection.rs new file mode 100644 index 0000000000..1ca4cfea00 --- /dev/null +++ b/nym-node/src/node/lp/control/egress/connection.rs @@ -0,0 +1,102 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +// not yet used +#![allow(dead_code)] + +use crate::node::lp::control::LpConnectionStats; +use crate::node::lp::directory::LpNodeDetails; +use crate::node::lp::error::LpHandlerError; +use crate::node::lp::state::SharedLpNodeControlState; +use nym_lp::LpTransportSession; +use nym_lp::transport::{LpHandshakeChannel, LpTransportChannel}; +use std::net::SocketAddr; +use tracing::debug; + +/// Initial connection handler for an egress LP node before completing the KKT/PSQ handshake. +pub struct InitialLpEgressNodeConnectionHandler { + stream: S, + remote_addr: SocketAddr, + responder_details: LpNodeDetails, + + state: SharedLpNodeControlState, + stats: LpConnectionStats, +} + +impl InitialLpEgressNodeConnectionHandler +where + S: LpHandshakeChannel + LpTransportChannel + Unpin, +{ + pub(crate) fn new( + stream: S, + remote_addr: SocketAddr, + responder_details: LpNodeDetails, + state: SharedLpNodeControlState, + ) -> Self { + Self { + stream, + remote_addr, + responder_details, + state, + stats: LpConnectionStats::new(), + } + } + + pub(crate) async fn complete_initial_handshake( + mut self, + ) -> Option> { + let remote = self.remote_addr; + + if self.responder_details.kem_key_hashes.is_empty() { + return Some(Err(LpHandlerError::MissingNodeKEMKeyHashes { + node_ip: self.remote_addr.ip(), + node_id: self.responder_details.node_id, + })); + } + + // 1. complete KKT/PSQ handshake before doing anything else. + // bail if it takes too long + let timeout = self.state.shared.lp_config.debug.handshake_ttl; + let stream = &mut self.stream; + + let handshake_state = match LpTransportSession::psq_handshake_initiator_mutual_internode( + stream, + self.state.local_lp_peer.clone(), + self.responder_details.to_lp_peer(), + self.responder_details.supported_protocol, + ) { + Ok(handshake_state) => handshake_state, + Err(err) => { + debug!("failed to initiate mutual KTT/PSQ handshake with {remote}: {err}"); + self.stats.emit_lifecycle_node_metrics(false); + return None; + } + }; + + let session = match tokio::time::timeout(timeout, handshake_state.complete_handshake()) + .await + { + Err(_timeout) => { + debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}"); + self.stats.emit_lifecycle_node_metrics(false); + return None; + } + Ok(Err(handshake_failure)) => { + debug!( + "failed to complete mutual KKT/PSQ handshake with {remote}: {handshake_failure}" + ); + self.stats.emit_lifecycle_node_metrics(false); + return None; + } + Ok(Ok(session)) => session, + }; + + debug!( + "completed egress KKT/PSQ handshake with node {}: {remote}", + self.responder_details.node_id + ); + + // TODO: change return type into complete handler once defined + Some(Ok(session)) + } +} diff --git a/nym-node/src/node/lp/control/egress/mod.rs b/nym-node/src/node/lp/control/egress/mod.rs index c43f964ded..f036b0ad6e 100644 --- a/nym-node/src/node/lp/control/egress/mod.rs +++ b/nym-node/src/node/lp/control/egress/mod.rs @@ -1,99 +1,4 @@ // Copyright 2026 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 +// SPDX-License-Identifier: GPL-3.0-only -use crate::node::lp::control::LpConnectionStats; -use crate::node::lp::directory::LpNodeDetails; -use crate::node::lp::error::LpHandlerError; -use crate::node::lp::state::SharedLpNodeControlState; -use nym_lp::LpTransportSession; -use nym_lp::transport::LpHandshakeChannel; -use std::net::SocketAddr; -use tracing::debug; - -/// Initial connection handler for an egress LP node before completing the KKT/PSQ handshake. -pub struct InitialLpEgressNodeConnectionHandler { - stream: S, - remote_addr: SocketAddr, - responder_details: LpNodeDetails, - - state: SharedLpNodeControlState, - stats: LpConnectionStats, -} - -impl InitialLpEgressNodeConnectionHandler -where - S: LpHandshakeChannel + LpHandshakeChannel + Unpin, -{ - pub(crate) fn new( - stream: S, - remote_addr: SocketAddr, - responder_details: LpNodeDetails, - state: SharedLpNodeControlState, - ) -> Self { - Self { - stream, - remote_addr, - responder_details, - state, - stats: LpConnectionStats::new(), - } - } - - pub(crate) async fn complete_initial_handshake( - mut self, - ) -> Option> { - let remote = self.remote_addr; - - if self.responder_details.kem_key_hashes.is_empty() { - return Some(Err(LpHandlerError::MissingNodeKEMKeyHashes { - node_ip: self.remote_addr.ip(), - node_id: self.responder_details.node_id, - })); - } - - // 1. complete KKT/PSQ handshake before doing anything else. - // bail if it takes too long - let timeout = self.state.shared.lp_config.debug.handshake_ttl; - let stream = &mut self.stream; - - let handshake_state = match LpTransportSession::psq_handshake_initiator_mutual_internode( - stream, - self.state.local_lp_peer.clone(), - self.responder_details.to_lp_peer(), - self.responder_details.supported_protocol, - ) { - Ok(handshake_state) => handshake_state, - Err(err) => { - debug!("failed to initiate mutual KTT/PSQ handshake with {remote}: {err}"); - self.stats.emit_lifecycle_node_metrics(false); - return None; - } - }; - - let session = match tokio::time::timeout(timeout, handshake_state.complete_handshake()) - .await - { - Err(_timeout) => { - debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}"); - self.stats.emit_lifecycle_node_metrics(false); - return None; - } - Ok(Err(handshake_failure)) => { - debug!( - "failed to complete mutual KKT/PSQ handshake with {remote}: {handshake_failure}" - ); - self.stats.emit_lifecycle_node_metrics(false); - return None; - } - Ok(Ok(session)) => session, - }; - - debug!( - "completed egress KKT/PSQ handshake with node {}: {remote}", - self.responder_details.node_id - ); - - // TODO: change return type into complete handler - Some(Ok(session)) - } -} +pub mod connection; diff --git a/nym-node/src/node/lp/control/ingress/client_handler.rs b/nym-node/src/node/lp/control/ingress/client_handler.rs index b8dd26a36f..7a3e9e1bd5 100644 --- a/nym-node/src/node/lp/control/ingress/client_handler.rs +++ b/nym-node/src/node/lp/control/ingress/client_handler.rs @@ -13,7 +13,7 @@ use nym_lp::session::{LpAction, LpInput}; use nym_lp::transport::LpHandshakeChannel; use nym_lp::transport::traits::LpTransportChannel; use nym_lp::{LpTransportSession, packet::message::ExpectedResponseSize}; -use nym_metrics::{add_histogram_obs, inc, inc_by}; +use nym_metrics::{add_histogram_obs, inc}; use nym_node_metrics::NymNodeMetrics; use nym_registration_common::{LpRegistrationRequest, RegistrationStatus}; use std::net::SocketAddr; diff --git a/nym-node/src/node/lp/control/ingress/listener.rs b/nym-node/src/node/lp/control/ingress/listener.rs index 34a09fba77..3d038abebe 100644 --- a/nym-node/src/node/lp/control/ingress/listener.rs +++ b/nym-node/src/node/lp/control/ingress/listener.rs @@ -88,13 +88,13 @@ impl LpControlListener { initiator_details: LpNodeDetails, ) { warn!( - "received an incoming LP node connection - we can't support this yet beyond the handshake!" + "received an incoming LP node connection - we don't support this yet beyond the handshake!" ); debug!("Accepting LP node connection from {remote_addr}"); // Spawn handler task - let mut handler = InitialLpIngressNodeConnectionHandler::new( + let handler = InitialLpIngressNodeConnectionHandler::new( stream, remote_addr, initiator_details, diff --git a/nym-node/src/node/lp/control/ingress/mod.rs b/nym-node/src/node/lp/control/ingress/mod.rs index 79a7f15682..b1c7fb39bc 100644 --- a/nym-node/src/node/lp/control/ingress/mod.rs +++ b/nym-node/src/node/lp/control/ingress/mod.rs @@ -1,6 +1,6 @@ // Copyright 2026 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 +// SPDX-License-Identifier: GPL-3.0-only pub mod client_handler; pub(crate) mod listener; -mod node_handler; +pub mod node_handler; diff --git a/nym-node/src/node/lp/control/ingress/node_handler.rs b/nym-node/src/node/lp/control/ingress/node_handler.rs index 2ade110c63..a06f887304 100644 --- a/nym-node/src/node/lp/control/ingress/node_handler.rs +++ b/nym-node/src/node/lp/control/ingress/node_handler.rs @@ -103,7 +103,7 @@ where })) } - pub async fn handle(mut self) -> Result<(), LpHandlerError> { + pub async fn handle(self) -> Result<(), LpHandlerError> { // Track total LP connections handled inc!("lp_node_connections_total"); @@ -119,6 +119,7 @@ where } /// Connection handler for an LP node after completing the KKT/PSQ handshake. +#[allow(dead_code)] pub struct LpIngressNodeConnectionHandler { stream: S, remote_addr: SocketAddr, @@ -136,13 +137,15 @@ impl LpIngressNodeConnectionHandler where S: LpHandshakeChannel + LpTransportChannel + Unpin, { - async fn handle(mut self) -> Result<(), LpHandlerError> { + async fn handle(self) -> Result<(), LpHandlerError> { // handle all the forwarding here self.stats.emit_lifecycle_node_metrics(true); Ok(()) } + // used in a unit test + #[allow(dead_code)] pub(crate) fn transport_session(&self) -> &LpTransportSession { &self.transport_session } diff --git a/nym-node/src/node/lp/control/mod.rs b/nym-node/src/node/lp/control/mod.rs index a51c6bacd9..a66c8f3728 100644 --- a/nym-node/src/node/lp/control/mod.rs +++ b/nym-node/src/node/lp/control/mod.rs @@ -3,9 +3,9 @@ use nym_metrics::{add_histogram_obs, inc, inc_by}; +mod control_tests; pub mod egress; pub mod ingress; -mod tests; // Histogram buckets for LP operation duration (legacy - used by unused forwarding methods) const LP_DURATION_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]; diff --git a/nym-node/src/node/lp/directory.rs b/nym-node/src/node/lp/directory.rs index 78e16357a2..c5496f532b 100644 --- a/nym-node/src/node/lp/directory.rs +++ b/nym-node/src/node/lp/directory.rs @@ -21,22 +21,15 @@ impl LpNodes { pub(crate) fn get_node_details(&self, node_ip: IpAddr) -> Option { self.nodes.load().get(&node_ip).cloned() } - - pub(crate) fn get_node_id(&self, node_ip: IpAddr) -> Option { - self.nodes - .load() - .get(&node_ip) - .map(|details| details.node_id) - } } #[derive(Clone)] -pub(crate) struct LpNodeDetails { +pub struct LpNodeDetails { inner: Arc, } impl LpNodeDetails { - pub(crate) fn new( + pub fn new( node_id: NodeId, kem_key_hashes: BTreeMap, x25519: DHPublicKey, @@ -61,11 +54,11 @@ impl Deref for LpNodeDetails { } } -pub(crate) struct LpNodeDetailsInner { - pub(crate) node_id: NodeId, - pub(crate) kem_key_hashes: BTreeMap, - pub(crate) x25519: DHPublicKey, - pub(crate) supported_protocol: u8, +pub struct LpNodeDetailsInner { + pub node_id: NodeId, + pub kem_key_hashes: BTreeMap, + pub x25519: DHPublicKey, + pub supported_protocol: u8, } impl LpNodeDetailsInner { diff --git a/nym-node/src/node/lp/mod.rs b/nym-node/src/node/lp/mod.rs index 8bc14fab6e..fd6dd3bcda 100644 --- a/nym-node/src/node/lp/mod.rs +++ b/nym-node/src/node/lp/mod.rs @@ -72,7 +72,6 @@ use crate::error::NymNodeError; use crate::node::lp::cleanup::CleanupTask; use crate::node::lp::data::listener::LpDataListener; use control::ingress::listener::LpControlListener; -use dashmap::DashMap; use nym_gateway::node::wireguard::PeerRegistrator; use nym_lp::peer::LpLocalPeer; use nym_mixnet_client::forwarder::MixForwardingSender;