diff --git a/gateway/src/node/lp_listener/data_handler.rs b/gateway/src/node/lp_listener/data_handler.rs new file mode 100644 index 0000000000..cfb9b5f521 --- /dev/null +++ b/gateway/src/node/lp_listener/data_handler.rs @@ -0,0 +1,249 @@ +// Copyright 2025 - Nym Technologies SA +// SPDX-License-Identifier: GPL-3.0-only + +//! LP Data Handler - UDP listener for LP data plane (port 51264) +//! +//! This module handles the data plane for LP clients that have completed registration +//! via the control plane (TCP:41264). LP-wrapped Sphinx packets arrive here, get +//! decrypted, and are forwarded into the mixnet. +//! +//! # Packet Flow +//! +//! ```text +//! LP Client → UDP:51264 → LP Data Handler → Mixnet Entry +//! LP(Sphinx) decrypt LP forward Sphinx +//! ``` +//! +//! # Wire Format +//! +//! Each UDP packet is a complete LP packet: +//! - Header (8 bytes): receiver_idx (4) + counter (4) +//! - Payload: Outer AEAD encrypted Sphinx packet +//! +//! The receiver_idx is used to look up the session established during LP registration. + +use super::LpHandlerState; +use crate::error::GatewayError; +use nym_metrics::inc; +use nym_sphinx::forwarding::packet::MixPacket; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::net::UdpSocket; +use tracing::*; + +/// Maximum UDP packet size we'll accept +/// Sphinx packets are typically ~2KB, LP overhead is ~50 bytes, so 4KB is plenty +const MAX_UDP_PACKET_SIZE: usize = 4096; + +/// LP Data Handler for UDP data plane +pub struct LpDataHandler { + /// UDP socket for receiving LP-wrapped Sphinx packets + socket: Arc, + + /// Shared state with TCP control plane + state: LpHandlerState, + + /// Shutdown token + shutdown: nym_task::ShutdownToken, +} + +impl LpDataHandler { + /// Create a new LP data handler + pub async fn new( + bind_addr: SocketAddr, + state: LpHandlerState, + shutdown: nym_task::ShutdownToken, + ) -> Result { + let socket = UdpSocket::bind(bind_addr).await.map_err(|e| { + error!("Failed to bind LP data socket to {}: {}", bind_addr, e); + GatewayError::ListenerBindFailure { + address: bind_addr.to_string(), + source: Box::new(e), + } + })?; + + info!("LP data handler listening on UDP {}", bind_addr); + + Ok(Self { + socket: Arc::new(socket), + state, + shutdown, + }) + } + + /// Run the UDP packet receive loop + pub async fn run(self) -> Result<(), GatewayError> { + let mut buf = vec![0u8; MAX_UDP_PACKET_SIZE]; + + loop { + tokio::select! { + biased; + + _ = self.shutdown.cancelled() => { + info!("LP data handler: received shutdown signal"); + break; + } + + result = self.socket.recv_from(&mut buf) => { + match result { + Ok((len, src_addr)) => { + // Process packet in place (no spawn - UDP is fast) + if let Err(e) = self.handle_packet(&buf[..len], src_addr).await { + debug!("LP data packet error from {}: {}", src_addr, e); + inc!("lp_data_packet_errors"); + } + } + Err(e) => { + warn!("LP data socket recv error: {}", e); + inc!("lp_data_recv_errors"); + } + } + } + } + } + + info!("LP data handler shutdown complete"); + Ok(()) + } + + /// Handle a single UDP packet + /// + /// # Packet Processing Steps + /// 1. Parse LP header to get receiver_idx + /// 2. Look up session by receiver_idx + /// 3. Decrypt LP layer using session's outer AEAD key + /// 4. Parse inner payload as Sphinx packet + /// 5. Forward Sphinx packet to mixnet + async fn handle_packet(&self, packet: &[u8], src_addr: SocketAddr) -> Result<(), GatewayError> { + inc!("lp_data_packets_received"); + + // Step 1: Parse LP header (always cleartext for routing) + let header = nym_lp::codec::parse_lp_header_only(packet).map_err(|e| { + GatewayError::LpProtocolError(format!("Failed to parse LP header: {}", e)) + })?; + + let receiver_idx = header.receiver_idx; + + trace!( + "LP data packet from {} (receiver_idx={}, counter={}, len={})", + src_addr, + receiver_idx, + header.counter, + packet.len() + ); + + // Step 2: Look up session by receiver_idx + let session_entry = self + .state + .session_states + .get(&receiver_idx) + .ok_or_else(|| { + inc!("lp_data_unknown_session"); + GatewayError::LpProtocolError(format!( + "Unknown session for receiver_idx {}", + receiver_idx + )) + })?; + + // Update last activity timestamp + session_entry.value().touch(); + + // Step 3: Get outer AEAD key and decrypt LP layer + let outer_key = session_entry + .value() + .state + .session() + .map_err(|e| GatewayError::LpProtocolError(format!("Session error: {}", e)))? + .outer_aead_key() + .ok_or_else(|| { + GatewayError::LpProtocolError("Session has no outer AEAD key".to_string()) + })?; + + // Parse full packet with decryption + let lp_packet = + nym_lp::codec::parse_lp_packet(packet, Some(&outer_key)).map_err(|e| { + inc!("lp_data_decrypt_errors"); + GatewayError::LpProtocolError(format!("Failed to decrypt LP packet: {}", e)) + })?; + + // Step 4: Extract Sphinx packet from decrypted payload + // The LP message should be EncryptedData containing the Sphinx packet bytes + let sphinx_bytes = match lp_packet.message() { + nym_lp::LpMessage::EncryptedData(_) => { + // The session needs to decrypt the inner Noise layer + let session = session_entry + .value() + .state + .session() + .map_err(|e| GatewayError::LpProtocolError(format!("Session error: {}", e)))?; + + // decrypt_data expects &LpMessage, not raw bytes + session.decrypt_data(lp_packet.message()).map_err(|e| { + inc!("lp_data_inner_decrypt_errors"); + GatewayError::LpProtocolError(format!("Failed to decrypt inner data: {}", e)) + })? + } + other => { + return Err(GatewayError::LpProtocolError(format!( + "Expected EncryptedData in LP data packet, got {}", + other + ))); + } + }; + + // Release session lock before forwarding + drop(session_entry); + + // Step 5: Parse and forward Sphinx packet to mixnet + self.forward_sphinx_packet(&sphinx_bytes).await?; + + inc!("lp_data_packets_forwarded"); + Ok(()) + } + + /// Parse Sphinx packet bytes and forward to mixnet + /// + /// The decrypted LP payload contains a serialized MixPacket that includes: + /// - Packet type (1 byte) + /// - Key rotation (1 byte) + /// - Next hop address (first mix node) + /// - Sphinx packet data + async fn forward_sphinx_packet(&self, sphinx_bytes: &[u8]) -> Result<(), GatewayError> { + // Parse as MixPacket v2 format (packet_type || key_rotation || next_hop || packet) + let mix_packet = MixPacket::try_from_v2_bytes(sphinx_bytes).map_err(|e| { + inc!("lp_data_sphinx_parse_errors"); + GatewayError::LpProtocolError(format!("Failed to parse MixPacket: {}", e)) + })?; + + trace!( + "Forwarding Sphinx packet to mixnet (next_hop={}, type={:?})", + mix_packet.next_hop(), + mix_packet.packet_type() + ); + + // Forward to mixnet via the shared channel + if let Err(e) = self.state.outbound_mix_sender.forward_packet(mix_packet) { + error!("Failed to forward Sphinx packet to mixnet: {}", e); + inc!("lp_data_forward_errors"); + return Err(GatewayError::InternalError(format!( + "Mix packet forwarding failed: {}", + e + ))); + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_max_packet_size_reasonable() { + // Sphinx packets are typically around 2KB + // LP overhead is small (~50 bytes header + AEAD tag) + // 4KB should be plenty with room to spare + assert!(MAX_UDP_PACKET_SIZE >= 2048 + 100); + } +} diff --git a/gateway/src/node/lp_listener/handler.rs b/gateway/src/node/lp_listener/handler.rs index b2cabecce2..2a27fbd3c0 100644 --- a/gateway/src/node/lp_listener/handler.rs +++ b/gateway/src/node/lp_listener/handler.rs @@ -1299,6 +1299,9 @@ mod tests { lp_config.max_concurrent_forwards, )); + // Create mix forwarding channel (unused in tests but required by struct) + let (mix_sender, _mix_receiver) = nym_mixnet_client::forwarder::mix_forwarding_channels(); + LpHandlerState { lp_config, ecash_verifier: Arc::new(ecash_verifier) @@ -1309,6 +1312,7 @@ mod tests { active_clients_store: ActiveClientsStore::new(), wg_peer_controller: None, wireguard_data: None, + outbound_mix_sender: mix_sender, handshake_states: Arc::new(dashmap::DashMap::new()), session_states: Arc::new(dashmap::DashMap::new()), forward_semaphore, diff --git a/gateway/src/node/lp_listener/mod.rs b/gateway/src/node/lp_listener/mod.rs index 7eefd1b9f3..b5f0983118 100644 --- a/gateway/src/node/lp_listener/mod.rs +++ b/gateway/src/node/lp_listener/mod.rs @@ -73,6 +73,7 @@ use dashmap::DashMap; use nym_crypto::asymmetric::ed25519; use nym_gateway_storage::GatewayStorage; use nym_lp::state_machine::LpStateMachine; +use nym_mixnet_client::forwarder::MixForwardingSender; use nym_node_metrics::NymNodeMetrics; use nym_task::ShutdownTracker; use nym_wireguard::{PeerControlRequest, WireguardGatewayData}; @@ -82,6 +83,7 @@ use tokio::net::TcpListener; use tokio::sync::{mpsc, Semaphore}; use tracing::*; +mod data_handler; mod handler; mod messages; mod registration; @@ -333,6 +335,12 @@ pub struct LpHandlerState { /// LP configuration (for timestamp validation, etc.) pub lp_config: LpConfig, + /// Channel for forwarding Sphinx packets into the mixnet + /// + /// Used by the LP data handler (UDP:51264) to forward decrypted Sphinx packets + /// from LP clients into the mixnet for routing. + pub outbound_mix_sender: MixForwardingSender, + /// In-progress handshakes keyed by session_id /// /// Session ID is deterministically computed from both parties' X25519 keys immediately @@ -430,7 +438,7 @@ impl LpListener { })?; info!( - "LP listener started on {} (data port reserved: {})", + "LP listener started on {} (data port: {})", self.control_address, self.data_port ); @@ -439,6 +447,9 @@ impl LpListener { // Spawn background task for state cleanup let _cleanup_handle = self.spawn_state_cleanup_task(); + // Spawn UDP data handler for LP data plane (port 51264) + let _data_handler_handle = self.spawn_data_handler().await?; + loop { tokio::select! { biased; @@ -509,6 +520,43 @@ impl LpListener { ); } + /// Spawn the UDP data handler for LP data plane + /// + /// The data handler listens on UDP port 51264 and processes LP-wrapped Sphinx packets + /// from registered clients. It decrypts the LP layer and forwards the Sphinx packets + /// into the mixnet. + async fn spawn_data_handler(&self) -> Result, GatewayError> { + // Build data port address using same bind address as control port + let data_addr: SocketAddr = format!( + "{}:{}", + self.handler_state.lp_config.bind_address, self.data_port + ) + .parse() + .map_err(|e| { + GatewayError::InternalError(format!("Invalid LP data bind address: {}", e)) + })?; + + // Create data handler + let data_handler = data_handler::LpDataHandler::new( + data_addr, + self.handler_state.clone(), + self.shutdown.clone_shutdown_token(), + ) + .await?; + + // Spawn data handler task + let handle = self.shutdown.try_spawn_named( + async move { + if let Err(e) = data_handler.run().await { + error!("LP data handler error: {}", e); + } + }, + "LP::DataHandler", + ); + + Ok(handle) + } + /// Spawn background task for cleaning up stale state entries /// /// This task runs periodically (every `state_cleanup_interval_secs`) to remove: diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 189c250ada..0d6d1e1517 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -338,6 +338,7 @@ impl GatewayTasksBuilder { wg_peer_controller, wireguard_data: self.wireguard_data.as_ref().map(|wd| wd.inner.clone()), lp_config: self.config.lp.clone(), + outbound_mix_sender: self.mix_packet_sender.clone(), handshake_states: Arc::new(dashmap::DashMap::new()), session_states: Arc::new(dashmap::DashMap::new()), forward_semaphore: Arc::new(Semaphore::new(self.config.lp.max_concurrent_forwards)),