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

use derived receiver index

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2026-02-24 16:23:26 +00:00
parent 211f90692a
commit 7c23cd2183
17 changed files with 180 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nym-kkt/src/frame.rs
+20 -12
View File
@@ -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<KKTRequestEncryptionResult, KKTError>
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<u64, KKTError> {
let mut receiver_index_bytes: [u8; 8] = [0u8; 8];
fn derive_receiver_index(
&self,
kkt_outer_headers: &KKTRequestPlaintext,
) -> Result<u64, KKTError> {
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))
common/nym-kkt/src/initiator.rs
+15 -6
View File
@@ -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<R>(
@@ -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,
},
))
}
common/nym-kkt/src/message.rs
+15 -1
View File
@@ -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,
}
common/nym-kkt/src/responder.rs
+1
View File
@@ -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,
})
}
common/nym-lp-packet/src/header.rs
+9 -9
View File
@@ -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
}
}
common/nym-lp/src/error.rs
+5 -6
View File
@@ -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}")]
common/nym-lp/src/lib.rs
+18 -4
View File
@@ -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(),
}
}
common/nym-lp/src/psq/initiator.rs
+5 -2
View File
@@ -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)
}
}
common/nym-lp/src/psq/mod.rs
+1 -1
View File
@@ -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;
common/nym-lp/src/psq/responder.rs
+10 -2
View File
@@ -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::<handshake_message::KKTRequest>(request.into(), kem)
.send_handshake_message::<handshake_message::KKTRequest>(
request.request.into(),
kem,
)
.timeboxed()
.await??;
common/nym-lp/src/session.rs
+7 -4
View File
@@ -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<Self, LpError> {
// 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.
common/nym-lp/src/session_integration/mod.rs
+12 -13
View File
@@ -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.");
}
}
common/nym-lp/src/session_manager.rs
+20 -24
View File
@@ -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<SessionId, LpStateMachine>,
state_machines: HashMap<u64, LpStateMachine>,
}
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<Option<LpAction>, 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<LpAction, LpError> {
pub fn send_data(&mut self, lp_id: u64, data: LpData) -> Result<LpAction, LpError> {
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<Option<LpAction>, LpError> {
self.process_input(lp_id, LpInput::ReceivePacket(packet))
}
pub fn closed(&self, lp_id: SessionId) -> Result<bool, LpError> {
pub fn closed(&self, lp_id: u64) -> Result<bool, LpError> {
Ok(self.get_state(lp_id)? == LpStateBare::Closed)
}
pub fn transport(&self, lp_id: SessionId) -> Result<bool, LpError> {
pub fn transport(&self, lp_id: u64) -> Result<bool, LpError> {
Ok(self.get_state(lp_id)? == LpStateBare::Transport)
}
#[cfg(test)]
fn get_state_machine_id(&self, lp_id: SessionId) -> Result<SessionId, LpError> {
self.with_state_machine(lp_id, |sm| sm.session_identifier())?
fn get_state_machine_id(&self, lp_id: u64) -> Result<u64, LpError> {
self.with_state_machine(lp_id, |sm| sm.receiver_index())?
}
pub fn get_state(&self, lp_id: SessionId) -> Result<LpStateBare, LpError> {
pub fn get_state(&self, lp_id: u64) -> Result<LpStateBare, LpError> {
self.with_state_machine(lp_id, |sm| Ok(sm.bare_state()))?
}
pub fn current_packet_cnt(&self, lp_id: SessionId) -> Result<PacketCount, LpError> {
pub fn current_packet_cnt(&self, lp_id: u64) -> Result<PacketCount, LpError> {
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<F, R>(&self, lp_id: SessionId, f: F) -> Result<R, LpError>
pub fn with_state_machine<F, R>(&self, lp_id: u64, f: F) -> Result<R, LpError>
where
F: FnOnce(&LpStateMachine) -> R,
{
@@ -98,7 +97,7 @@ impl SessionManager {
}
// For mutable access (like running process_input)
pub fn with_state_machine_mut<F, R>(&mut self, lp_id: SessionId, f: F) -> Result<R, LpError>
pub fn with_state_machine_mut<F, R>(&mut self, lp_id: u64, f: F) -> Result<R, LpError>
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<SessionId, LpError> {
let session_id = *lp_session.session_identifier();
pub fn create_session_state_machine(&mut self, lp_session: LpSession) -> Result<u64, LpError> {
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);
}
common/nym-lp/src/state_machine.rs
+4
View File
@@ -186,6 +186,10 @@ impl LpStateMachine {
Ok(*self.session()?.session_identifier())
}
pub fn receiver_index(&self) -> Result<u64, LpError> {
Ok(self.session()?.receiver_index())
}
/// Creates a new state machine in `Transport` state post-KKT/PSQ handshake
pub fn new(session: LpSession) -> Self {
LpStateMachine {
gateway/src/node/lp_listener/error.rs
+11 -1
View File
@@ -1,6 +1,7 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// 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,
gateway/src/node/lp_listener/handler.rs
+23 -27
View File
@@ -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<S = TcpStream> {
/// 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<u32>,
bound_receiver_idx: Option<ReceiverIndex>,
/// 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<u32, LpHandlerError> {
// self.bound_receiver_idx.ok_or_else(|| {
// LpHandlerError::LpProtocolError(
// "missing bound receiver index after KKT/PSQ handshake".into(),
// )
// })
// }
fn bound_receiver_index(&self) -> Result<ReceiverIndex, LpHandlerError> {
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<u8>,
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
gateway/src/node/lp_listener/mod.rs
+4 -4
View File
@@ -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<DashMap<u32, TimestampedState<LpStateMachine>>>,
session_states: &Arc<DashMap<ReceiverIndex, TimestampedState<LpStateMachine>>>,
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<DashMap<u32, TimestampedState<LpStateMachine>>>,
session_states: Arc<DashMap<ReceiverIndex, TimestampedState<LpStateMachine>>>,
cfg: LpDebug,
shutdown: nym_task::ShutdownToken,
_metrics: NymNodeMetrics,