From ca31c42794ef942d7fb85e6b96f0191ff7133aef Mon Sep 17 00:00:00 2001 From: durch Date: Sun, 30 Nov 2025 22:45:11 +0100 Subject: [PATCH] Add gateway subsession support with collision check and fast cleanup - Store LpStateMachine in session_states (not LpSession) for subsession handling - Add LpStateMachine::from_subsession() factory for promoted sessions - Rewrite handle_transport_packet() to use state machine for all messages - Add handle_subsession_complete() for session promotion flow - Add collision check for new_receiver_index before insert (nym-90rw) - Add demoted_session_ttl_secs config (default 60s) for ReadOnlyTransport sessions to be cleaned up quickly after subsession promotion (nym-atza) - Track demoted session cleanup separately with lp_states_cleanup_demoted_removed --- common/nym-lp/src/state_machine.rs | 24 +++ gateway/src/node/lp_listener/handler.rs | 208 +++++++++++++++++++----- gateway/src/node/lp_listener/mod.rs | 65 +++++++- 3 files changed, 248 insertions(+), 49 deletions(-) diff --git a/common/nym-lp/src/state_machine.rs b/common/nym-lp/src/state_machine.rs index c32e6c9f4e..1ae84f0166 100644 --- a/common/nym-lp/src/state_machine.rs +++ b/common/nym-lp/src/state_machine.rs @@ -236,6 +236,30 @@ impl LpStateMachine { state: LpState::ReadyToHandshake { session }, }) } + + /// Creates a state machine in Transport state from a completed subsession handshake. + /// + /// This is used when a subsession (rekeying) completes and we need a new state machine + /// for the promoted session that can handle further subsession initiations (chained rekeying). + /// + /// # Arguments + /// + /// * `subsession` - The completed subsession handshake + /// * `receiver_index` - The new session's receiver index + /// + /// # Errors + /// + /// Returns error if the subsession handshake is not complete. + pub fn from_subsession( + subsession: SubsessionHandshake, + receiver_index: u32, + ) -> Result { + let session = subsession.into_session(receiver_index)?; + Ok(LpStateMachine { + state: LpState::Transport { session }, + }) + } + /// Processes an input event and returns a list of actions to perform. pub fn process_input(&mut self, input: LpInput) -> Option> { // 1. Replace current state with a placeholder, taking ownership of the real current state. diff --git a/gateway/src/node/lp_listener/handler.rs b/gateway/src/node/lp_listener/handler.rs index 5760c00296..caf3e1b0af 100644 --- a/gateway/src/node/lp_listener/handler.rs +++ b/gateway/src/node/lp_listener/handler.rs @@ -122,7 +122,13 @@ impl LpConnectionHandler { .and_then(|session| session.outer_aead_key()) } else if let Some(session_entry) = self.state.session_states.get(&receiver_idx) { // Established session - should always have PSK - session_entry.value().state.outer_aead_key() + // session_states now stores LpStateMachine (not LpSession) for subsession support + session_entry + .value() + .state + .session() + .ok() + .and_then(|s| s.outer_aead_key()) } else { // Unknown session - will error during routing, parse cleartext None @@ -326,16 +332,15 @@ impl LpConnectionHandler { self.remote_addr, receiver_idx ); - // Extract session and move to session_states + // Move state machine to session_states (already in Transport state) + // We keep the state machine (not just session) to enable + // subsession/rekeying support during transport phase drop(state_entry); // Release mutable borrow let (_receiver_idx, timestamped_state) = self.state.handshake_states.remove(&receiver_idx) .ok_or_else(|| GatewayError::LpHandshakeError("Failed to remove handshake state".to_string()))?; - let session = timestamped_state.state.into_session() - .map_err(|e| GatewayError::LpHandshakeError(format!("Failed to extract session: {}", e)))?; - - self.state.session_states.insert(receiver_idx, super::TimestampedState::new(session)); + self.state.session_states.insert(receiver_idx, timestamped_state); inc!("lp_handshakes_success"); @@ -363,51 +368,101 @@ impl LpConnectionHandler { /// Handle transport packet (receiver_idx!=0, session established) /// /// This handles packets on established sessions, which can be either: - /// 1. LpRegistrationRequest - Client registering for dVPN/Mixnet access - /// 2. ForwardPacketData - Client forwarding packets to exit gateway (telescoping) + /// 1. EncryptedData containing LpRegistrationRequest or ForwardPacketData + /// 2. SubsessionKK1 - Client initiates subsession/rekeying + /// 3. SubsessionReady - Client confirms subsession promotion + /// + /// We process all transport packets through the state machine to enable + /// subsession support. The state machine returns appropriate actions: + /// - DeliverData: decrypted application data to process + /// - SendPacket: subsession response (KK2) to send + /// - SubsessionComplete: subsession promoted, create new session async fn handle_transport_packet( &mut self, receiver_idx: u32, packet: LpPacket, ) -> Result<(), GatewayError> { + use nym_lp::state_machine::{LpAction, LpInput}; + debug!( "Processing transport packet from {} (receiver_idx={})", self.remote_addr, receiver_idx ); - let counter = packet.header().counter(); + // Get state machine and process packet + let mut state_entry = self.state.session_states.get_mut(&receiver_idx).ok_or_else(|| { + GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx)) + })?; - // Get session and decrypt payload - let decrypted_bytes = { - let session_entry = self.state.session_states.get(&receiver_idx).ok_or_else(|| { - GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx)) - })?; + // Update last activity timestamp + state_entry.value().touch(); - // Update last activity timestamp - session_entry.value().touch(); + let state_machine = &mut state_entry.value_mut().state; - let session = &session_entry.value().state; + // Process packet through state machine + let action = state_machine + .process_input(LpInput::ReceivePacket(packet)) + .ok_or_else(|| { + GatewayError::LpProtocolError("No action from state machine".to_string()) + })? + .map_err(|e| GatewayError::LpProtocolError(format!("State machine error: {}", e)))?; - // Validate counter BEFORE decryption to prevent replay DoS attacks. - // Counter is from cleartext header but authenticated by AEAD AAD, so this is safe. - session.receiving_counter_quick_check(counter).map_err(|e| { - inc!("lp_errors_replay_check"); - GatewayError::LpProtocolError(format!("Replay check failed: {}", e)) - })?; + // Get outer key before releasing borrow + let outer_key = state_machine.session().ok().and_then(|s| s.outer_aead_key()); + drop(state_entry); - // Decrypt packet (Noise inner layer) - let decrypted = session.decrypt_data(packet.message()).map_err(|e| { - GatewayError::LpProtocolError(format!("Failed to decrypt packet: {}", e)) - })?; - - // Mark counter as received after successful decryption - session.receiving_counter_mark(counter).map_err(|e| { - GatewayError::LpProtocolError(format!("Failed to mark counter: {}", e)) - })?; - - decrypted - }; + match action { + LpAction::SendPacket(response_packet) => { + // Subsession KK2 response - gateway is responder + // This means we received SubsessionKK1 and are responding + debug!( + "Sending subsession KK2 response to {} (receiver_idx={})", + self.remote_addr, receiver_idx + ); + inc!("lp_subsession_kk2_sent"); + self.send_lp_packet(&response_packet, outer_key.as_ref()).await?; + self.emit_lifecycle_metrics(true); + Ok(()) + } + LpAction::DeliverData(data) => { + // Decrypted application data - process as registration/forwarding + self.handle_decrypted_payload(receiver_idx, data.to_vec()).await + } + LpAction::SubsessionComplete { + packet: ready_packet, + subsession, + new_receiver_index, + } => { + // Subsession complete - promote to new session + self.handle_subsession_complete( + receiver_idx, + ready_packet, + subsession, + new_receiver_index, + outer_key, + ) + .await + } + other => { + warn!( + "Unexpected action in transport from {}: {:?}", + self.remote_addr, other + ); + self.emit_lifecycle_metrics(false); + Err(GatewayError::LpProtocolError(format!( + "Unexpected action: {:?}", + other + ))) + } + } + } + /// Handle decrypted transport payload (registration or forwarding request) + async fn handle_decrypted_payload( + &mut self, + receiver_idx: u32, + decrypted_bytes: Vec, + ) -> Result<(), GatewayError> { // Try to deserialize as LpRegistrationRequest first (most common case after handshake) if let Ok(request) = bincode::deserialize::(&decrypted_bytes) { debug!( @@ -433,9 +488,72 @@ impl LpConnectionHandler { ); inc!("lp_errors_unknown_payload_type"); self.emit_lifecycle_metrics(false); - Err(GatewayError::LpProtocolError(format!( - "Unknown transport payload type (not registration or forwarding)" - ))) + Err(GatewayError::LpProtocolError( + "Unknown transport payload type (not registration or forwarding)".to_string(), + )) + } + + /// Handle subsession completion - promote subsession to new session + /// + /// When a subsession handshake completes (SubsessionReady received): + /// 1. Send SubsessionReady packet if present (for initiator - gateway is responder, so None) + /// 2. Create new state machine from completed subsession + /// 3. Store new session under new_receiver_index + /// 4. Old session stays in ReadOnlyTransport state until TTL cleanup + async fn handle_subsession_complete( + &mut self, + old_receiver_idx: u32, + ready_packet: Option, + subsession: nym_lp::session::SubsessionHandshake, + new_receiver_index: u32, + outer_key: Option, + ) -> Result<(), GatewayError> { + use nym_lp::state_machine::LpStateMachine; + + info!( + "Subsession complete from {}: old_idx={}, new_idx={}", + self.remote_addr, old_receiver_idx, new_receiver_index + ); + + // Send SubsessionReady packet if present (for initiator - gateway is responder, so typically None) + if let Some(packet) = ready_packet { + self.send_lp_packet(&packet, outer_key.as_ref()).await?; + } + + // Create new state machine from completed subsession + let new_state_machine = LpStateMachine::from_subsession(subsession, new_receiver_index) + .map_err(|e| { + GatewayError::LpProtocolError(format!("Failed to create session from subsession: {}", e)) + })?; + + // Check for receiver_index collision before inserting + // new_receiver_index is client-generated (rand::random() in state machine). + // Collisions are statistically unlikely (1 in 4 billion) but could cause DoS if exploited. + if self.state.session_states.contains_key(&new_receiver_index) + || self.state.handshake_states.contains_key(&new_receiver_index) + { + warn!( + "Subsession receiver_index collision: {} from {}", + new_receiver_index, self.remote_addr + ); + inc!("lp_subsession_receiver_index_collision"); + self.emit_lifecycle_metrics(false); + return Err(GatewayError::LpProtocolError( + "Subsession receiver index collision - client should retry".to_string(), + )); + } + + // Store new session under new_receiver_index + self.state + .session_states + .insert(new_receiver_index, super::TimestampedState::new(new_state_machine)); + + // Old session is now in ReadOnlyTransport state (handled by state machine) + // It will be cleaned up by TTL-based cleanup task + + inc!("lp_subsession_complete"); + self.emit_lifecycle_metrics(true); + Ok(()) } /// Handle registration request on an established session @@ -452,7 +570,12 @@ impl LpConnectionHandler { let session_entry = self.state.session_states.get(&receiver_idx).ok_or_else(|| { GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx)) })?; - let session = &session_entry.value().state; + // Access session via state machine for subsession support + let session = session_entry + .value() + .state + .session() + .map_err(|e| GatewayError::LpProtocolError(format!("Session error: {}", e)))?; // Serialize and encrypt response let response_bytes = bincode::serialize(&response).map_err(|e| { @@ -509,7 +632,12 @@ impl LpConnectionHandler { let session_entry = self.state.session_states.get(&receiver_idx).ok_or_else(|| { GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx)) })?; - let session = &session_entry.value().state; + // Access session via state machine for subsession support + let session = session_entry + .value() + .state + .session() + .map_err(|e| GatewayError::LpProtocolError(format!("Session error: {}", e)))?; let encrypted_message = session.encrypt_data(&response_bytes).map_err(|e| { GatewayError::LpProtocolError(format!("Failed to encrypt forward response: {}", e)) diff --git a/gateway/src/node/lp_listener/mod.rs b/gateway/src/node/lp_listener/mod.rs index b75b6523dc..ee1f3301ed 100644 --- a/gateway/src/node/lp_listener/mod.rs +++ b/gateway/src/node/lp_listener/mod.rs @@ -56,6 +56,12 @@ // ## State Cleanup Metrics (in cleanup task) // - lp_states_cleanup_handshake_removed: Counter for stale handshakes removed by cleanup task // - 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) +// - 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 // // ## Usage Example // To view metrics, the nym-metrics registry automatically collects all metrics. @@ -67,7 +73,6 @@ use dashmap::DashMap; use nym_crypto::asymmetric::ed25519; use nym_gateway_storage::GatewayStorage; use nym_lp::state_machine::LpStateMachine; -use nym_lp::LpSession; use nym_node_metrics::NymNodeMetrics; use nym_task::ShutdownTracker; use nym_wireguard::{PeerControlRequest, WireguardGatewayData}; @@ -149,6 +154,14 @@ pub struct LpConfig { #[serde(default = "default_session_ttl_secs")] pub session_ttl_secs: u64, + /// Maximum age of demoted (read-only) sessions before cleanup (default: 60s) + /// + /// After subsession promotion, old sessions enter ReadOnlyTransport state. + /// They only need to stay alive briefly to drain in-flight packets. + /// This shorter TTL prevents memory buildup from frequent rekeying. + #[serde(default = "default_demoted_session_ttl_secs")] + pub demoted_session_ttl_secs: u64, + /// How often to run the state cleanup task (default: 5 minutes) /// /// The cleanup task scans for and removes stale handshakes and sessions. @@ -170,6 +183,7 @@ impl Default for LpConfig { use_mock_ecash: default_use_mock_ecash(), handshake_ttl_secs: default_handshake_ttl_secs(), session_ttl_secs: default_session_ttl_secs(), + demoted_session_ttl_secs: default_demoted_session_ttl_secs(), state_cleanup_interval_secs: default_state_cleanup_interval_secs(), } } @@ -207,6 +221,10 @@ fn default_session_ttl_secs() -> u64 { 86400 // 24 hours - for long-lived dVPN sessions } +fn default_demoted_session_ttl_secs() -> u64 { + 60 // 1 minute - enough to drain in-flight packets after subsession promotion +} + fn default_state_cleanup_interval_secs() -> u64 { 300 // 5 minutes - balances memory reclamation with task overhead } @@ -314,7 +332,12 @@ pub struct LpHandlerState { /// by session_id, decrypt/process, respond. /// /// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions. - pub session_states: Arc>>, + /// + /// Sessions are stored as LpStateMachine (not LpSession) to enable + /// subsession/rekeying support. The state machine handles subsession initiation + /// (SubsessionKK1/KK2/Ready) during transport phase, allowing long-lived connections + /// to rekey without re-authentication. + pub session_states: Arc>>, } /// LP listener that accepts TCP connections on port 41264 @@ -456,13 +479,14 @@ impl LpListener { let session_states = Arc::clone(&self.handler_state.session_states); let handshake_ttl = self.handler_state.lp_config.handshake_ttl_secs; let session_ttl = self.handler_state.lp_config.session_ttl_secs; + let demoted_session_ttl = self.handler_state.lp_config.demoted_session_ttl_secs; let interval_secs = self.handler_state.lp_config.state_cleanup_interval_secs; let shutdown = self.shutdown.clone_shutdown_token(); let metrics = self.handler_state.metrics.clone(); info!( - "Starting LP state cleanup task (handshake_ttl={}s, session_ttl={}s, interval={}s)", - handshake_ttl, session_ttl, interval_secs + "Starting LP state cleanup task (handshake_ttl={}s, session_ttl={}s, demoted_ttl={}s, interval={}s)", + handshake_ttl, session_ttl, demoted_session_ttl, interval_secs ); self.shutdown.try_spawn_named( @@ -471,6 +495,7 @@ impl LpListener { session_states, handshake_ttl, session_ttl, + demoted_session_ttl, interval_secs, shutdown, metrics, @@ -483,15 +508,20 @@ impl LpListener { /// /// Runs periodically to scan handshake_states and session_states maps, /// removing entries that have exceeded their TTL. + /// + /// Demoted sessions (ReadOnlyTransport) use shorter TTL since they + /// only need to drain in-flight packets after subsession promotion. async fn cleanup_loop( handshake_states: Arc>>, - session_states: Arc>>, + session_states: Arc>>, handshake_ttl_secs: u64, session_ttl_secs: u64, + demoted_session_ttl_secs: u64, interval_secs: u64, shutdown: nym_task::ShutdownToken, _metrics: NymNodeMetrics, ) { + use nym_lp::state_machine::LpStateBare; use nym_metrics::inc_by; let mut cleanup_interval = @@ -510,6 +540,7 @@ impl LpListener { let start = std::time::Instant::now(); let mut hs_removed = 0u64; let mut ss_removed = 0u64; + let mut demoted_removed = 0u64; // Remove stale handshakes (based on age since creation) handshake_states.retain(|_, timestamped| { @@ -522,21 +553,34 @@ impl LpListener { }); // Remove stale sessions (based on time since last activity) + // Use shorter TTL for demoted (ReadOnlyTransport) sessions session_states.retain(|_, timestamped| { - if timestamped.seconds_since_activity() > session_ttl_secs { - ss_removed += 1; + let is_demoted = timestamped.state.bare_state() == LpStateBare::ReadOnlyTransport; + let ttl = if is_demoted { + demoted_session_ttl_secs + } else { + session_ttl_secs + }; + + if timestamped.seconds_since_activity() > ttl { + if is_demoted { + demoted_removed += 1; + } else { + ss_removed += 1; + } false } else { true } }); - if hs_removed > 0 || ss_removed > 0 { + if hs_removed > 0 || ss_removed > 0 || demoted_removed > 0 { let duration = start.elapsed(); info!( - "LP state cleanup: removed {} handshakes, {} sessions (took {:.3}s)", + "LP state cleanup: removed {} handshakes, {} sessions, {} demoted (took {:.3}s)", hs_removed, ss_removed, + demoted_removed, duration.as_secs_f64() ); @@ -547,6 +591,9 @@ impl LpListener { if ss_removed > 0 { inc_by!("lp_states_cleanup_session_removed", ss_removed as i64); } + if demoted_removed > 0 { + inc_by!("lp_states_cleanup_demoted_removed", demoted_removed as i64); + } } } }