diff --git a/common/nym-kkt/src/frame.rs b/common/nym-kkt/src/frame.rs index 6e76661619..f3a866aa2f 100644 --- a/common/nym-kkt/src/frame.rs +++ b/common/nym-kkt/src/frame.rs @@ -8,9 +8,10 @@ // [6] => Reserved use crate::context::{KKTMode, KKTRole}; -use crate::message::{DecryptedRequestFrame, KKTRequest, KKTRequestPlaintext}; +use crate::message::{ + DecryptedRequestFrame, KKTRequest, KKTRequestEncryptionResult, KKTRequestPlaintext, +}; use crate::{ - carrier::Carrier, context::{KKT_CONTEXT_LEN, KKTContext}, error::KKTError, }; @@ -70,7 +71,7 @@ impl KKTFrame { rng: &mut R, responder_public_key: &DHPublicKey, version_byte: u8, - ) -> Result<(Carrier, KKTRequest), KKTError> + ) -> Result where R: CryptoRng + RngCore, { @@ -79,11 +80,17 @@ impl KKTFrame { let plaintext = KKTRequestPlaintext::new(ephemeral_keypair.pk, responder_public_key, version_byte); + let receiver_index = self.derive_receiver_index(&plaintext)?; + let mut carrier = plaintext.derive_initiator_carrier(ephemeral_keypair.sk(), responder_public_key)?; let full_kkt_message = plaintext.into_request(&mut carrier, self)?; - Ok((carrier, full_kkt_message)) + Ok(KKTRequestEncryptionResult { + carrier, + receiver_index, + request: full_kkt_message, + }) } pub fn decrypt_initiator_frame( @@ -106,31 +113,32 @@ impl KKTFrame { .plaintext .derive_responder_carrier(responder_keypair)?; - let decrypted_message = carrier.decrypt(&message.encrypted_frame)?; let frame = KKTFrame::from_bytes(&decrypted_message)?; - - let receiver_index: u64 = frame.derive_receiver_index( &message.plaintext)?; + + let receiver_index: u64 = frame.derive_receiver_index(&message.plaintext)?; Ok(DecryptedRequestFrame { carrier, remote_frame: frame, outer_protocol_version, - receiver_index + receiver_index, }) } // HASH(context || pub_key || masked_version || decrypted frame) - fn derive_receiver_index(&self, kkt_outer_headers: &KKTRequestPlaintext) -> Result { - - let mut receiver_index_bytes: [u8; 8] = [0u8; 8]; + fn derive_receiver_index( + &self, + kkt_outer_headers: &KKTRequestPlaintext, + ) -> Result { + let mut receiver_index_bytes = [0u8; 8]; let mut hasher = blake3::Hasher::new(); hasher.update(KKT_RECEIVER_INDEX_CONTEXT); hasher.update(&kkt_outer_headers.to_bytes()); hasher.update(&self.try_to_bytes()?); - + hasher.finalize_xof().fill(&mut receiver_index_bytes); Ok(u64::from_le_bytes(receiver_index_bytes)) diff --git a/common/nym-kkt/src/initiator.rs b/common/nym-kkt/src/initiator.rs index d5d98a20ea..06d665132e 100644 --- a/common/nym-kkt/src/initiator.rs +++ b/common/nym-kkt/src/initiator.rs @@ -27,6 +27,11 @@ pub struct KKTInitiator<'a> { expected_hash: &'a [u8], } +pub struct KKTRequestWithReceiverIndex { + pub request: KKTRequest, + pub receiver_index: u64, +} + impl<'a> KKTInitiator<'a> { // to be used by clients pub fn generate_one_way_request( @@ -35,7 +40,7 @@ impl<'a> KKTInitiator<'a> { responder_dh_public_key: &DHPublicKey, expected_hash: &'a [u8], outer_protocol_version: u8, - ) -> Result<(Self, KKTRequest), KKTError> + ) -> Result<(Self, KKTRequestWithReceiverIndex), KKTError> where R: CryptoRng + RngCore, { @@ -58,7 +63,7 @@ impl<'a> KKTInitiator<'a> { responder_dh_public_key: &DHPublicKey, expected_hash: &'a [u8], outer_protocol_version: u8, - ) -> Result<(Self, KKTRequest), KKTError> + ) -> Result<(Self, KKTRequestWithReceiverIndex), KKTError> where R: CryptoRng + RngCore, { @@ -81,22 +86,26 @@ impl<'a> KKTInitiator<'a> { responder_dh_public_key: &DHPublicKey, expected_hash: &'a [u8], outer_protocol_version: u8, - ) -> Result<(Self, KKTRequest), KKTError> + ) -> Result<(Self, KKTRequestWithReceiverIndex), KKTError> where R: CryptoRng + RngCore, { let frame = initiator_process(mode, ciphersuite, local_encapsulation_key)?; let context = *frame.context(); - let (carrier, request) = + + let request = frame.encrypt_initiator_frame(rng, responder_dh_public_key, outer_protocol_version)?; Ok(( Self { - carrier, + carrier: request.carrier, context, expected_hash, }, - request, + KKTRequestWithReceiverIndex { + request: request.request, + receiver_index: request.receiver_index, + }, )) } diff --git a/common/nym-kkt/src/message.rs b/common/nym-kkt/src/message.rs index 51825afcdd..7b57aeae82 100644 --- a/common/nym-kkt/src/message.rs +++ b/common/nym-kkt/src/message.rs @@ -179,6 +179,17 @@ impl KKTRequestPlaintext { } } +pub struct KKTRequestEncryptionResult { + /// Derived carrier used for decrypting this frame and encrypting the response + pub(crate) carrier: Carrier, + + /// A unique index associated to the request sender + pub(crate) receiver_index: u64, + + /// The underlying request that is going to get sent to the remote + pub(crate) request: KKTRequest, +} + pub struct DecryptedRequestFrame { /// Derived carrier used for decrypting this frame and encrypting the response pub(crate) carrier: Carrier, @@ -190,7 +201,7 @@ pub struct DecryptedRequestFrame { pub(crate) outer_protocol_version: u8, /// A unique index associated to the request sender - pub(crate) receiver_index: u64 + pub(crate) receiver_index: u64, } impl DecryptedRequestFrame { @@ -208,6 +219,9 @@ pub struct ProcessedKKTRequest { /// The KEM key requested in the original request pub requested_kem: KEM, + /// Established receiver index used for session lookup + pub receiver_index: u64, + /// The unmasked byte representing the outer protocol version sent by the initiator pub outer_protocol_version: u8, } diff --git a/common/nym-kkt/src/responder.rs b/common/nym-kkt/src/responder.rs index 93588d2dd7..8233487c58 100644 --- a/common/nym-kkt/src/responder.rs +++ b/common/nym-kkt/src/responder.rs @@ -128,6 +128,7 @@ impl<'a> KKTResponder<'a> { response: KKTResponse { encrypted_frame }, remote_encapsulation_key, requested_kem: remote_context.ciphersuite().kem(), + receiver_index: processed_req.receiver_index, outer_protocol_version: processed_req.outer_protocol_version, }) } diff --git a/common/nym-lp-packet/src/header.rs b/common/nym-lp-packet/src/header.rs index f40f5947cd..37c86de9f5 100644 --- a/common/nym-lp-packet/src/header.rs +++ b/common/nym-lp-packet/src/header.rs @@ -14,14 +14,14 @@ use tracing::warn; /// For encrypted packets, this is the AAD (additional authenticated data). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct OuterHeader { - pub receiver_idx: u32, + pub receiver_idx: u64, pub counter: u64, } impl OuterHeader { - pub const SIZE: usize = 12; // receiver_idx(4) + counter(8) + pub const SIZE: usize = 16; // receiver_idx(8) + counter(8) - pub fn new(receiver_idx: u32, counter: u64) -> Self { + pub fn new(receiver_idx: u64, counter: u64) -> Self { Self { receiver_idx, counter, @@ -34,15 +34,15 @@ impl OuterHeader { } #[allow(clippy::unwrap_used)] Ok(Self { - receiver_idx: u32::from_le_bytes(src[0..4].try_into().unwrap()), - counter: u64::from_le_bytes(src[4..12].try_into().unwrap()), + receiver_idx: u64::from_le_bytes(src[0..8].try_into().unwrap()), + counter: u64::from_le_bytes(src[8..16].try_into().unwrap()), }) } pub fn to_bytes(&self) -> [u8; Self::SIZE] { let mut bytes = [0u8; Self::SIZE]; - bytes[0..4].copy_from_slice(&self.receiver_idx.to_le_bytes()); - bytes[4..12].copy_from_slice(&self.counter.to_le_bytes()); + bytes[0..8].copy_from_slice(&self.receiver_idx.to_le_bytes()); + bytes[8..16].copy_from_slice(&self.counter.to_le_bytes()); bytes } @@ -130,7 +130,7 @@ pub struct LpHeader { impl LpHeader { pub fn new( - receiver_idx: u32, + receiver_idx: u64, counter: u64, protocol_version: u8, message_type: MessageType, @@ -159,7 +159,7 @@ impl LpHeader { } /// Get the sender index from the header - pub fn receiver_idx(&self) -> u32 { + pub fn receiver_idx(&self) -> u64 { self.outer.receiver_idx } } diff --git a/common/nym-lp/src/error.rs b/common/nym-lp/src/error.rs index 635db970f5..46883e4498 100644 --- a/common/nym-lp/src/error.rs +++ b/common/nym-lp/src/error.rs @@ -2,7 +2,6 @@ // SPDX-License-Identifier: Apache-2.0 use crate::replay::ReplayError; -use crate::session::SessionId; use libcrux_psq::handshake::HandshakeError; use libcrux_psq::handshake::builders::BuilderError; use libcrux_psq::session::SessionError; @@ -27,8 +26,8 @@ pub enum LpError { #[error("Attempted operation on closed session")] SessionClosed, - #[error("There already exists an LP session with id {0:?}")] - DuplicateSessionId(SessionId), + #[error("There already exists an LP session with receiver index {0}")] + DuplicateSessionId(u64), #[error("Internal error: {0}")] Internal(String), @@ -47,7 +46,7 @@ pub enum LpError { /// Session ID from incoming packet does not match any known session. #[error("Received packet with unknown session ID: {0}")] - UnknownSessionId(u32), + UnknownSessionId(u64), /// Invalid state transition attempt in the state machine. #[error("Invalid input '{input}' for current state '{state}'")] @@ -62,8 +61,8 @@ pub enum LpError { LpSessionProcessing, /// State machine not found. - #[error("State machine not found for lp_id: {lp_id:?}")] - StateMachineNotFound { lp_id: SessionId }, + #[error("State machine not found for lp_id: {lp_id}")] + StateMachineNotFound { lp_id: u64 }, /// Ed25519 to X25519 conversion error. #[error("Ed25519 key conversion error: {0}")] diff --git a/common/nym-lp/src/lib.rs b/common/nym-lp/src/lib.rs index c4475239e7..8e4c5de707 100644 --- a/common/nym-lp/src/lib.rs +++ b/common/nym-lp/src/lib.rs @@ -48,13 +48,16 @@ pub struct SessionsMock { impl SessionsMock { pub fn mock_seeded_post_handshake(seed: u64, kem: KEM) -> SessionsMock { use crate::peer::mock_peers; + use nym_test_utils::helpers::RngCore09; let (init, resp) = mock_peers(); let resp_remote = resp.as_remote(); - let init_rng = u64_seeded_rng_09(seed); + let mut init_rng = u64_seeded_rng_09(seed); let resp_rng = u64_seeded_rng_09(seed + 1); + let receiver_index = init_rng.next_u64(); + let kem_keys = resp.kem_keypairs.as_ref().unwrap(); // skip KKT by just deriving the kem key locally @@ -109,9 +112,20 @@ impl SessionsMock { }; SessionsMock { - initiator: LpSession::new(initiator.into_session().unwrap(), binding.clone(), 1) - .unwrap(), - responder: LpSession::new(responder.into_session().unwrap(), binding, 1).unwrap(), + initiator: LpSession::new( + initiator.into_session().unwrap(), + binding.clone(), + receiver_index, + 1, + ) + .unwrap(), + responder: LpSession::new( + responder.into_session().unwrap(), + binding, + receiver_index, + 1, + ) + .unwrap(), } } diff --git a/common/nym-lp/src/psq/initiator.rs b/common/nym-lp/src/psq/initiator.rs index 2004ee72a6..b1b387d48c 100644 --- a/common/nym-lp/src/psq/initiator.rs +++ b/common/nym-lp/src/psq/initiator.rs @@ -130,8 +130,11 @@ where &dir_hash, self.initiator_data.protocol_version, )?; + // derive the receiver index from the request + // let receiver_index = kkt_request + debug!("sending KKT request"); - self.send_kkt_request(kkt_request).await?; + self.send_kkt_request(kkt_request.request).await?; // 3. receive and process KKT response let raw_response = self.receive_kkt_response().await?; @@ -185,7 +188,7 @@ where }; let psq_session = psq_initiator.into_session()?; - LpSession::new(psq_session, binding, protocol) + LpSession::new(psq_session, binding, kkt_request.receiver_index, protocol) } } diff --git a/common/nym-lp/src/psq/mod.rs b/common/nym-lp/src/psq/mod.rs index 0aa9b89ed3..0de6bc8de7 100644 --- a/common/nym-lp/src/psq/mod.rs +++ b/common/nym-lp/src/psq/mod.rs @@ -238,7 +238,7 @@ mod tests { ) .unwrap(); - let processed_req = kkt_responder.process_request(request).unwrap(); + let processed_req = kkt_responder.process_request(request.request).unwrap(); let response = initiator.process_response(processed_req.response).unwrap(); let encapsulation_key = response.encapsulation_key; diff --git a/common/nym-lp/src/psq/responder.rs b/common/nym-lp/src/psq/responder.rs index 6cb7cb9048..e7295372f8 100644 --- a/common/nym-lp/src/psq/responder.rs +++ b/common/nym-lp/src/psq/responder.rs @@ -202,7 +202,12 @@ where }; let psq_session = psq_responder.into_session()?; - LpSession::new(psq_session, binding, processed_req.outer_protocol_version) + LpSession::new( + psq_session, + binding, + processed_req.receiver_index, + processed_req.outer_protocol_version, + ) } } @@ -267,7 +272,10 @@ mod tests { // 1. send kkt request conn_init - .send_handshake_message::(request.into(), kem) + .send_handshake_message::( + request.request.into(), + kem, + ) .timeboxed() .await??; diff --git a/common/nym-lp/src/session.rs b/common/nym-lp/src/session.rs index 336d28ed45..893dfae9d6 100644 --- a/common/nym-lp/src/session.rs +++ b/common/nym-lp/src/session.rs @@ -38,6 +38,9 @@ pub struct LpSession { // In the future it might get split between UDP and TCP transports active_transport: libcrux_psq::session::Transport, + /// Look-up index established during the initial KKT exchange + receiver_index: u64, + /// Negotiated protocol version from handshake. protocol_version: u8, @@ -102,6 +105,7 @@ impl LpSession { pub fn new( mut psq_session: Session, session_binding: PersistentSessionBinding, + receiver_index: u64, protocol_version: u8, ) -> Result { // attempt to derive initial transport @@ -113,6 +117,7 @@ impl LpSession { psq_session, session_binding, active_transport: transport, + receiver_index, protocol_version, sending_counter: 0, receiving_counter: Default::default(), @@ -163,10 +168,8 @@ impl LpSession { self.psq_session.identifier() } - pub fn receiver_index(&self) -> u32 { - let TODO = "this is temporary..."; - let id = self.psq_session.identifier(); - u32::from_le_bytes([id[0], id[1], id[2], id[3]]) + pub fn receiver_index(&self) -> u64 { + self.receiver_index } /// Returns the negotiated protocol version from the handshake. diff --git a/common/nym-lp/src/session_integration/mod.rs b/common/nym-lp/src/session_integration/mod.rs index 33fdf91b51..7612e791c5 100644 --- a/common/nym-lp/src/session_integration/mod.rs +++ b/common/nym-lp/src/session_integration/mod.rs @@ -33,11 +33,10 @@ mod tests { /// Tests simultaneous bidirectional communication between sessions #[test] fn test_bidirectional_communication() { - // 1. Initialize session manager - let mut session_manager_1 = SessionManager::new(); - let mut session_manager_2 = SessionManager::new(); - for kem in KEM::iter() { + // 1. Initialize session manager + let mut session_manager_1 = SessionManager::new(); + let mut session_manager_2 = SessionManager::new(); let sessions = SessionsMock::mock_post_handshake(kem); // 2. Create sessions using the pre-built Noise states @@ -122,9 +121,9 @@ mod tests { let mut session_manager = SessionManager::new(); let sessions = SessionsMock::mock_post_handshake(kem); - let session_id = *sessions.initiator.session_identifier(); + let session_id = sessions.initiator.receiver_index(); - let non_existent = [42u8; 32]; + let non_existent = 123; // sanity check in case of the 1 in 2^256 assert_ne!(session_id, non_existent); @@ -135,7 +134,7 @@ mod tests { let _session = session_manager.create_session_state_machine(session1); // 3. Try to get a non-existent session - let result = session_manager.state_machine_exists(&non_existent); + let result = session_manager.state_machine_exists(non_existent); assert!(!result, "Non-existent session should return None"); // 4. Try to remove a non-existent session @@ -169,7 +168,7 @@ mod tests { for kem in KEM::iter() { let sessions = SessionsMock::mock_post_handshake(kem); - let session_id = *sessions.responder.session_identifier(); + let session_id = sessions.responder.receiver_index(); // 2. Create sessions state machines session_manager_1 @@ -181,8 +180,8 @@ mod tests { assert_eq!(session_manager_1.session_count(), 1); assert_eq!(session_manager_2.session_count(), 1); - assert!(session_manager_1.state_machine_exists(&session_id)); - assert!(session_manager_2.state_machine_exists(&session_id)); + assert!(session_manager_1.state_machine_exists(session_id)); + assert!(session_manager_2.state_machine_exists(session_id)); // Verify initial states are Transport assert_eq!( @@ -379,13 +378,13 @@ mod tests { // --- 7. Session Removal --- assert!(session_manager_1.remove_state_machine(session_id)); assert_eq!(session_manager_1.session_count(), 0); - assert!(!session_manager_1.state_machine_exists(&session_id)); + assert!(!session_manager_1.state_machine_exists(session_id)); // B's session manager still has it until removed - assert!(session_manager_2.state_machine_exists(&session_id)); + assert!(session_manager_2.state_machine_exists(session_id)); assert!(session_manager_2.remove_state_machine(session_id)); assert_eq!(session_manager_2.session_count(), 0); - assert!(!session_manager_2.state_machine_exists(&session_id)); + assert!(!session_manager_2.state_machine_exists(session_id)); println!("Session removal test passed."); } } diff --git a/common/nym-lp/src/session_manager.rs b/common/nym-lp/src/session_manager.rs index a23164bd65..1c987501af 100644 --- a/common/nym-lp/src/session_manager.rs +++ b/common/nym-lp/src/session_manager.rs @@ -6,7 +6,6 @@ //! This module implements session lifecycle management functionality, handling //! creation, retrieval, and storage of sessions. -use crate::session::SessionId; use crate::state_machine::{LpAction, LpData, LpInput, LpStateBare}; use crate::{LpError, LpSession, LpStateMachine}; use nym_lp_packet::EncryptedLpPacket; @@ -19,7 +18,7 @@ pub use crate::replay::validator::PacketCount; /// The SessionManager is responsible for creating, storing, and retrieving sessions pub struct SessionManager { /// Manages state machines directly, keyed by lp_id - state_machines: HashMap, + state_machines: HashMap, } impl Default for SessionManager { @@ -38,43 +37,43 @@ impl SessionManager { pub fn process_input( &mut self, - lp_id: SessionId, + lp_id: u64, input: LpInput, ) -> Result, LpError> { self.with_state_machine_mut(lp_id, |sm| sm.process_input(input).transpose())? } - pub fn send_data(&mut self, lp_id: SessionId, data: LpData) -> Result { + pub fn send_data(&mut self, lp_id: u64, data: LpData) -> Result { self.process_input(lp_id, LpInput::SendData(data))? .ok_or(LpError::NotInTransport) } pub fn receive_packet( &mut self, - lp_id: SessionId, + lp_id: u64, packet: EncryptedLpPacket, ) -> Result, LpError> { self.process_input(lp_id, LpInput::ReceivePacket(packet)) } - pub fn closed(&self, lp_id: SessionId) -> Result { + pub fn closed(&self, lp_id: u64) -> Result { Ok(self.get_state(lp_id)? == LpStateBare::Closed) } - pub fn transport(&self, lp_id: SessionId) -> Result { + pub fn transport(&self, lp_id: u64) -> Result { Ok(self.get_state(lp_id)? == LpStateBare::Transport) } #[cfg(test)] - fn get_state_machine_id(&self, lp_id: SessionId) -> Result { - self.with_state_machine(lp_id, |sm| sm.session_identifier())? + fn get_state_machine_id(&self, lp_id: u64) -> Result { + self.with_state_machine(lp_id, |sm| sm.receiver_index())? } - pub fn get_state(&self, lp_id: SessionId) -> Result { + pub fn get_state(&self, lp_id: u64) -> Result { self.with_state_machine(lp_id, |sm| Ok(sm.bare_state()))? } - pub fn current_packet_cnt(&self, lp_id: SessionId) -> Result { + pub fn current_packet_cnt(&self, lp_id: u64) -> Result { self.with_state_machine(lp_id, |sm| Ok(sm.session()?.current_packet_cnt()))? } @@ -82,11 +81,11 @@ impl SessionManager { self.state_machines.len() } - pub fn state_machine_exists(&self, lp_id: &SessionId) -> bool { - self.state_machines.contains_key(lp_id) + pub fn state_machine_exists(&self, lp_id: u64) -> bool { + self.state_machines.contains_key(&lp_id) } - pub fn with_state_machine(&self, lp_id: SessionId, f: F) -> Result + pub fn with_state_machine(&self, lp_id: u64, f: F) -> Result where F: FnOnce(&LpStateMachine) -> R, { @@ -98,7 +97,7 @@ impl SessionManager { } // For mutable access (like running process_input) - pub fn with_state_machine_mut(&mut self, lp_id: SessionId, f: F) -> Result + pub fn with_state_machine_mut(&mut self, lp_id: u64, f: F) -> Result where F: FnOnce(&mut LpStateMachine) -> R, // Closure takes mutable ref { @@ -109,11 +108,8 @@ impl SessionManager { } } - pub fn create_session_state_machine( - &mut self, - lp_session: LpSession, - ) -> Result { - let session_id = *lp_session.session_identifier(); + pub fn create_session_state_machine(&mut self, lp_session: LpSession) -> Result { + let session_id = lp_session.receiver_index(); if self.state_machines.contains_key(&session_id) { return Err(LpError::DuplicateSessionId(session_id)); @@ -125,7 +121,7 @@ impl SessionManager { } /// Method to remove a state machine - pub fn remove_state_machine(&mut self, lp_id: SessionId) -> bool { + pub fn remove_state_machine(&mut self, lp_id: u64) -> bool { let removed = self.state_machines.remove(&lp_id); removed.is_some() @@ -142,15 +138,15 @@ mod tests { fn test_session_manager_get() { let mut manager = SessionManager::new(); let local_session = mock_session_for_test(); - let id = *local_session.session_identifier(); + let id = local_session.receiver_index(); let sm_1_id = manager.create_session_state_machine(local_session).unwrap(); assert_eq!(sm_1_id, id); - let retrieved = manager.state_machine_exists(&id); + let retrieved = manager.state_machine_exists(id); assert!(retrieved); - let not_found = manager.state_machine_exists(&[99u8; 32]); + let not_found = manager.state_machine_exists(123); assert!(!not_found); } diff --git a/common/nym-lp/src/state_machine.rs b/common/nym-lp/src/state_machine.rs index 8d0fbbfa4f..fc6133f1bc 100644 --- a/common/nym-lp/src/state_machine.rs +++ b/common/nym-lp/src/state_machine.rs @@ -186,6 +186,10 @@ impl LpStateMachine { Ok(*self.session()?.session_identifier()) } + pub fn receiver_index(&self) -> Result { + Ok(self.session()?.receiver_index()) + } + /// Creates a new state machine in `Transport` state post-KKT/PSQ handshake pub fn new(session: LpSession) -> Self { LpStateMachine { diff --git a/gateway/src/node/lp_listener/error.rs b/gateway/src/node/lp_listener/error.rs index 97a5936cd7..dc45852254 100644 --- a/gateway/src/node/lp_listener/error.rs +++ b/gateway/src/node/lp_listener/error.rs @@ -1,6 +1,7 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::node::lp_listener::ReceiverIndex; use nym_lp::state_machine::{LpAction, LpDataKind}; use nym_lp::{LpError, MalformedLpPacketError}; use nym_lp_transport::LpTransportError; @@ -16,11 +17,20 @@ pub enum LpHandlerError { LpTransportError(#[from] LpTransportError), #[error("missing session state for {receiver_index} - has it been removed due to inactivity?")] - MissingLpSession { receiver_index: u32 }, + MissingLpSession { receiver_index: ReceiverIndex }, #[error(transparent)] LpProtocolError(#[from] LpError), + #[error("the initial KKT/PSQ handshake has not been completed")] + IncompleteHandshake, + + #[error("receiver_idx mismatch: connection bound to {established}, packet has {received}")] + MismatchedReceiverIndex { + established: ReceiverIndex, + received: ReceiverIndex, + }, + #[error("no action has been emitted from the LP State Machine")] UnexpectedStateMachineHalt, diff --git a/gateway/src/node/lp_listener/handler.rs b/gateway/src/node/lp_listener/handler.rs index 6d31a6ac41..63d84ea544 100644 --- a/gateway/src/node/lp_listener/handler.rs +++ b/gateway/src/node/lp_listener/handler.rs @@ -3,10 +3,9 @@ use super::{LpHandlerState, ReceiverIndex, TimestampedState}; use crate::node::lp_listener::error::LpHandlerError; -use nym_crypto::asymmetric::x25519; use nym_lp::state_machine::{LpAction, LpData, LpDataKind, LpInput}; use nym_lp::{ - EncryptedLpPacket, ExpectedResponseSize, ForwardPacketData, LpPacket, LpSession, LpStateMachine, + EncryptedLpPacket, ExpectedResponseSize, ForwardPacketData, LpSession, LpStateMachine, }; use nym_lp_transport::traits::LpTransportChannel; use nym_lp_transport::LpHandshakeChannel; @@ -14,7 +13,6 @@ use nym_metrics::{add_histogram_obs, inc}; use nym_registration_common::{LpRegistrationRequest, RegistrationStatus}; use std::net::SocketAddr; use std::time::Duration; -use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::net::TcpStream; use tokio::time::timeout; use tracing::*; @@ -85,7 +83,7 @@ pub struct LpConnectionHandler { /// Bound receiver_idx for this connection (set after first packet). /// All subsequent packets on this connection must use this receiver_idx. /// Set from ClientHello's proposed receiver_index, or from header for non-bootstrap packets. - bound_receiver_idx: Option, + bound_receiver_idx: Option, /// Persistent connection to exit gateway for forwarding. /// Opened on first forward, reused for subsequent forwards, closed when client disconnects. @@ -208,28 +206,26 @@ where Ok(()) } - // fn bound_receiver_index(&self) -> Result { - // self.bound_receiver_idx.ok_or_else(|| { - // LpHandlerError::LpProtocolError( - // "missing bound receiver index after KKT/PSQ handshake".into(), - // ) - // }) - // } + fn bound_receiver_index(&self) -> Result { + self.bound_receiver_idx + .ok_or_else(|| LpHandlerError::IncompleteHandshake) + } /// Validate that the receiver_idx matches the bound session. - fn validate_binding(&self, receiver_idx: u32) -> Result<(), LpHandlerError> { - // let bound_receiver_idx = self.bound_receiver_index()?; - // - // if bound_receiver_idx != receiver_idx { - // warn!( - // "Receiver_idx mismatch from {}: expected {bound_receiver_idx}, got {receiver_idx}", - // self.remote_addr - // ); - // inc!("lp_errors_receiver_idx_mismatch"); - // return Err(LpHandlerError::LpProtocolError(format!( - // "receiver_idx mismatch: connection bound to {bound_receiver_idx}, packet has {receiver_idx}", - // ))); - // } + fn validate_binding(&self, receiver_idx: ReceiverIndex) -> Result<(), LpHandlerError> { + let bound_receiver_idx = self.bound_receiver_index()?; + + if bound_receiver_idx != receiver_idx { + warn!( + "Receiver_idx mismatch from {}: expected {bound_receiver_idx}, got {receiver_idx}", + self.remote_addr + ); + inc!("lp_errors_receiver_idx_mismatch"); + return Err(LpHandlerError::MismatchedReceiverIndex { + established: bound_receiver_idx, + received: receiver_idx, + }); + } Ok(()) } @@ -296,7 +292,7 @@ where /// Handle decrypted transport payload (registration or forwarding request) async fn handle_decrypted_payload( &mut self, - receiver_idx: u32, + receiver_idx: ReceiverIndex, decrypted_data: LpData, ) -> Result<(), LpHandlerError> { let remote = self.remote_addr; @@ -332,7 +328,7 @@ where /// Attempt to wrap and send specified response back to the client async fn send_response_packet( &mut self, - receiver_index: u32, + receiver_index: ReceiverIndex, serialised_response: Vec, response_kind: LpDataKind, ) -> Result<(), LpHandlerError> { @@ -406,7 +402,7 @@ where /// to exit gateway, receives response, encrypts it, and sends back to client. async fn handle_forwarding_request( &mut self, - receiver_idx: u32, + receiver_idx: ReceiverIndex, forward_data: ForwardPacketData, ) -> Result<(), LpHandlerError> { // Forward the packet to the target gateway and retrieve its response diff --git a/gateway/src/node/lp_listener/mod.rs b/gateway/src/node/lp_listener/mod.rs index 23b9e3c6c9..cfae5884ee 100644 --- a/gateway/src/node/lp_listener/mod.rs +++ b/gateway/src/node/lp_listener/mod.rs @@ -95,7 +95,7 @@ pub mod error; pub mod handler; mod registration; -pub type ReceiverIndex = u32; +pub type ReceiverIndex = u64; /// Configuration for LP listener #[derive(Debug, Clone, Copy, serde::Deserialize, serde::Serialize)] @@ -565,7 +565,7 @@ impl LpListener { } pub(crate) mod cleanup_task { - use crate::node::lp_listener::{LpDebug, TimestampedState}; + use crate::node::lp_listener::{LpDebug, ReceiverIndex, TimestampedState}; use dashmap::DashMap; use nym_lp::state_machine::LpStateBare; use nym_lp::LpStateMachine; @@ -575,7 +575,7 @@ pub(crate) mod cleanup_task { use tracing::{debug, info}; async fn perform_cleanup( - session_states: &Arc>>, + session_states: &Arc>>, cfg: LpDebug, ) { let session_ttl = cfg.session_ttl; @@ -616,7 +616,7 @@ pub(crate) mod cleanup_task { /// Demoted sessions (ReadOnlyTransport) use shorter TTL since they /// only need to drain in-flight packets after subsession promotion. pub(crate) async fn cleanup_loop( - session_states: Arc>>, + session_states: Arc>>, cfg: LpDebug, shutdown: nym_task::ShutdownToken, _metrics: NymNodeMetrics,