GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add gateway subsession support with collision check and fast cleanup

Browse Source 
- 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
This commit is contained in:
durch
2025-11-30 22:45:11 +01:00
parent 0146bbfbb9
commit ca31c42794
3 changed files with 248 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nym-lp/src/state_machine.rs
+24
View File
@@ -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<Self, LpError> {
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<Result<LpAction, LpError>> {
// 1. Replace current state with a placeholder, taking ownership of the real current state.
gateway/src/node/lp_listener/handler.rs
+168 -40
View File
@@ -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<u8>,
) -> Result<(), GatewayError> {
// Try to deserialize as LpRegistrationRequest first (most common case after handshake)
if let Ok(request) = bincode::deserialize::<LpRegistrationRequest>(&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<LpPacket>,
subsession: nym_lp::session::SubsessionHandshake,
new_receiver_index: u32,
outer_key: Option<nym_lp::codec::OuterAeadKey>,
) -> 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))
gateway/src/node/lp_listener/mod.rs
+56 -9
View File
@@ -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<DashMap<u32, TimestampedState<LpSession>>>,
///
/// 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<DashMap<u32, TimestampedState<LpStateMachine>>>,
}
/// 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<DashMap<u32, TimestampedState<LpStateMachine>>>,
session_states: Arc<DashMap<u32, TimestampedState<LpSession>>>,
session_states: Arc<DashMap<u32, TimestampedState<LpStateMachine>>>,
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);
}
}
}
}