Implement LP data handler on UDP:51264

- Add LpDataHandler for UDP data plane (port 51264) - Decrypt LP layer and forward Sphinx packets to mixnet - Add outbound_mix_sender to LpHandlerState - Integrate data handler spawn into LpListener::run() - Add metrics for data packets received/forwarded/errors Implements nym-yzzm
2025-12-23 10:18:26 +01:00
parent d9b0d5d599
commit 1945ad2e4d
4 changed files with 303 additions and 1 deletions
@@ -0,0 +1,249 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// 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<UdpSocket>,
+
+    /// 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<Self, GatewayError> {
+        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);
+    }
+}
@@ -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,
@@ -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<tokio::task::JoinHandle<()>, 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:
@@ -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)),