From bc0c2e5d19f2faf91eae24d828a5852ae3f8db8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 5 Mar 2026 10:38:13 +0000 Subject: [PATCH] basic node<>node KKT --- common/nym-lp/src/peer.rs | 6 +- nym-node/Cargo.toml | 1 + nym-node/src/node/lp/cleanup.rs | 7 +- .../src/node/lp/control/egress/connection.rs | 97 ++++++++++++++- .../node/lp/control/ingress/client_handler.rs | 16 +-- .../src/node/lp/control/ingress/listener.rs | 4 +- nym-node/src/node/lp/control/ingress/mod.rs | 113 ----------------- .../node/lp/control/ingress/node_handler.rs | 52 +++++--- nym-node/src/node/lp/control/mod.rs | 114 ++++++++++++++++++ nym-node/src/node/lp/control/tests.rs | 89 ++++++++++++++ nym-node/src/node/lp/directory.rs | 29 ++++- .../node/lp/forwarding/client_connection.rs | 2 +- nym-node/src/node/lp/forwarding/controller.rs | 2 +- nym-node/src/node/lp/mod.rs | 3 +- nym-node/src/node/lp/state.rs | 39 +++++- 15 files changed, 417 insertions(+), 157 deletions(-) create mode 100644 nym-node/src/node/lp/control/tests.rs diff --git a/common/nym-lp/src/peer.rs b/common/nym-lp/src/peer.rs index 3cfe665ee8..a018c9661c 100644 --- a/common/nym-lp/src/peer.rs +++ b/common/nym-lp/src/peer.rs @@ -1,5 +1,5 @@ // Copyright 2026 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 +// SPDX-License-Identifier: GPL-3.0-only use crate::LpError; use nym_kkt::keys::EncapsulationKey; @@ -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/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/src/node/lp/cleanup.rs b/nym-node/src/node/lp/cleanup.rs index 4b3ff23de6..0bc522a3d5 100644 --- a/nym-node/src/node/lp/cleanup.rs +++ b/nym-node/src/node/lp/cleanup.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::config::LpDebug; +use crate::node::lp::state::ActiveLpSessions; use dashmap::DashMap; use nym_lp::LpTransportSession; use nym_lp::peer_config::LpReceiverIndex; @@ -74,14 +75,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 +101,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/connection.rs b/nym-node/src/node/lp/control/egress/connection.rs index 582fee28d3..59f29215cf 100644 --- a/nym-node/src/node/lp/control/egress/connection.rs +++ b/nym-node/src/node/lp/control/egress/connection.rs @@ -1,11 +1,17 @@ // 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::forwarding::client_connection::NestedClientConnectionSender; +use crate::node::lp::state::SharedLpNodeControlState; +use nym_lp::LpTransportSession; use nym_lp::peer_config::LpReceiverIndex; +use nym_lp::transport::LpHandshakeChannel; use std::collections::HashMap; use std::net::SocketAddr; -use tracing::warn; +use tracing::{debug, warn}; pub(crate) type NestedNodeConnectionSender = (); pub(crate) type NestedNodeConnectionReceiver = (); @@ -13,6 +19,95 @@ pub(crate) type NestedNodeConnectionReceiver = (); pub(crate) type NestedNodeControlSender = (); pub(crate) type NestedNodeControlReceiver = (); +/// 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( + 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: return session, etc. + Some(Ok(session)) + } +} + pub(crate) struct NestedNodeConnectionHandler { /// Persistent connection to exit gateway for forwarding. /// Currently, it uses raw TCP socket, later it will be wrapped with dedicated PSQ tunnel 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 d8ddbf5cf2..6a5628730c 100644 --- a/nym-node/src/node/lp/control/ingress/client_handler.rs +++ b/nym-node/src/node/lp/control/ingress/client_handler.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::node::lp::cleanup::TimestampedState; -use crate::node::lp::control::ingress::{LP_DURATION_BUCKETS, LpConnectionStats}; +use crate::node::lp::control::{LP_DURATION_BUCKETS, LpConnectionStats}; use crate::node::lp::error::LpHandlerError; use crate::node::lp::state::SharedLpClientControlState; use dashmap::mapref::one::RefMut; @@ -80,8 +80,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 @@ -130,10 +129,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 @@ -584,7 +580,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}; @@ -623,8 +619,8 @@ mod tests { forward_semaphore, shared: SharedLpState { lp_config, - metrics: nym_node_metrics::NymNodeMetrics::default(), - session_states: Arc::new(dashmap::DashMap::new()), + metrics: NymNodeMetrics::default(), + session_states: ActiveLpSessions::new(), }, } } diff --git a/nym-node/src/node/lp/control/ingress/listener.rs b/nym-node/src/node/lp/control/ingress/listener.rs index f773eaefd3..ca316b57be 100644 --- a/nym-node/src/node/lp/control/ingress/listener.rs +++ b/nym-node/src/node/lp/control/ingress/listener.rs @@ -4,7 +4,7 @@ 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::InitialLpNodeConnectionHandler; +use crate::node::lp::control::ingress::node_handler::InitialLpIngressNodeConnectionHandler; use crate::node::lp::directory::{LpNodeDetails, LpNodes}; use crate::node::lp::state::{SharedLpClientControlState, SharedLpNodeControlState}; use nym_task::ShutdownTracker; @@ -89,7 +89,7 @@ impl LpControlListener { debug!("Accepting LP node connection from {remote_addr}"); // Spawn handler task - let mut handler = InitialLpNodeConnectionHandler::new( + let mut 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 b4bb0af30e..f25813149f 100644 --- a/nym-node/src/node/lp/control/ingress/mod.rs +++ b/nym-node/src/node/lp/control/ingress/mod.rs @@ -1,119 +1,6 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use nym_metrics::{add_histogram_obs, inc, inc_by}; - pub mod client_handler; pub(crate) mod listener; pub mod node_handler; - -// 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 -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/ingress/node_handler.rs b/nym-node/src/node/lp/control/ingress/node_handler.rs index 1faf39d2de..8f08f4d69d 100644 --- a/nym-node/src/node/lp/control/ingress/node_handler.rs +++ b/nym-node/src/node/lp/control/ingress/node_handler.rs @@ -1,7 +1,7 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::node::lp::control::ingress::LpConnectionStats; +use crate::node::lp::control::LpConnectionStats; use crate::node::lp::directory::LpNodeDetails; use crate::node::lp::error::LpHandlerError; use crate::node::lp::state::SharedLpNodeControlState; @@ -14,8 +14,8 @@ use std::net::SocketAddr; use tokio::net::TcpStream; use tracing::debug; -/// Initial connection handler for an LP node before completing the KKT/PSQ handshake. -pub struct InitialLpNodeConnectionHandler { +/// 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, @@ -24,7 +24,7 @@ pub struct InitialLpNodeConnectionHandler { stats: LpConnectionStats, } -impl InitialLpNodeConnectionHandler +impl InitialLpIngressNodeConnectionHandler where S: LpHandshakeChannel + LpTransportChannel + Unpin, { @@ -47,16 +47,16 @@ where &self.state.shared.metrics } - pub async fn handle(mut self) -> Result<(), LpHandlerError> { - // Track total LP connections handled - inc!("lp_node_connections_total"); + 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 Err(LpHandlerError::MissingNodeKEMKeyHashes { + 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. @@ -76,38 +76,50 @@ where Err(_timeout) => { debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}"); self.stats.emit_lifecycle_node_metrics(false); - return Ok(()); + 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 Ok(()); + return None; } Ok(Ok(session)) => session, }; debug!( - "completed KKT/PSQ handshake with node {}: {remote}", + "completed ingress KKT/PSQ handshake with node {}: {remote}", self.initiator_details.node_id ); - LpNodeConnectionHandler { + 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, - } - .handle() - .await + })) + } + + 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 LpNodeConnectionHandler { +pub struct LpIngressNodeConnectionHandler { stream: S, remote_addr: SocketAddr, remote_node_id: NodeId, @@ -117,7 +129,7 @@ pub struct LpNodeConnectionHandler { transport_session: LpTransportSession, } -impl LpNodeConnectionHandler +impl LpIngressNodeConnectionHandler where S: LpHandshakeChannel + LpTransportChannel + Unpin, { @@ -127,4 +139,8 @@ where 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/mod.rs b/nym-node/src/node/lp/control/mod.rs index 2567dc8f9e..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 +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..3580dea28b --- /dev/null +++ b/nym-node/src/node/lp/control/tests.rs @@ -0,0 +1,89 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +#[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 = resp.as_remote(); + let resp_remote = init.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 index 812f118f41..b1f7c0a79a 100644 --- a/nym-node/src/node/lp/directory.rs +++ b/nym-node/src/node/lp/directory.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: GPL-3.0-only use arc_swap::ArcSwap; -use nym_lp::peer::DHPublicKey; +use nym_lp::peer::{DHPublicKey, LpRemotePeer}; use nym_lp::{KEM, KEMKeyDigests}; use nym_topology::NodeId; use std::collections::{BTreeMap, HashMap}; @@ -11,7 +11,7 @@ use std::ops::Deref; use std::sync::Arc; /// Wrapper around all known LP nodes -#[derive(Clone)] +#[derive(Clone, Default)] pub struct LpNodes { // map between all available ip addresses of other nodes and their details nodes: Arc>>, @@ -32,6 +32,24 @@ 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; @@ -44,4 +62,11 @@ 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/forwarding/client_connection.rs b/nym-node/src/node/lp/forwarding/client_connection.rs index d179cefa6a..49dd255b9f 100644 --- a/nym-node/src/node/lp/forwarding/client_connection.rs +++ b/nym-node/src/node/lp/forwarding/client_connection.rs @@ -1,7 +1,7 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::node::lp::forwarding::node_connection::NestedNodeConnectionSender; +use crate::node::lp::control::egress::connection::NestedNodeConnectionSender; pub(crate) type NestedClientConnectionSender = (); pub(crate) type NestedClientConnectionReceiver = (); diff --git a/nym-node/src/node/lp/forwarding/controller.rs b/nym-node/src/node/lp/forwarding/controller.rs index 54625262dd..768edd7a22 100644 --- a/nym-node/src/node/lp/forwarding/controller.rs +++ b/nym-node/src/node/lp/forwarding/controller.rs @@ -1,7 +1,7 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -use crate::node::lp::forwarding::node_connection::NestedNodeControlSender; +use crate::node::lp::control::egress::connection::NestedNodeControlSender; use crate::node::lp::forwarding::{ ConnectionHandlerResponse, GetConnectionHandler, NestedConnectionControllerRequest, }; diff --git a/nym-node/src/node/lp/mod.rs b/nym-node/src/node/lp/mod.rs index 036dae45de..1e6d8d9c35 100644 --- a/nym-node/src/node/lp/mod.rs +++ b/nym-node/src/node/lp/mod.rs @@ -82,6 +82,7 @@ use std::sync::Arc; use tokio::sync::Semaphore; use tracing::error; +use crate::node::lp::state::ActiveLpSessions; pub use nym_mixnet_client::forwarder::{MixForwardingReceiver, mix_forwarding_channels}; pub use state::{SharedLpClientControlState, SharedLpDataState, SharedLpState}; @@ -113,7 +114,7 @@ impl LpSetup { 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, diff --git a/nym-node/src/node/lp/state.rs b/nym-node/src/node/lp/state.rs index 1549372068..416e1f45a7 100644 --- a/nym-node/src/node/lp/state.rs +++ b/nym-node/src/node/lp/state.rs @@ -4,7 +4,9 @@ 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; @@ -63,6 +65,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 { @@ -72,8 +105,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, }