From 32cfb3fff8b9f2672fcbb556dfb4f4b208787d69 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 4 Mar 2026 15:59:40 +0000 Subject: [PATCH] basic node LP handler --- common/nym-lp/src/session.rs | 30 +++++ integration-tests/src/lp_registration.rs | 2 +- nym-node/nym-node-metrics/src/network.rs | 44 +++++- nym-node/src/node/lp/client_handler.rs | 2 - .../control/{handler.rs => client_handler.rs} | 123 ++++------------- nym-node/src/node/lp/control/listener.rs | 123 ++++++++++++----- nym-node/src/node/lp/control/mod.rs | 116 +++++++++++++++- nym-node/src/node/lp/control/node_handler.rs | 125 ++++++++++++++++++ nym-node/src/node/lp/directory.rs | 35 ++++- nym-node/src/node/lp/error.rs | 6 +- nym-node/src/node/lp/listener.rs | 2 - nym-node/src/node/lp/mod.rs | 77 +++++------ nym-node/src/node/lp/registration.rs | 4 +- nym-node/src/node/lp/state.rs | 19 ++- 14 files changed, 518 insertions(+), 190 deletions(-) delete mode 100644 nym-node/src/node/lp/client_handler.rs rename nym-node/src/node/lp/control/{handler.rs => client_handler.rs} (89%) create mode 100644 nym-node/src/node/lp/control/node_handler.rs delete mode 100644 nym-node/src/node/lp/listener.rs diff --git a/common/nym-lp/src/session.rs b/common/nym-lp/src/session.rs index 3d0d51efdd..0ac55f5e62 100644 --- a/common/nym-lp/src/session.rs +++ b/common/nym-lp/src/session.rs @@ -19,6 +19,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. @@ -162,6 +164,21 @@ impl LpTransportSession { .as_initiator(InitiatorData::new(remote_protocol_version, remote_peer)) } + /// Helper function to create `PSQHandshakeState` for the handshake initiator for mutual KKT + pub fn psq_handshake_initiator_mutual( + connection: &'_ mut S, + local_peer: LpLocalPeer, + remote_peer: LpRemotePeer, + remote_protocol_version: u8, + ) -> Result, LpError> + where + S: LpHandshakeChannel + Unpin, + { + PSQHandshakeState::new(connection, local_peer) + .as_initiator(InitiatorData::new(remote_protocol_version, remote_peer)) + .set_mutual_kkt() + } + /// Helper function to create `PSQHandshakeState` for the handshake responder pub fn psq_handshake_responder( connection: &'_ mut S, @@ -173,6 +190,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 592ad2dd01..8af34edc9c 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::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/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/client_handler.rs b/nym-node/src/node/lp/client_handler.rs deleted file mode 100644 index bdc05451cb..0000000000 --- a/nym-node/src/node/lp/client_handler.rs +++ /dev/null @@ -1,2 +0,0 @@ -// Copyright 2026 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only diff --git a/nym-node/src/node/lp/control/handler.rs b/nym-node/src/node/lp/control/client_handler.rs similarity index 89% rename from nym-node/src/node/lp/control/handler.rs rename to nym-node/src/node/lp/control/client_handler.rs index 8d5ceb0249..312bd4b01a 100644 --- a/nym-node/src/node/lp/control/handler.rs +++ b/nym-node/src/node/lp/control/client_handler.rs @@ -2,8 +2,11 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::node::lp::cleanup::TimestampedState; +use crate::node::lp::control::{ + LP_CONNECTION_DURATION_BUCKETS, 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 +16,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 +24,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,7 +47,7 @@ pub struct LpConnectionHandler { exit_stream: Option<(S, SocketAddr)>, } -impl LpConnectionHandler +impl LpClientConnectionHandler where S: LpTransportChannel + LpHandshakeChannel + Unpin, { @@ -102,19 +55,23 @@ where stream: S, // forwarded_connection: bool, remote_addr: SocketAddr, - state: SharedLpControlState, + 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( @@ -135,11 +92,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 +118,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(()); } @@ -194,7 +146,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 +577,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); } } @@ -665,7 +594,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,7 +619,7 @@ 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, @@ -724,7 +653,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/listener.rs b/nym-node/src/node/lp/control/listener.rs index 742f526fe8..677f9e65fa 100644 --- a/nym-node/src/node/lp/control/listener.rs +++ b/nym-node/src/node/lp/control/listener.rs @@ -3,8 +3,12 @@ use crate::config::LpConfig; use crate::error::NymNodeError; -use crate::node::lp::control::handler::LpConnectionHandler; -use crate::node::lp::state::SharedLpControlState; +use crate::node::lp::control::client_handler::LpClientConnectionHandler; +use crate::node::lp::control::node_handler::{ + InitialLpNodeConnectionHandler, LpNodeConnectionHandler, +}; +use crate::node::lp::directory::{LpNodeDetails, LpNodes}; +use crate::node::lp::state::{SharedLpClientControlState, SharedLpNodeControlState}; use nym_task::ShutdownTracker; use std::net::SocketAddr; use tokio::net::TcpListener; @@ -15,8 +19,11 @@ pub struct LpControlListener { /// Address to bind to bind_address: SocketAddr, - /// Shared state for connection handlers - handler_state: SharedLpControlState, + /// Shared state for clients connection handlers + clients_handler_state: SharedLpClientControlState, + + /// Shared state for nodes connection handlers + nodes_handler_state: SharedLpNodeControlState, /// Shutdown coordination shutdown: ShutdownTracker, @@ -25,18 +32,19 @@ pub struct LpControlListener { impl LpControlListener { pub fn new( bind_address: SocketAddr, - handler_state: SharedLpControlState, + handler_state: SharedLpClientControlState, shutdown: ShutdownTracker, ) -> Self { - Self { - bind_address, - handler_state, - shutdown, - } + todo!() + // Self { + // bind_address, + // handler_state, + // shutdown, + // } } fn lp_config(&self) -> LpConfig { - self.handler_state.shared.lp_config + self.clients_handler_state.shared.lp_config } pub async fn run(&mut self) -> Result<(), NymNodeError> { @@ -74,9 +82,50 @@ impl LpControlListener { Ok(()) } - fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) { - // Check connection limit - let active_connections = self.active_lp_connections(); + fn handle_node_connection( + &self, + stream: tokio::net::TcpStream, + remote_addr: SocketAddr, + initiator_details: LpNodeDetails, + ) { + debug!("Accepting LP node connection from {remote_addr}"); + + // Spawn handler task + let mut handler = InitialLpNodeConnectionHandler::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!( @@ -86,45 +135,55 @@ impl LpControlListener { } debug!( - "Accepting LP connection from {remote_addr} ({active_connections} active connections)" + "Accepting LP client 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 mut handler = + LpClientConnectionHandler::new(stream, remote_addr, self.clients_handler_state.clone()); - let metrics = self.handler_state.shared.metrics.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 handler error for {remote_addr}: {e}"); + 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 } - - // Decrement connection counter on exit - metrics.network.lp_connection_closed(); }, - &format!("LP::{remote_addr}"), + &format!("LP_CLIENT::{remote_addr}"), ); } - fn active_lp_connections(&self) -> usize { - self.handler_state + 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_connections_count() + .active_lp_client_connections_count() } } diff --git a/nym-node/src/node/lp/control/mod.rs b/nym-node/src/node/lp/control/mod.rs index 5453ecddf5..3af37c6b08 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; +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/node_handler.rs b/nym-node/src/node/lp/control/node_handler.rs new file mode 100644 index 0000000000..2a3dc1cacc --- /dev/null +++ b/nym-node/src/node/lp/control/node_handler.rs @@ -0,0 +1,125 @@ +// 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 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 { + stream: S, + remote_addr: SocketAddr, + initiator_details: LpNodeDetails, + + state: SharedLpNodeControlState, + stats: LpConnectionStats, +} + +impl InitialLpNodeConnectionHandler +where + S: LpHandshakeChannel + Unpin, +{ + pub fn new( + stream: S, + remote_addr: SocketAddr, + initiator_details: LpNodeDetails, + state: SharedLpNodeControlState, + ) -> Self { + Self { + stream, + remote_addr, + initiator_details, + state, + stats: Default::default(), + } + } + + pub(crate) fn metrics(&self) -> &NymNodeMetrics { + &self.state.shared.metrics + } + + pub async fn handle(mut self) -> Result<(), LpHandlerError> { + // Track total LP connections handled + inc!("lp_node_connections_total"); + let remote = self.remote_addr; + + if self.initiator_details.kem_key_hashes.is_empty() { + return 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 session = match tokio::time::timeout(timeout, async move { + LpTransportSession::psq_handshake_responder_mutual( + stream, + local_peer, + self.initiator_details.kem_key_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 Ok(()); + } + 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(()); + } + Ok(Ok(session)) => session, + }; + + LpNodeConnectionHandler { + stream, + remote_addr: remote, + state: self.state, + stats: self.stats, + transport_session: session, + } + .handle() + .await + } +} + +/// Connection handler for an LP node after completing the KKT/PSQ handshake. +pub struct LpNodeConnectionHandler { + stream: S, + remote_addr: SocketAddr, + + state: SharedLpNodeControlState, + stats: LpConnectionStats, + transport_session: LpTransportSession, +} + +impl LpNodeConnectionHandler +where + S: LpTransportChannel + Unpin, +{ + async fn handle(mut self) -> Result<(), LpHandlerError> { + todo!(); + + self.stats.emit_lifecycle_node_metrics(true); + Ok(()) + } +} diff --git a/nym-node/src/node/lp/directory.rs b/nym-node/src/node/lp/directory.rs index 57638548bd..812f118f41 100644 --- a/nym-node/src/node/lp/directory.rs +++ b/nym-node/src/node/lp/directory.rs @@ -4,21 +4,44 @@ use arc_swap::ArcSwap; use nym_lp::peer::DHPublicKey; 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 -pub(crate) struct LpNodes { +#[derive(Clone)] +pub struct LpNodes { // map between all available ip addresses of other nodes and their details - nodes: ArcSwap>, + nodes: Arc>>, } -struct LpNodeDetails { +impl LpNodes { + pub(crate) fn is_from_known_node(&self, node_ip: IpAddr) -> bool { + self.nodes.load().contains_key(&node_ip) + } + + pub(crate) fn get_node_details(&self, node_ip: IpAddr) -> Option { + self.nodes.load().get(&node_ip).cloned() + } +} + +#[derive(Clone)] +pub(crate) struct LpNodeDetails { inner: Arc, } -struct LpNodeDetailsInner { - kem_key_hashes: BTreeMap, - x25519: DHPublicKey, +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, } diff --git a/nym-node/src/node/lp/error.rs b/nym-node/src/node/lp/error.rs index 8952e9f680..f26b680a83 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,6 +48,9 @@ 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, diff --git a/nym-node/src/node/lp/listener.rs b/nym-node/src/node/lp/listener.rs deleted file mode 100644 index bdc05451cb..0000000000 --- a/nym-node/src/node/lp/listener.rs +++ /dev/null @@ -1,2 +0,0 @@ -// Copyright 2026 - Nym Technologies SA -// SPDX-License-Identifier: GPL-3.0-only diff --git a/nym-node/src/node/lp/mod.rs b/nym-node/src/node/lp/mod.rs index cd550cbb19..c361fd33b5 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 @@ -83,11 +83,12 @@ use tokio::sync::Semaphore; use tracing::error; 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(crate) mod directory; pub mod error; mod registration; pub mod state; @@ -119,40 +120,42 @@ impl LpSetup { session_states: session_states.clone(), }; - let control_state = SharedLpControlState { - local_lp_peer, - peer_registrator, - forward_semaphore: Arc::new(Semaphore::new(lp_config.debug.max_concurrent_forwards)), - shared: shared_lp_state.clone(), - }; + todo!() - let data_state = SharedLpDataState { - outbound_mix_sender: mix_packet_sender, - shared: shared_lp_state, - }; - - let control_listener = LpControlListener::new( - lp_config.control_bind_address, - control_state, - shutdown.clone(), - ); - let data_listener = LpDataListener::new( - lp_config.data_bind_address, - data_state, - shutdown.clone_shutdown_token(), - ); - let cleanup_task = CleanupTask::new( - session_states, - lp_config.debug, - shutdown.clone_shutdown_token(), - ); - - Ok(LpSetup { - control_listener, - data_listener, - cleanup_task, - shutdown, - }) + // let control_state = SharedLpControlState { + // local_lp_peer, + // peer_registrator, + // forward_semaphore: Arc::new(Semaphore::new(lp_config.debug.max_concurrent_forwards)), + // shared: shared_lp_state.clone(), + // }; + // + // let data_state = SharedLpDataState { + // outbound_mix_sender: mix_packet_sender, + // shared: shared_lp_state, + // }; + // + // let control_listener = LpControlListener::new( + // lp_config.control_bind_address, + // control_state, + // shutdown.clone(), + // ); + // let data_listener = LpDataListener::new( + // lp_config.data_bind_address, + // data_state, + // shutdown.clone_shutdown_token(), + // ); + // let cleanup_task = CleanupTask::new( + // session_states, + // lp_config.debug, + // shutdown.clone_shutdown_token(), + // ); + // + // Ok(LpSetup { + // control_listener, + // data_listener, + // cleanup_task, + // shutdown, + // }) } pub fn start_tasks(mut self) { 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..1549372068 100644 --- a/nym-node/src/node/lp/state.rs +++ b/nym-node/src/node/lp/state.rs @@ -3,6 +3,7 @@ use crate::config::LpConfig; use crate::node::lp::cleanup::TimestampedState; +use crate::node::lp::directory::LpNodes; use dashmap::DashMap; use nym_gateway::node::wireguard::PeerRegistrator; use nym_lp::LpTransportSession; @@ -13,9 +14,9 @@ use nym_node_metrics::NymNodeMetrics; use std::sync::Arc; use tokio::sync::Semaphore; -/// Shared state for LP control connections +/// Shared state for LP clients 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, @@ -28,12 +29,26 @@ pub struct SharedLpControlState { /// telescope setup. When at capacity, forward requests return an error /// so clients can choose a different gateway. // this is temporary until there is persistent KKT/PSQ session between nodes + #[deprecated] pub forward_semaphore: Arc, /// Common shared data pub shared: SharedLpState, } +/// 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 {