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

Merge pull request #6526 from nymtech/chore/lp-improvements

Browse Source 
chore: LP improvements
This commit is contained in:
Jędrzej Stuczyński
2026-03-09 10:57:26 +00:00
committed by GitHub
parent 03d5a87826 ef65cf4c9e
commit 1fd17c5cb3
31 changed files with 1191 additions and 385 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -6833,6 +6833,7 @@ dependencies = [
"nym-kkt-ciphersuite",
"nym-kkt-context",
"nym-pemstore",
"nym-test-utils",
"rand 0.9.2",
"rand_chacha 0.9.0",
"strum 0.28.0",
common/nym-kkt/Cargo.toml
+1
View File
@@ -30,6 +30,7 @@ libcrux-ml-kem = { workspace = true }
[dev-dependencies]
rand_chacha = "0.9.0"
anyhow = { workspace = true }
nym-test-utils = { workspace = true }
[lints]
common/nym-kkt/src/lib.rs
+122 -59
View File
@@ -16,10 +16,6 @@ pub use nym_kkt_context as context;
#[cfg(test)]
mod test {
use nym_kkt_ciphersuite::{Ciphersuite, HashFunction, HashLength, KEM, SignatureScheme};
use rand09::RngCore;
use std::collections::BTreeMap;
use crate::keys::KEMKeys;
use crate::{
initiator::KKTInitiator,
@@ -29,9 +25,13 @@ mod test {
},
responder::KKTResponder,
};
use nym_kkt_ciphersuite::{Ciphersuite, HashFunction, HashLength, KEM, SignatureScheme};
use nym_test_utils::helpers::deterministic_rng_09;
use rand09::RngCore;
use std::collections::BTreeMap;
#[test]
fn test_kkt_psq_e2e_encrypted_carrier() {
fn test_kkt_psq_e2e_one_way_encrypted_carrier() {
let mut rng = rand09::rng();
let mut payload: Vec<u8> = vec![0u8; 900_000];
@@ -47,7 +47,6 @@ mod test {
HashFunction::Shake256,
] {
// generate kem public keys
let responder_mlkem_keypair = generate_keypair_mlkem(&mut rng);
let responder_mceliece_keypair = generate_keypair_mceliece(&mut rng);
@@ -64,20 +63,6 @@ mod test {
HashLength::Default.value(),
responder_kem.mc_eliece_encapsulation_key().as_ref(),
);
let initiator_mlkem_keypair = generate_keypair_mlkem(&mut rng);
let initiator_mceliece_keypair = generate_keypair_mceliece(&mut rng);
let _i_dir_hash_mlkem = hash_encapsulation_key(
hash_function,
HashLength::Default.value(),
initiator_mlkem_keypair.public_key().as_slice(),
);
let _i_dir_hash_mceliece = hash_encapsulation_key(
hash_function,
HashLength::Default.value(),
initiator_mceliece_keypair.pk.as_ref(),
);
let init_hashes = BTreeMap::new();
@@ -128,41 +113,6 @@ mod test {
responder_kem.ml_kem768_encapsulation_key().as_slice(),
)
}
// Mutual - MlKem
{
let ciphersuite = Ciphersuite::resolve_ciphersuite(
KEM::MlKem768,
hash_function,
SignatureScheme::Ed25519,
None,
)
.unwrap();
let (mut initiator, request) = KKTInitiator::generate_one_way_request(
&mut rng,
ciphersuite,
&responder_x25519_keypair.pk,
&r_dir_hash_mlkem,
1u8,
Some(payload.clone()),
)
.unwrap();
let processed_request = responder.process_request(request, payload.len()).unwrap();
assert_eq!(processed_request.request_payload, payload);
// if we keep unverified keys, this should change
assert!(processed_request.remote_encapsulation_key.is_none());
let processed_response = initiator
.process_response(processed_request.response, 0)
.unwrap();
assert_eq!(
processed_response.encapsulation_key.as_bytes(),
responder_kem.ml_kem768_encapsulation_key().as_slice(),
)
}
// OneWay - McEliece
{
@@ -195,7 +145,110 @@ mod test {
responder_kem.mc_eliece_encapsulation_key().as_ref()
)
}
}
}
#[test]
fn test_kkt_psq_e2e_mutual_encrypted_carrier() {
let mut rng = deterministic_rng_09();
let mut payload: Vec<u8> = vec![0u8; 50000];
rng.fill_bytes(&mut payload);
// generate kem public keys
let initiator_mlkem_keypair = generate_keypair_mlkem(&mut rng);
let initiator_mceliece_keypair = generate_keypair_mceliece(&mut rng);
let responder_mlkem_keypair = generate_keypair_mlkem(&mut rng);
let responder_mceliece_keypair = generate_keypair_mceliece(&mut rng);
let responder_x25519_keypair = generate_lp_keypair_x25519(&mut rng);
let initiator_kem = KEMKeys::new(initiator_mceliece_keypair, initiator_mlkem_keypair);
let responder_kem = KEMKeys::new(responder_mceliece_keypair, responder_mlkem_keypair);
let init_hashes = initiator_kem.encapsulation_keys_digests();
let responder = KKTResponder::new(
&responder_x25519_keypair,
&responder_kem,
&init_hashes,
&[
HashFunction::Blake3,
HashFunction::SHA256,
HashFunction::Shake128,
HashFunction::Shake256,
],
&[SignatureScheme::Ed25519],
&[1],
)
.unwrap();
for hash_function in [
HashFunction::Blake3,
HashFunction::SHA256,
HashFunction::Shake128,
HashFunction::Shake256,
] {
let r_dir_hash_mlkem = hash_encapsulation_key(
hash_function,
HashLength::Default.value(),
responder_kem.ml_kem768_encapsulation_key().as_slice(),
);
let r_dir_hash_mceliece = hash_encapsulation_key(
hash_function,
HashLength::Default.value(),
responder_kem.mc_eliece_encapsulation_key().as_ref(),
);
// Mutual - MlKem
{
let ciphersuite = Ciphersuite::resolve_ciphersuite(
KEM::MlKem768,
hash_function,
SignatureScheme::Ed25519,
None,
)
.unwrap();
let (mut initiator, request) = KKTInitiator::generate_mutual_request(
&mut rng,
ciphersuite,
initiator_kem
.encoded_encapsulation_key(KEM::MlKem768)
.unwrap(),
&responder_x25519_keypair.pk,
&r_dir_hash_mlkem,
1u8,
Some(payload.clone()),
)
.unwrap();
let processed_request = responder.process_request(request, payload.len()).unwrap();
assert_eq!(processed_request.request_payload, payload);
assert_eq!(
processed_request
.remote_encapsulation_key
.unwrap()
.as_bytes(),
initiator_kem
.encapsulation_key(KEM::MlKem768)
.unwrap()
.as_bytes()
);
let processed_response = initiator
.process_response(processed_request.response, 0)
.unwrap();
assert_eq!(
processed_response.encapsulation_key.as_bytes(),
responder_kem.ml_kem768_encapsulation_key().as_slice(),
)
}
// Mutual - McEliece is not supported due to the key being too large
{
let ciphersuite = Ciphersuite::resolve_ciphersuite(
KEM::McEliece,
@@ -204,9 +257,12 @@ mod test {
None,
)
.unwrap();
let (mut initiator, request) = KKTInitiator::generate_one_way_request(
let (mut initiator, request) = KKTInitiator::generate_mutual_request(
&mut rng,
ciphersuite,
initiator_kem
.encoded_encapsulation_key(KEM::McEliece)
.unwrap(),
&responder_x25519_keypair.pk,
&r_dir_hash_mceliece,
1u8,
@@ -217,9 +273,16 @@ mod test {
let processed_request = responder.process_request(request, payload.len()).unwrap();
assert_eq!(processed_request.request_payload, payload);
// if we keep unverified keys, this should change
assert!(processed_request.remote_encapsulation_key.is_none());
assert_eq!(
processed_request
.remote_encapsulation_key
.unwrap()
.as_bytes(),
initiator_kem
.encapsulation_key(KEM::McEliece)
.unwrap()
.as_bytes()
);
let processed_response = initiator
.process_response(processed_request.response, 0)
common/nym-lp/src/peer.rs
+4
View File
@@ -143,6 +143,10 @@ impl LpRemotePeer {
.ok_or(LpError::NoKnownKEMKeyDigests { kem, hash_function })
.cloned()
}
pub fn kem_key_digests(&self) -> &BTreeMap<KEM, KEMKeyDigests> {
&self.expected_kem_key_digests
}
}
impl From<DHPublicKey> for LpRemotePeer {
common/nym-lp/src/peer_config.rs
+5
View File
@@ -65,6 +65,7 @@ impl LpPeerConfig {
rng.random(),
)
}
/// Creates a new client to exit config.
/// Inputs:
/// hop_id: this value must be in the range (1..=15). This function returns an error if this is not the case.
@@ -79,6 +80,7 @@ impl LpPeerConfig {
{
Self::new(rng, hop_id, true, false, censorship_resistance)
}
/// Creates a new client to an intermediate node config.
/// Inputs:
/// hop_id: this value must be in the range (1..=14). This function returns an error if this is not the case.
@@ -130,6 +132,7 @@ impl LpPeerConfig {
rng.random(),
)
}
fn build(
hop_id: u8,
is_exit: bool,
@@ -147,6 +150,7 @@ impl LpPeerConfig {
seed,
}
}
fn build_checked(
hop_id: u8,
is_exit: bool,
@@ -203,6 +207,7 @@ impl LpPeerConfig {
output_bytes[4..].copy_from_slice(&self.seed);
output_bytes
}
pub fn deserialize(bytes: &[u8]) -> Result<Self, LpError> {
if bytes.len() != LP_PEER_CONFIG_SIZE {
return Err(LpError::DeserializationError(format!(
common/nym-lp/src/psq/initiator.rs
+43 -13
View File
@@ -24,10 +24,31 @@ use nym_kkt::message::{KKTRequest, KKTResponse};
use rand09::SeedableRng;
use tracing::debug;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HandshakeMode {
// Client <> Entry
OneWayEntry,
// Client <> Exit
OneWayExit,
// Entry <> Exit
MutualInternode,
// in the future more variants will be supported (such as individual mix hops)
}
impl HandshakeMode {
pub fn is_mutual(&self) -> bool {
matches!(self, HandshakeMode::MutualInternode)
}
}
pub struct PSQHandshakeStateInitiator<'a, S> {
pub(super) inner_state: PSQHandshakeState<'a, S>,
pub(super) initiator_data: InitiatorData,
pub(super) mutual: bool,
/// The mode of the handshake (mutual node-node, client-entry, entry-exit)
pub(super) mode: HandshakeMode,
}
pub(crate) fn build_psq_principal<R>(
@@ -78,13 +99,23 @@ impl<'a, S> PSQHandshakeStateInitiator<'a, S>
where
S: LpHandshakeChannel + Unpin,
{
pub fn set_mutual_kkt(mut self) -> Result<Self, LpError> {
if self.inner_state.local_peer.kem_keypairs.is_none() {
return Err(LpError::PSQMutualInitiatorMissingKemKey);
}
fn lp_peer_config<R>(&self, rng: &mut R) -> Result<LpPeerConfig, LpError>
where
R: rand09::CryptoRng,
{
// for now we don't support censorship resistance flag
let censorship_resistance = false;
self.mutual = true;
Ok(self)
match self.mode {
HandshakeMode::OneWayEntry => Ok(LpPeerConfig::new_client_to_entry(
rng,
censorship_resistance,
)),
HandshakeMode::OneWayExit => {
LpPeerConfig::new_client_to_exit(rng, 1, censorship_resistance)
}
HandshakeMode::MutualInternode => LpPeerConfig::new_node_to_node(rng),
}
}
/// Attempt to send KKT request to begin the handshake
@@ -132,7 +163,7 @@ where
let ciphersuite = self.inner_state.local_peer.ciphersuite();
let kem = ciphersuite.kem();
let lp_peer_config = LpPeerConfig::new_client_to_entry(rng, false);
let lp_peer_config = self.lp_peer_config(rng)?;
// 1. retrieve the expected kem key hash. if we don't know it,
let dir_hash = self
@@ -141,7 +172,7 @@ where
.expected_kem_key_hash(ciphersuite)?;
// 2. prepare and send KKT request
let (mut initiator, kkt_request) = if self.mutual {
let (mut initiator, kkt_request) = if self.mode.is_mutual() {
// this has been verified when setting the mutual flag
let Some(local_encapsulation_key) = self.inner_state.local_peer.encoded_kem_key(kem)
else {
@@ -273,8 +304,8 @@ mod tests {
resp.ciphersuite = ciphersuite;
let initiator_data = InitiatorData::new(1, resp_remote);
let handshake_init =
PSQHandshakeState::new(conn_init, init).as_initiator(initiator_data);
let handshake_init = PSQHandshakeState::new(conn_init, init)
.as_initiator(initiator_data, HandshakeMode::OneWayEntry)?;
let mut init_rng = DeterministicRng09Send::new(u64_seeded_rng_09(1));
@@ -396,8 +427,7 @@ mod tests {
let initiator_data = InitiatorData::new(1, resp_remote);
let handshake_init = PSQHandshakeState::new(conn_init, init)
.as_initiator(initiator_data)
.set_mutual_kkt()?;
.as_initiator(initiator_data, HandshakeMode::MutualInternode)?;
let mut init_rng = DeterministicRng09Send::new(u64_seeded_rng_09(1));
common/nym-lp/src/psq/mod.rs
+22 -9
View File
@@ -12,6 +12,8 @@ mod helpers;
pub mod initiator;
pub mod responder;
use crate::LpError;
use crate::psq::initiator::HandshakeMode;
pub use initiator::PSQHandshakeStateInitiator;
pub use responder::PSQHandshakeStateResponder;
@@ -107,12 +109,20 @@ where
}
}
pub fn as_initiator(self, initiator_data: InitiatorData) -> PSQHandshakeStateInitiator<'a, S> {
PSQHandshakeStateInitiator {
pub fn as_initiator(
self,
initiator_data: InitiatorData,
mode: HandshakeMode,
) -> Result<PSQHandshakeStateInitiator<'a, S>, LpError> {
if mode.is_mutual() && self.local_peer.kem_keypairs.is_none() {
return Err(LpError::PSQMutualInitiatorMissingKemKey);
}
Ok(PSQHandshakeStateInitiator {
initiator_data,
inner_state: self,
mutual: false,
}
mode,
})
}
pub fn as_responder(self, responder_data: ResponderData) -> PSQHandshakeStateResponder<'a, S> {
@@ -160,8 +170,10 @@ mod tests {
resp.ciphersuite = ciphersuite;
let resp_remote = resp.as_remote();
let handshake_init = PSQHandshakeState::new(conn_init, init)
.as_initiator(InitiatorData::new(1, resp_remote));
let handshake_init = PSQHandshakeState::new(conn_init, init).as_initiator(
InitiatorData::new(1, resp_remote),
HandshakeMode::OneWayEntry,
)?;
let handshake_resp =
PSQHandshakeState::new(conn_resp, resp).as_responder(ResponderData::default());
@@ -232,9 +244,10 @@ mod tests {
let resp_remote = resp.as_remote();
let init_remote = init.as_remote();
let handshake_init = PSQHandshakeState::new(conn_init, init)
.as_initiator(InitiatorData::new(1, resp_remote))
.set_mutual_kkt()?;
let handshake_init = PSQHandshakeState::new(conn_init, init).as_initiator(
InitiatorData::new(1, resp_remote),
HandshakeMode::MutualInternode,
)?;
let handshake_resp = PSQHandshakeState::new(conn_resp, resp).as_responder(
ResponderData::default()
.with_initiator_kem_hashes(init_remote.expected_kem_key_digests),
common/nym-lp/src/session.rs
+41 -3
View File
@@ -9,6 +9,7 @@ use crate::codec::{decrypt_lp_packet, encrypt_lp_packet};
use crate::packet::{EncryptedLpPacket, LpHeader, LpMessage, LpPacket};
use crate::peer::{LpLocalPeer, LpRemotePeer};
use crate::peer_config::LpReceiverIndex;
use crate::psq::initiator::HandshakeMode;
use crate::psq::{
InitiatorData, PSQHandshakeState, PSQHandshakeStateInitiator, PSQHandshakeStateResponder,
ResponderData,
@@ -19,6 +20,8 @@ use crate::{LpError, replay::ReceivingKeyCounterValidator};
use libcrux_psq::handshake::types::{Authenticator, DHPublicKey};
use libcrux_psq::session::{Session, SessionBinding};
use nym_kkt::keys::EncapsulationKey;
use nym_kkt_ciphersuite::{KEM, KEMKeyDigests};
use std::collections::BTreeMap;
use std::fmt::{Debug, Formatter};
/// Represents inputs that drive the state machine transitions.
@@ -154,12 +157,34 @@ impl LpTransportSession {
local_peer: LpLocalPeer,
remote_peer: LpRemotePeer,
remote_protocol_version: u8,
) -> PSQHandshakeStateInitiator<'_, S>
mode: HandshakeMode,
) -> Result<PSQHandshakeStateInitiator<'_, S>, LpError>
where
S: LpHandshakeChannel + Unpin,
{
PSQHandshakeState::new(connection, local_peer)
.as_initiator(InitiatorData::new(remote_protocol_version, remote_peer))
PSQHandshakeState::new(connection, local_peer).as_initiator(
InitiatorData::new(remote_protocol_version, remote_peer),
mode,
)
}
/// Helper function to create `PSQHandshakeState` for the handshake initiator for mutual KKT
pub fn psq_handshake_initiator_mutual_internode<S>(
connection: &'_ mut S,
local_peer: LpLocalPeer,
remote_peer: LpRemotePeer,
remote_protocol_version: u8,
) -> Result<PSQHandshakeStateInitiator<'_, S>, LpError>
where
S: LpHandshakeChannel + Unpin,
{
Self::psq_handshake_initiator(
connection,
local_peer,
remote_peer,
remote_protocol_version,
HandshakeMode::MutualInternode,
)
}
/// Helper function to create `PSQHandshakeState` for the handshake responder
@@ -173,6 +198,19 @@ impl LpTransportSession {
PSQHandshakeState::new(connection, local_peer).as_responder(ResponderData::default())
}
/// Helper function to create `PSQHandshakeState` for the handshake responder for mutual KKT
pub fn psq_handshake_responder_mutual<S>(
connection: &'_ mut S,
local_peer: LpLocalPeer,
initiator_kem_hashes: BTreeMap<KEM, KEMKeyDigests>,
) -> PSQHandshakeStateResponder<'_, S>
where
S: LpHandshakeChannel + Unpin,
{
PSQHandshakeState::new(connection, local_peer)
.as_responder(ResponderData::default().with_initiator_kem_hashes(initiator_kem_hashes))
}
pub fn session_binding(&self) -> &PersistentSessionBinding {
&self.session_binding
}
integration-tests/src/lp_registration.rs
+9 -8
View File
@@ -17,9 +17,10 @@ mod tests {
use nym_lp::peer::LpLocalPeer;
use nym_node::config::{LpConfig, LpDebug};
use nym_node::node::GatewayStorage;
use nym_node::node::lp::control::handler::LpConnectionHandler;
use nym_node::node::lp::control::ingress::client_handler::LpClientConnectionHandler;
use nym_node::node::lp::error::LpHandlerError;
use nym_node::node::lp::{SharedLpControlState, SharedLpState};
use nym_node::node::lp::state::ActiveLpSessions;
use nym_node::node::lp::{SharedLpClientControlState, SharedLpState};
use nym_node::wireguard::{PeerManager, PeerRegistrator};
use nym_registration_client::{LpClientError, LpRegistrationClient};
use nym_test_utils::helpers::{CryptoRng09, seeded_rng};
@@ -120,7 +121,7 @@ mod tests {
enum SpawnedLpConnectionHandlerState {
NotCreated,
Ready {
handler: LpConnectionHandler<MockIOStream>,
handler: LpClientConnectionHandler<MockIOStream>,
},
Running {
handle: JoinHandle<Option<Result<(), LpHandlerError>>>,
@@ -130,7 +131,7 @@ mod tests {
struct Gateway {
base: Party,
lp_state: SharedLpControlState,
lp_state: SharedLpClientControlState,
ip_pool: IpPool,
mock_peer_controller: SpawnedPeerController,
@@ -225,7 +226,7 @@ mod tests {
upgrade_mode_details,
);
let lp_state = SharedLpControlState {
let lp_state = SharedLpClientControlState {
local_lp_peer: base.peer.clone(),
forward_semaphore,
@@ -235,7 +236,7 @@ mod tests {
shared: SharedLpState {
metrics: Default::default(),
lp_config,
session_states: Arc::new(Default::default()),
session_states: ActiveLpSessions::new(),
},
};
@@ -262,7 +263,7 @@ mod tests {
};
self.lp_connection_handler = SpawnedLpConnectionHandlerState::Ready {
handler: LpConnectionHandler::new(
handler: LpClientConnectionHandler::new(
client_connection,
client_address,
self.lp_state.clone(),
@@ -290,7 +291,7 @@ mod tests {
}
fn spawn_lp_handler(&mut self) {
let SpawnedLpConnectionHandlerState::Ready { handler } = mem::replace(
let SpawnedLpConnectionHandlerState::Ready { mut handler } = mem::replace(
&mut self.lp_connection_handler,
SpawnedLpConnectionHandlerState::NotCreated,
) else {
nym-node/Cargo.toml
+1
View File
@@ -139,6 +139,7 @@ harness = false
cargo_metadata = { workspace = true }
[dev-dependencies]
nym-lp = { workspace = true, features = ["mock"] }
criterion = { workspace = true, features = ["async_tokio"] }
nym-test-utils = { workspace = true }
nym-node/nym-node-metrics/src/network.rs
+37 -7
View File
@@ -16,7 +16,14 @@ pub struct NetworkStats {
// the call stack
active_egress_mixnet_connections: Arc<AtomicUsize>,
active_lp_connections: AtomicUsize,
// incoming LP control connections from clients
active_lp_ingress_client_connections: AtomicUsize,
// incoming LP control connections from nodes
active_lp_ingress_node_connections: AtomicUsize,
// outgoing LP control connections to nodes
active_lp_egress_node_connections: AtomicUsize,
}
impl NetworkStats {
@@ -59,15 +66,38 @@ impl NetworkStats {
.load(Ordering::Relaxed)
}
pub fn new_lp_connection(&self) {
self.active_lp_connections.fetch_add(1, Ordering::Relaxed);
pub fn new_ingress_lp_client_connection(&self) {
self.active_lp_ingress_client_connections
.fetch_add(1, Ordering::Relaxed);
}
pub fn lp_connection_closed(&self) {
self.active_lp_connections.fetch_sub(1, Ordering::Relaxed);
pub fn closed_ingress_lp_client_connection(&self) {
self.active_lp_ingress_client_connections
.fetch_sub(1, Ordering::Relaxed);
}
pub fn active_lp_connections_count(&self) -> usize {
self.active_lp_connections.load(Ordering::Relaxed)
pub fn new_ingress_lp_node_connection(&self) {
self.active_lp_ingress_node_connections
.fetch_add(1, Ordering::Relaxed);
}
pub fn closed_ingress_lp_node_connection(&self) {
self.active_lp_ingress_node_connections
.fetch_sub(1, Ordering::Relaxed);
}
pub fn new_egress_lp_node_connection(&self) {
self.active_lp_egress_node_connections
.fetch_add(1, Ordering::Relaxed);
}
pub fn closed_egress_lp_node_connection(&self) {
self.active_lp_egress_node_connections
.fetch_sub(1, Ordering::Relaxed);
}
pub fn active_lp_client_connections_count(&self) -> usize {
self.active_lp_ingress_client_connections
.load(Ordering::Relaxed)
}
}
nym-node/src/node/lp/cleanup.rs
+4 -7
View File
@@ -2,11 +2,8 @@
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::LpDebug;
use dashmap::DashMap;
use nym_lp::LpTransportSession;
use nym_lp::peer_config::LpReceiverIndex;
use crate::node::lp::state::ActiveLpSessions;
use nym_metrics::inc_by;
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, info};
@@ -74,14 +71,14 @@ impl<T> TimestampedState<T> {
}
pub(crate) struct CleanupTask {
session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpTransportSession>>>,
session_states: ActiveLpSessions,
cfg: LpDebug,
shutdown: nym_task::ShutdownToken,
}
impl CleanupTask {
pub fn new(
session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpTransportSession>>>,
session_states: ActiveLpSessions,
cfg: LpDebug,
shutdown: nym_task::ShutdownToken,
) -> Self {
@@ -100,7 +97,7 @@ impl CleanupTask {
// Remove stale sessions (based on time since last activity)
// Use shorter TTL for demoted (ReadOnlyTransport) sessions
self.session_states.retain(|_, timestamped| {
self.session_states.sessions.retain(|_, timestamped| {
if timestamped.since_activity() > session_ttl {
ss_removed += 1;
false
nym-node/src/node/lp/control/control_tests.rs
+89
View File
@@ -0,0 +1,89 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
#[cfg(test)]
mod tests {
use crate::node::lp::SharedLpState;
use crate::node::lp::control::egress::connection::InitialLpEgressNodeConnectionHandler;
use crate::node::lp::control::ingress::node_handler::InitialLpIngressNodeConnectionHandler;
use crate::node::lp::directory::LpNodeDetails;
use crate::node::lp::state::SharedLpNodeControlState;
use anyhow::Context;
use nym_lp::packet::version;
use nym_lp::peer::{LpLocalPeer, LpRemotePeer, mock_peers};
use nym_test_utils::helpers::seeded_rng;
use nym_test_utils::mocks::async_read_write::MockIOStream;
use nym_test_utils::traits::TimeboxedSpawnable;
use rand::RngCore;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
fn shared_node_state(peer: LpLocalPeer) -> SharedLpNodeControlState {
SharedLpNodeControlState {
local_lp_peer: peer,
nodes: Default::default(),
shared: SharedLpState {
metrics: Default::default(),
lp_config: Default::default(),
session_states: Default::default(),
},
}
}
fn lp_node_details(peer: LpRemotePeer) -> LpNodeDetails {
let key_bytes = peer.x25519().as_ref().try_into().unwrap();
let mut rng = seeded_rng(key_bytes);
LpNodeDetails::new(
rng.next_u32(),
peer.kem_key_digests().clone(),
*peer.x25519(),
version::CURRENT,
)
}
#[tokio::test]
async fn basic_node_to_node_handshake() -> anyhow::Result<()> {
// nym_test_utils::helpers::setup_test_logger();
let (init, resp) = mock_peers();
let init_remote = init.as_remote();
let resp_remote = resp.as_remote();
let conn_init = MockIOStream::default();
let conn_resp = conn_init.try_get_remote_handle();
let init_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 1234);
let init_details = lp_node_details(init_remote);
let resp_details = lp_node_details(resp_remote);
let init_state = shared_node_state(init);
let resp_state = shared_node_state(resp);
let init_handler = InitialLpEgressNodeConnectionHandler::new(
conn_init,
init_addr,
resp_details,
init_state,
);
let resp_handler = InitialLpIngressNodeConnectionHandler::new(
conn_resp,
init_addr,
init_details,
resp_state,
);
let init_future = init_handler.complete_initial_handshake().spawn_timeboxed();
let resp_future = resp_handler.complete_initial_handshake().spawn_timeboxed();
let (init_result, resp_result) = tokio::join!(init_future, resp_future);
let init_result = init_result??.context("handshake failure")??;
let resp_result = resp_result??.context("handshake failure")??;
assert_eq!(
init_result.receiver_index(),
resp_result.transport_session().receiver_index()
);
Ok(())
}
}
nym-node/src/node/lp/control/egress/connection.rs
+102
View File
@@ -0,0 +1,102 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
// not yet used
#![allow(dead_code)]
use crate::node::lp::control::LpConnectionStats;
use crate::node::lp::directory::LpNodeDetails;
use crate::node::lp::error::LpHandlerError;
use crate::node::lp::state::SharedLpNodeControlState;
use nym_lp::LpTransportSession;
use nym_lp::transport::{LpHandshakeChannel, LpTransportChannel};
use std::net::SocketAddr;
use tracing::debug;
/// Initial connection handler for an egress LP node before completing the KKT/PSQ handshake.
pub struct InitialLpEgressNodeConnectionHandler<S> {
stream: S,
remote_addr: SocketAddr,
responder_details: LpNodeDetails,
state: SharedLpNodeControlState,
stats: LpConnectionStats,
}
impl<S> InitialLpEgressNodeConnectionHandler<S>
where
S: LpHandshakeChannel + LpTransportChannel + Unpin,
{
pub(crate) fn new(
stream: S,
remote_addr: SocketAddr,
responder_details: LpNodeDetails,
state: SharedLpNodeControlState,
) -> Self {
Self {
stream,
remote_addr,
responder_details,
state,
stats: LpConnectionStats::new(),
}
}
pub(crate) async fn complete_initial_handshake(
mut self,
) -> Option<Result<LpTransportSession, LpHandlerError>> {
let remote = self.remote_addr;
if self.responder_details.kem_key_hashes.is_empty() {
return Some(Err(LpHandlerError::MissingNodeKEMKeyHashes {
node_ip: self.remote_addr.ip(),
node_id: self.responder_details.node_id,
}));
}
// 1. complete KKT/PSQ handshake before doing anything else.
// bail if it takes too long
let timeout = self.state.shared.lp_config.debug.handshake_ttl;
let stream = &mut self.stream;
let handshake_state = match LpTransportSession::psq_handshake_initiator_mutual_internode(
stream,
self.state.local_lp_peer.clone(),
self.responder_details.to_lp_peer(),
self.responder_details.supported_protocol,
) {
Ok(handshake_state) => handshake_state,
Err(err) => {
debug!("failed to initiate mutual KTT/PSQ handshake with {remote}: {err}");
self.stats.emit_lifecycle_node_metrics(false);
return None;
}
};
let session = match tokio::time::timeout(timeout, handshake_state.complete_handshake())
.await
{
Err(_timeout) => {
debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}");
self.stats.emit_lifecycle_node_metrics(false);
return None;
}
Ok(Err(handshake_failure)) => {
debug!(
"failed to complete mutual KKT/PSQ handshake with {remote}: {handshake_failure}"
);
self.stats.emit_lifecycle_node_metrics(false);
return None;
}
Ok(Ok(session)) => session,
};
debug!(
"completed egress KKT/PSQ handshake with node {}: {remote}",
self.responder_details.node_id
);
// TODO: change return type into complete handler once defined
Some(Ok(session))
}
}
nym-node/src/node/lp/control/egress/mod.rs
+4
View File
@@ -0,0 +1,4 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
pub mod connection;
nym-node/src/node/lp/control/handler.rs → nym-node/src/node/lp/control/ingress/client_handler.rs
+29 -112
View File
@@ -2,8 +2,9 @@
// SPDX-License-Identifier: GPL-3.0-only
use crate::node::lp::cleanup::TimestampedState;
use crate::node::lp::control::{LP_DURATION_BUCKETS, LpConnectionStats};
use crate::node::lp::error::LpHandlerError;
use crate::node::lp::state::SharedLpControlState;
use crate::node::lp::state::SharedLpClientControlState;
use dashmap::mapref::one::RefMut;
use nym_lp::packet::message::LpMessageType;
use nym_lp::packet::{EncryptedLpPacket, ForwardPacketData, LpMessage};
@@ -12,7 +13,8 @@ use nym_lp::session::{LpAction, LpInput};
use nym_lp::transport::LpHandshakeChannel;
use nym_lp::transport::traits::LpTransportChannel;
use nym_lp::{LpTransportSession, packet::message::ExpectedResponseSize};
use nym_metrics::{add_histogram_obs, inc, inc_by};
use nym_metrics::{add_histogram_obs, inc};
use nym_node_metrics::NymNodeMetrics;
use nym_registration_common::{LpRegistrationRequest, RegistrationStatus};
use std::net::SocketAddr;
use std::time::Duration;
@@ -20,66 +22,15 @@ use tokio::net::TcpStream;
use tokio::time::timeout;
use tracing::*;
// Histogram buckets for LP operation duration (legacy - used by unused forwarding methods)
const LP_DURATION_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0];
// Timeout for forward I/O operations (send + receive on exit stream)
// Must be long enough to cover exit gateway processing time
const FORWARD_IO_TIMEOUT_SECS: u64 = 30;
// Histogram buckets for LP connection lifecycle duration
// LP connections can be very short (registration only: ~1s) or very long (dVPN sessions: hours/days)
// Covers full range from seconds to 24 hours
const LP_CONNECTION_DURATION_BUCKETS: &[f64] = &[
1.0, // 1 second
5.0, // 5 seconds
10.0, // 10 seconds
30.0, // 30 seconds
60.0, // 1 minute
300.0, // 5 minutes
600.0, // 10 minutes
1800.0, // 30 minutes
3600.0, // 1 hour
7200.0, // 2 hours
14400.0, // 4 hours
28800.0, // 8 hours
43200.0, // 12 hours
86400.0, // 24 hours
];
/// Connection lifecycle statistics tracking
struct ConnectionStats {
/// When the connection started
start_time: std::time::Instant,
/// Total bytes received (including protocol framing)
bytes_received: u64,
/// Total bytes sent (including protocol framing)
bytes_sent: u64,
}
impl ConnectionStats {
fn new() -> Self {
Self {
start_time: std::time::Instant::now(),
bytes_received: 0,
bytes_sent: 0,
}
}
fn record_bytes_received(&mut self, bytes: usize) {
self.bytes_received += bytes as u64;
}
fn record_bytes_sent(&mut self, bytes: usize) {
self.bytes_sent += bytes as u64;
}
}
pub struct LpConnectionHandler<S = TcpStream> {
pub struct LpClientConnectionHandler<S = TcpStream> {
stream: S,
remote_addr: SocketAddr,
state: SharedLpControlState,
stats: ConnectionStats,
state: SharedLpClientControlState,
stats: LpConnectionStats,
// /// Flag indicating whether this is a connection from an entry gateway serving as a proxy
// forwarded_connection: bool,
@@ -94,27 +45,25 @@ pub struct LpConnectionHandler<S = TcpStream> {
exit_stream: Option<(S, SocketAddr)>,
}
impl<S> LpConnectionHandler<S>
impl<S> LpClientConnectionHandler<S>
where
S: LpTransportChannel + LpHandshakeChannel + Unpin,
{
pub fn new(
stream: S,
// forwarded_connection: bool,
remote_addr: SocketAddr,
state: SharedLpControlState,
) -> Self {
pub fn new(stream: S, remote_addr: SocketAddr, state: SharedLpClientControlState) -> Self {
Self {
stream,
remote_addr,
// forwarded_connection,
state,
stats: ConnectionStats::new(),
stats: LpConnectionStats::new(),
bound_receiver_idx: None,
exit_stream: None,
}
}
pub(crate) fn metrics(&self) -> &NymNodeMetrics {
&self.state.shared.metrics
}
/// Get the mutable reference to the state machine associated with this client.
/// It is vital it's never held across await points or this might lead to a deadlock.
fn state_entry_mut(
@@ -125,8 +74,7 @@ where
self.state
.shared
.session_states
.get_mut(&receiver_index)
.ok_or_else(|| LpHandlerError::MissingLpSession { receiver_index })
.get_state_entry_mut(receiver_index)
}
/// AIDEV-NOTE: Stream-oriented packet loop
@@ -135,11 +83,12 @@ where
/// First packet binds the connection to a receiver_idx (session-affine).
/// Binding is set by handle_client_hello() from payload's receiver_index,
/// or by validate_or_set_binding() for non-bootstrap first packets.
pub async fn handle(mut self) -> Result<(), LpHandlerError> {
debug!("Handling LP connection from {}", self.remote_addr);
pub async fn handle(&mut self) -> Result<(), LpHandlerError> {
let remote = self.remote_addr;
debug!("Handling LP connection from {remote}");
// Track total LP connections handled
inc!("lp_connections_total");
inc!("lp_client_connections_total");
// ============================================================
// STREAM-ORIENTED PROCESSING: Loop until connection closes
@@ -160,18 +109,12 @@ where
.await
{
Err(_timeout) => {
debug!(
"timed out attempting to complete KTT/PSQ handshake with {}",
self.remote_addr
);
debug!("timed out attempting to complete KTT/PSQ handshake with {remote}");
self.emit_lifecycle_metrics(false);
return Ok(());
}
Ok(Err(handshake_failure)) => {
debug!(
"failed to complete KKT/PSQ handshake with {}: {handshake_failure}",
self.remote_addr
);
debug!("failed to complete KKT/PSQ handshake with {remote}: {handshake_failure}",);
self.emit_lifecycle_metrics(false);
return Ok(());
}
@@ -180,10 +123,7 @@ where
let receiver_idx = session.receiver_index();
// 2. insert the state machine into the shared state
self.state
.shared
.session_states
.insert(receiver_idx, TimestampedState::new(session));
self.state.shared.session_states.insert_new_session(session);
self.bound_receiver_idx = Some(receiver_idx);
// 3. handle any new incoming packet
@@ -194,7 +134,7 @@ where
Err(err) => {
if err.is_connection_closed() {
// Graceful EOF - client closed connection
trace!("Connection closed by {} (EOF)", self.remote_addr);
trace!("Connection closed by {remote} (EOF)");
break;
} else {
inc!("lp_errors_receive_packet");
@@ -625,30 +565,7 @@ where
/// Emit connection lifecycle metrics
fn emit_lifecycle_metrics(&self, graceful: bool) {
// Track connection duration
let duration = self.stats.start_time.elapsed().as_secs_f64();
add_histogram_obs!(
"lp_connection_duration_seconds",
duration,
LP_CONNECTION_DURATION_BUCKETS
);
// Track bytes transferred
inc_by!(
"lp_connection_bytes_received_total",
self.stats.bytes_received as i64
);
inc_by!(
"lp_connection_bytes_sent_total",
self.stats.bytes_sent as i64
);
// Track completion type
if graceful {
inc!("lp_connections_completed_gracefully");
} else {
inc!("lp_connections_completed_with_error");
}
self.stats.emit_lifecycle_client_metrics(graceful);
}
}
@@ -657,7 +574,7 @@ mod tests {
use super::*;
use crate::config::LpConfig;
use crate::config::lp::LpDebug;
use crate::node::lp::state::SharedLpState;
use crate::node::lp::state::{ActiveLpSessions, SharedLpState};
use nym_lp::peer::{KEMKeys, LpLocalPeer, generate_keypair_mceliece, generate_keypair_mlkem};
use nym_lp::{Ciphersuite, SessionManager, sessions_for_tests};
use nym_test_utils::helpers::{deterministic_rng, deterministic_rng_09};
@@ -665,7 +582,7 @@ mod tests {
// ==================== Test Helpers ====================
/// Create a minimal test state for handler tests
async fn create_minimal_test_state() -> SharedLpControlState {
async fn create_minimal_test_state() -> SharedLpClientControlState {
use nym_crypto::asymmetric::ed25519;
let mut rng = deterministic_rng();
@@ -690,14 +607,14 @@ mod tests {
);
let lp_peer = LpLocalPeer::new(Ciphersuite::default(), x_keys).with_kem_keys(kem_keys);
SharedLpControlState {
SharedLpClientControlState {
local_lp_peer: lp_peer,
peer_registrator: None,
forward_semaphore,
shared: SharedLpState {
lp_config,
metrics: nym_node_metrics::NymNodeMetrics::default(),
session_states: Arc::new(dashmap::DashMap::new()),
session_states: ActiveLpSessions::new(),
},
}
}
@@ -724,7 +641,7 @@ mod tests {
let server_task = tokio::spawn(async move {
let (stream, remote_addr) = listener.accept().await.unwrap();
let state = create_minimal_test_state().await;
let mut handler = LpConnectionHandler::new(stream, remote_addr, state);
let mut handler = LpClientConnectionHandler::new(stream, remote_addr, state);
// Two-phase: receive raw bytes + header, then parse full packet
let packet = handler.receive_raw_packet().await?;
let header = packet.outer_header();
nym-node/src/node/lp/control/ingress/listener.rs
+192
View File
@@ -0,0 +1,192 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::config::LpConfig;
use crate::error::NymNodeError;
use crate::node::lp::control::ingress::client_handler::LpClientConnectionHandler;
use crate::node::lp::control::ingress::node_handler::InitialLpIngressNodeConnectionHandler;
use crate::node::lp::directory::LpNodeDetails;
use crate::node::lp::state::{SharedLpClientControlState, SharedLpNodeControlState};
use nym_task::ShutdownTracker;
use std::net::SocketAddr;
use tokio::net::TcpListener;
use tracing::{debug, error, info, trace, warn};
/// LP listener that accepts TCP connections on port 41264
pub struct LpControlListener {
/// Address to bind to
bind_address: SocketAddr,
/// Shared state for clients connection handlers
clients_handler_state: SharedLpClientControlState,
/// Shared state for nodes connection handlers
nodes_handler_state: SharedLpNodeControlState,
/// Shutdown coordination
shutdown: ShutdownTracker,
}
impl LpControlListener {
pub fn new(
bind_address: SocketAddr,
clients_handler_state: SharedLpClientControlState,
nodes_handler_state: SharedLpNodeControlState,
shutdown: ShutdownTracker,
) -> Self {
Self {
bind_address,
clients_handler_state,
nodes_handler_state,
shutdown,
}
}
fn lp_config(&self) -> LpConfig {
self.clients_handler_state.shared.lp_config
}
pub async fn run(&mut self) -> Result<(), NymNodeError> {
let bind_address = self.bind_address;
info!("Starting LP control listener on {bind_address}");
let listener = TcpListener::bind(bind_address).await.map_err(|source| {
error!("Failed to bind LP listener to {bind_address}: {source}",);
NymNodeError::LpBindFailure {
address: bind_address,
source,
}
})?;
let shutdown_token = self.shutdown.clone_shutdown_token();
loop {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
trace!("LP listener: received shutdown signal");
break;
}
result = listener.accept() => {
match result {
Ok((stream, addr)) => self.handle_connection(stream, addr),
Err(e) => warn!("Failed to accept LP connection: {e}")
}
}
}
}
info!("LP listener shutdown complete");
Ok(())
}
fn handle_node_connection(
&self,
stream: tokio::net::TcpStream,
remote_addr: SocketAddr,
initiator_details: LpNodeDetails,
) {
warn!(
"received an incoming LP node connection - we don't support this yet beyond the handshake!"
);
debug!("Accepting LP node connection from {remote_addr}");
// Spawn handler task
let handler = InitialLpIngressNodeConnectionHandler::new(
stream,
remote_addr,
initiator_details,
self.nodes_handler_state.clone(),
);
self.shutdown.try_spawn_named_with_shutdown(
async move {
let metrics = handler.metrics().clone();
// Increment connection counter
metrics.network.new_ingress_lp_node_connection();
let result = handler.handle().await;
// Decrement connection counter
metrics.network.closed_ingress_lp_node_connection();
// Handler emits lifecycle metrics internally on success
// For errors, we need to emit them here since handler is consumed
if let Err(e) = result {
warn!("LP node handler error for {remote_addr}: {e}");
// Note: metrics are emitted in handle() for graceful path
// On error path, handle() returns early without emitting
// So we track errors here
}
},
&format!("LP_NODE::{remote_addr}"),
);
}
fn handle_client_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) {
// Check connection limit (only for clients, nodes must always be allowed regardless of the limit)
let active_connections = self.active_client_connections();
let max_connections = self.lp_config().debug.max_connections;
if active_connections >= max_connections {
warn!(
"LP connection limit exceeded ({active_connections}/{max_connections}), rejecting connection from {remote_addr}"
);
return;
}
debug!(
"Accepting LP client connection from {remote_addr} ({active_connections} active connections)"
);
// Spawn handler task
let mut handler =
LpClientConnectionHandler::new(stream, remote_addr, self.clients_handler_state.clone());
self.shutdown.try_spawn_named_with_shutdown(
async move {
// Increment connection counter
handler.metrics().network.new_ingress_lp_client_connection();
let result = handler.handle().await;
// Decrement connection counter
handler
.metrics()
.network
.closed_ingress_lp_client_connection();
// Handler emits lifecycle metrics internally on success
// For errors, we need to emit them here since handler is consumed
if let Err(e) = result {
warn!("LP client handler error for {remote_addr}: {e}");
// Note: metrics are emitted in handle() for graceful path
// On error path, handle() returns early without emitting
// So we track errors here
}
},
&format!("LP_CLIENT::{remote_addr}"),
);
}
fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) {
if let Some(initiator_details) = self
.nodes_handler_state
.nodes
.get_node_details(remote_addr.ip())
{
self.handle_node_connection(stream, remote_addr, initiator_details);
} else {
self.handle_client_connection(stream, remote_addr);
}
}
fn active_client_connections(&self) -> usize {
self.clients_handler_state
.shared
.metrics
.network
.active_lp_client_connections_count()
}
}
nym-node/src/node/lp/control/ingress/mod.rs
+6
View File
@@ -0,0 +1,6 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
pub mod client_handler;
pub(crate) mod listener;
pub mod node_handler;
nym-node/src/node/lp/control/ingress/node_handler.rs
+152
View File
@@ -0,0 +1,152 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::node::lp::control::LpConnectionStats;
use crate::node::lp::directory::LpNodeDetails;
use crate::node::lp::error::LpHandlerError;
use crate::node::lp::state::SharedLpNodeControlState;
use nym_lp::LpTransportSession;
use nym_lp::transport::{LpHandshakeChannel, LpTransportChannel};
use nym_metrics::inc;
use nym_node_metrics::NymNodeMetrics;
use nym_topology::NodeId;
use std::net::SocketAddr;
use tokio::net::TcpStream;
use tracing::debug;
/// Initial connection handler for an ingress LP node before completing the KKT/PSQ handshake.
pub struct InitialLpIngressNodeConnectionHandler<S = TcpStream> {
stream: S,
remote_addr: SocketAddr,
initiator_details: LpNodeDetails,
state: SharedLpNodeControlState,
stats: LpConnectionStats,
}
impl<S> InitialLpIngressNodeConnectionHandler<S>
where
S: LpHandshakeChannel + LpTransportChannel + Unpin,
{
pub fn new(
stream: S,
remote_addr: SocketAddr,
initiator_details: LpNodeDetails,
state: SharedLpNodeControlState,
) -> Self {
Self {
stream,
remote_addr,
initiator_details,
state,
stats: LpConnectionStats::new(),
}
}
pub(crate) fn metrics(&self) -> &NymNodeMetrics {
&self.state.shared.metrics
}
pub(crate) async fn complete_initial_handshake(
mut self,
) -> Option<Result<LpIngressNodeConnectionHandler<S>, LpHandlerError>> {
let remote = self.remote_addr;
if self.initiator_details.kem_key_hashes.is_empty() {
return Some(Err(LpHandlerError::MissingNodeKEMKeyHashes {
node_ip: self.remote_addr.ip(),
node_id: self.initiator_details.node_id,
}));
}
// 1. complete KKT/PSQ handshake before doing anything else.
// bail if it takes too long
let timeout = self.state.shared.lp_config.debug.handshake_ttl;
let local_peer = self.state.local_lp_peer.clone();
let stream = &mut self.stream;
let kem_hashes = self.initiator_details.kem_key_hashes.clone();
let session = match tokio::time::timeout(timeout, async move {
LpTransportSession::psq_handshake_responder_mutual(stream, local_peer, kem_hashes)
.complete_handshake()
.await
})
.await
{
Err(_timeout) => {
debug!("timed out attempting to complete mutual KTT/PSQ handshake with {remote}");
self.stats.emit_lifecycle_node_metrics(false);
return None;
}
Ok(Err(handshake_failure)) => {
debug!(
"failed to complete mutual KKT/PSQ handshake with {remote}: {handshake_failure}"
);
self.stats.emit_lifecycle_node_metrics(false);
return None;
}
Ok(Ok(session)) => session,
};
debug!(
"completed ingress KKT/PSQ handshake with node {}: {remote}",
self.initiator_details.node_id
);
Some(Ok(LpIngressNodeConnectionHandler {
stream: self.stream,
remote_addr: remote,
remote_node_id: self.initiator_details.node_id,
state: self.state,
stats: self.stats,
transport_session: session,
}))
}
pub async fn handle(self) -> Result<(), LpHandlerError> {
// Track total LP connections handled
inc!("lp_node_connections_total");
// attempt to complete initial handshake
let upgraded_handler = match self.complete_initial_handshake().await {
None => return Ok(()),
Some(handler_res) => handler_res?,
};
// continue handling the requests with the transport session
upgraded_handler.handle().await
}
}
/// Connection handler for an LP node after completing the KKT/PSQ handshake.
#[allow(dead_code)]
pub struct LpIngressNodeConnectionHandler<S = TcpStream> {
stream: S,
remote_addr: SocketAddr,
remote_node_id: NodeId,
state: SharedLpNodeControlState,
stats: LpConnectionStats,
transport_session: LpTransportSession,
// TODO [still WIP]:
// LOCAL receiver index to stream id
// client_streams: HashMap<ReceiverIndex, ClientStreamId>,
}
impl<S> LpIngressNodeConnectionHandler<S>
where
S: LpHandshakeChannel + LpTransportChannel + Unpin,
{
async fn handle(self) -> Result<(), LpHandlerError> {
// handle all the forwarding here
self.stats.emit_lifecycle_node_metrics(true);
Ok(())
}
// used in a unit test
#[allow(dead_code)]
pub(crate) fn transport_session(&self) -> &LpTransportSession {
&self.transport_session
}
}
nym-node/src/node/lp/control/listener.rs
-130
View File
@@ -1,130 +0,0 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::config::LpConfig;
use crate::error::NymNodeError;
use crate::node::lp::control::handler::LpConnectionHandler;
use crate::node::lp::state::SharedLpControlState;
use nym_task::ShutdownTracker;
use std::net::SocketAddr;
use tokio::net::TcpListener;
use tracing::{debug, error, info, trace, warn};
/// LP listener that accepts TCP connections on port 41264
pub struct LpControlListener {
/// Address to bind to
bind_address: SocketAddr,
/// Shared state for connection handlers
handler_state: SharedLpControlState,
/// Shutdown coordination
shutdown: ShutdownTracker,
}
impl LpControlListener {
pub fn new(
bind_address: SocketAddr,
handler_state: SharedLpControlState,
shutdown: ShutdownTracker,
) -> Self {
Self {
bind_address,
handler_state,
shutdown,
}
}
fn lp_config(&self) -> LpConfig {
self.handler_state.shared.lp_config
}
pub async fn run(&mut self) -> Result<(), NymNodeError> {
let bind_address = self.bind_address;
info!("Starting LP control listener on {bind_address}");
let listener = TcpListener::bind(bind_address).await.map_err(|source| {
error!("Failed to bind LP listener to {bind_address}: {source}",);
NymNodeError::LpBindFailure {
address: bind_address,
source,
}
})?;
let shutdown_token = self.shutdown.clone_shutdown_token();
loop {
tokio::select! {
biased;
_ = shutdown_token.cancelled() => {
trace!("LP listener: received shutdown signal");
break;
}
result = listener.accept() => {
match result {
Ok((stream, addr)) => self.handle_connection(stream, addr),
Err(e) => warn!("Failed to accept LP connection: {e}")
}
}
}
}
info!("LP listener shutdown complete");
Ok(())
}
fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) {
// Check connection limit
let active_connections = self.active_lp_connections();
let max_connections = self.lp_config().debug.max_connections;
if active_connections >= max_connections {
warn!(
"LP connection limit exceeded ({active_connections}/{max_connections}), rejecting connection from {remote_addr}"
);
return;
}
debug!(
"Accepting LP connection from {remote_addr} ({active_connections} active connections)"
);
// Increment connection counter
self.handler_state
.shared
.metrics
.network
.new_lp_connection();
// Spawn handler task
let handler = LpConnectionHandler::new(stream, remote_addr, self.handler_state.clone());
let metrics = self.handler_state.shared.metrics.clone();
self.shutdown.try_spawn_named_with_shutdown(
async move {
let result = handler.handle().await;
// Handler emits lifecycle metrics internally on success
// For errors, we need to emit them here since handler is consumed
if let Err(e) = result {
warn!("LP handler error for {remote_addr}: {e}");
// Note: metrics are emitted in handle() for graceful path
// On error path, handle() returns early without emitting
// So we track errors here
}
// Decrement connection counter on exit
metrics.network.lp_connection_closed();
},
&format!("LP::{remote_addr}"),
);
}
fn active_lp_connections(&self) -> usize {
self.handler_state
.shared
.metrics
.network
.active_lp_connections_count()
}
}
nym-node/src/node/lp/control/mod.rs
+116 -2
View File
@@ -1,5 +1,119 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
pub mod handler;
pub(crate) mod listener;
use nym_metrics::{add_histogram_obs, inc, inc_by};
mod control_tests;
pub mod egress;
pub mod ingress;
// Histogram buckets for LP operation duration (legacy - used by unused forwarding methods)
const LP_DURATION_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0];
// Histogram buckets for LP connection lifecycle duration
// LP connections can be very short (registration only: ~1s) or very long (dVPN sessions: hours/days)
// Covers full range from seconds to 24 hours
const LP_CONNECTION_DURATION_BUCKETS: &[f64] = &[
1.0, // 1 second
5.0, // 5 seconds
10.0, // 10 seconds
30.0, // 30 seconds
60.0, // 1 minute
300.0, // 5 minutes
600.0, // 10 minutes
1800.0, // 30 minutes
3600.0, // 1 hour
7200.0, // 2 hours
14400.0, // 4 hours
28800.0, // 8 hours
43200.0, // 12 hours
86400.0, // 24 hours
];
/// Connection lifecycle statistics tracking
pub(crate) struct LpConnectionStats {
/// When the connection started
start_time: std::time::Instant,
/// Total bytes received (including protocol framing)
bytes_received: u64,
/// Total bytes sent (including protocol framing)
bytes_sent: u64,
}
impl LpConnectionStats {
fn new() -> Self {
Self {
start_time: std::time::Instant::now(),
bytes_received: 0,
bytes_sent: 0,
}
}
fn duration(&self) -> std::time::Duration {
self.start_time.elapsed()
}
fn record_bytes_received(&mut self, bytes: usize) {
self.bytes_received += bytes as u64;
}
fn record_bytes_sent(&mut self, bytes: usize) {
self.bytes_sent += bytes as u64;
}
/// Emit connection lifecycle metrics for a client connection
fn emit_lifecycle_client_metrics(&self, graceful: bool) {
// Track connection duration
let duration = self.duration().as_secs_f64();
add_histogram_obs!(
"lp_client_connection_duration_seconds",
duration,
LP_CONNECTION_DURATION_BUCKETS
);
// Track bytes transferred
inc_by!(
"lp_client_connection_bytes_received_total",
self.bytes_received as i64
);
inc_by!(
"lp_client_connection_bytes_sent_total",
self.bytes_sent as i64
);
// Track completion type
if graceful {
inc!("lp_client_connections_completed_gracefully");
} else {
inc!("lp_client_connections_completed_with_error");
}
}
/// Emit connection lifecycle metrics for a node connection
fn emit_lifecycle_node_metrics(&self, graceful: bool) {
// Track connection duration
let duration = self.duration().as_secs_f64();
add_histogram_obs!(
"lp_node_connection_duration_seconds",
duration,
LP_CONNECTION_DURATION_BUCKETS
);
// Track bytes transferred
inc_by!(
"lp_node_connection_bytes_received_total",
self.bytes_received as i64
);
inc_by!(
"lp_node_connection_bytes_sent_total",
self.bytes_sent as i64
);
// Track completion type
if graceful {
inc!("lp_node_connections_completed_gracefully");
} else {
inc!("lp_node_connections_completed_with_error");
}
}
}
nym-node/src/node/lp/directory.rs
+68
View File
@@ -0,0 +1,68 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use arc_swap::ArcSwap;
use nym_lp::peer::{DHPublicKey, LpRemotePeer};
use nym_lp::{KEM, KEMKeyDigests};
use nym_topology::NodeId;
use std::collections::{BTreeMap, HashMap};
use std::net::IpAddr;
use std::ops::Deref;
use std::sync::Arc;
/// Wrapper around all known LP nodes
#[derive(Clone, Default)]
pub struct LpNodes {
// map between all available ip addresses of other nodes and their details
nodes: Arc<ArcSwap<HashMap<IpAddr, LpNodeDetails>>>,
}
impl LpNodes {
pub(crate) fn get_node_details(&self, node_ip: IpAddr) -> Option<LpNodeDetails> {
self.nodes.load().get(&node_ip).cloned()
}
}
#[derive(Clone)]
pub struct LpNodeDetails {
inner: Arc<LpNodeDetailsInner>,
}
impl LpNodeDetails {
pub fn new(
node_id: NodeId,
kem_key_hashes: BTreeMap<KEM, KEMKeyDigests>,
x25519: DHPublicKey,
supported_protocol: u8,
) -> Self {
LpNodeDetails {
inner: Arc::new(LpNodeDetailsInner {
node_id,
kem_key_hashes,
x25519,
supported_protocol,
}),
}
}
}
impl Deref for LpNodeDetails {
type Target = LpNodeDetailsInner;
fn deref(&self) -> &Self::Target {
self.inner.deref()
}
}
pub struct LpNodeDetailsInner {
pub node_id: NodeId,
pub kem_key_hashes: BTreeMap<KEM, KEMKeyDigests>,
pub x25519: DHPublicKey,
pub supported_protocol: u8,
}
impl LpNodeDetailsInner {
pub(crate) fn to_lp_peer(&self) -> LpRemotePeer {
LpRemotePeer::new(self.x25519).with_key_digests(self.kem_key_hashes.clone())
}
}
nym-node/src/node/lp/error.rs
+15 -1
View File
@@ -6,7 +6,8 @@ use nym_lp::peer_config::LpReceiverIndex;
use nym_lp::session::LpAction;
use nym_lp::transport::LpTransportError;
use nym_lp::{LpError, packet::MalformedLpPacketError};
use std::net::SocketAddr;
use nym_topology::NodeId;
use std::net::{IpAddr, SocketAddr};
use thiserror::Error;
#[derive(Debug, Error)]
@@ -47,9 +48,18 @@ pub enum LpHandlerError {
#[error("timed out while attempting to send to/receive from the connection")]
ConnectionTimeout,
#[error("missing KEM key hashes for node {node_id} connected from {node_ip}")]
MissingNodeKEMKeyHashes { node_ip: IpAddr, node_id: NodeId },
#[error("data channel is not yet implemented")]
UnimplementedDataChannel,
#[error("{ip_addr} does not correspond to any known LP node")]
NotLpNode { ip_addr: IpAddr },
#[error("{0}")]
Internal(String),
#[error("{0}")]
Other(String),
}
@@ -64,6 +74,10 @@ impl LpHandlerError {
}
}
pub fn internal(msg: impl Into<String>) -> Self {
LpHandlerError::Internal(msg.into())
}
pub fn other(msg: impl Into<String>) -> Self {
LpHandlerError::Other(msg.into())
}
nym-node/src/node/lp/mod.rs
+20 -10
View File
@@ -9,7 +9,7 @@
// ## Connection Metrics (via NetworkStats in nym-node-metrics)
// - active_lp_connections: Gauge tracking current active LP connections (incremented on accept, decremented on close)
//
// ## Handler Metrics (in handler.rs)
// ## Handler Metrics (in client_handler)
// - lp_connections_total: Counter for total LP connections handled
// - lp_client_hello_failed: Counter for ClientHello failures (timestamp validation, protocol errors)
// - lp_handshakes_success: Counter for successful handshake completions
@@ -46,7 +46,7 @@
// ## Error Categorization Metrics
// - lp_errors_wg_peer_registration: Counter for WireGuard peer registration failures
//
// ## Connection Lifecycle Metrics (in handler.rs)
// ## Connection Lifecycle Metrics (in client_handler)
// - lp_connection_duration_seconds: Histogram of connection duration from start to end (buckets: 1s to 24h)
// - lp_connection_bytes_received_total: Counter for total bytes received including protocol framing
// - lp_connection_bytes_sent_total: Counter for total bytes sent including protocol framing
@@ -58,7 +58,7 @@
// - 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)
// ## Subsession/Rekeying Metrics (in client_handler)
// - 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
@@ -70,9 +70,8 @@
use crate::config::LpConfig;
use crate::error::NymNodeError;
use crate::node::lp::cleanup::CleanupTask;
use crate::node::lp::control::listener::LpControlListener;
use crate::node::lp::data::listener::LpDataListener;
use dashmap::DashMap;
use control::ingress::listener::LpControlListener;
use nym_gateway::node::wireguard::PeerRegistrator;
use nym_lp::peer::LpLocalPeer;
use nym_mixnet_client::forwarder::MixForwardingSender;
@@ -82,12 +81,15 @@ use std::sync::Arc;
use tokio::sync::Semaphore;
use tracing::error;
use crate::node::lp::directory::LpNodes;
use crate::node::lp::state::{ActiveLpSessions, SharedLpNodeControlState};
pub use nym_mixnet_client::forwarder::{MixForwardingReceiver, mix_forwarding_channels};
pub use state::{SharedLpControlState, SharedLpDataState, SharedLpState};
pub use state::{SharedLpClientControlState, SharedLpDataState, SharedLpState};
mod cleanup;
pub mod control;
mod data;
pub mod directory;
pub mod error;
mod registration;
pub mod state;
@@ -107,11 +109,12 @@ impl LpSetup {
lp_config: LpConfig,
metrics: NymNodeMetrics,
peer_registrator: Option<PeerRegistrator>,
network_nodes: LpNodes,
mix_packet_sender: MixForwardingSender,
shutdown: ShutdownTracker,
) -> Result<Self, NymNodeError> {
// TODO: this will require loading old states from disk in the future
let session_states = Arc::new(DashMap::new());
let session_states = ActiveLpSessions::new();
let shared_lp_state = SharedLpState {
metrics,
@@ -119,13 +122,19 @@ impl LpSetup {
session_states: session_states.clone(),
};
let control_state = SharedLpControlState {
local_lp_peer,
let client_control_state = SharedLpClientControlState {
local_lp_peer: local_lp_peer.clone(),
peer_registrator,
forward_semaphore: Arc::new(Semaphore::new(lp_config.debug.max_concurrent_forwards)),
shared: shared_lp_state.clone(),
};
let nodes_control_state = SharedLpNodeControlState {
local_lp_peer,
nodes: network_nodes,
shared: shared_lp_state.clone(),
};
let data_state = SharedLpDataState {
outbound_mix_sender: mix_packet_sender,
shared: shared_lp_state,
@@ -133,7 +142,8 @@ impl LpSetup {
let control_listener = LpControlListener::new(
lp_config.control_bind_address,
control_state,
client_control_state,
nodes_control_state,
shutdown.clone(),
);
let data_listener = LpDataListener::new(
nym-node/src/node/lp/registration.rs
+2 -2
View File
@@ -1,7 +1,7 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::node::lp::state::SharedLpControlState;
use crate::node::lp::state::SharedLpClientControlState;
use nym_lp::peer_config::LpReceiverIndex;
use nym_metrics::{add_histogram_obs, inc};
use nym_registration_common::dvpn::{
@@ -29,7 +29,7 @@ const LP_REGISTRATION_DURATION_BUCKETS: &[f64] = &[
30.0, // 30s
];
impl SharedLpControlState {
impl SharedLpClientControlState {
async fn process_dvpn_initial_registration(
&self,
sender: LpReceiverIndex,
nym-node/src/node/lp/state.rs
+50 -5
View File
@@ -3,7 +3,10 @@
use crate::config::LpConfig;
use crate::node::lp::cleanup::TimestampedState;
use crate::node::lp::directory::LpNodes;
use crate::node::lp::error::LpHandlerError;
use dashmap::DashMap;
use dashmap::mapref::one::RefMut;
use nym_gateway::node::wireguard::PeerRegistrator;
use nym_lp::LpTransportSession;
use nym_lp::peer::LpLocalPeer;
@@ -15,7 +18,7 @@ use tokio::sync::Semaphore;
/// Shared state for LP control connections
#[derive(Clone)]
pub struct SharedLpControlState {
pub struct SharedLpClientControlState {
/// Encapsulates all required key information of a local Lewes Protocol Peer.
pub local_lp_peer: LpLocalPeer,
@@ -34,6 +37,19 @@ pub struct SharedLpControlState {
pub shared: SharedLpState,
}
/// [Placeholder] Shared state for LP nodes control connections
#[derive(Clone)]
pub struct SharedLpNodeControlState {
/// Encapsulates all required key information of a local Lewes Protocol Peer.
pub local_lp_peer: LpLocalPeer,
/// Information about all known LP nodes
pub nodes: LpNodes,
/// Common shared data
pub shared: SharedLpState,
}
/// Shared state for LP data connections
#[derive(Clone)]
pub struct SharedLpDataState {
@@ -48,6 +64,37 @@ pub struct SharedLpDataState {
pub shared: SharedLpState,
}
/// Established sessions keyed by the receiver index
///
/// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions.
#[derive(Clone, Default)]
pub struct ActiveLpSessions {
// TODO: this might require split between client and node sessions. TBD
pub(crate) sessions: Arc<DashMap<LpReceiverIndex, TimestampedState<LpTransportSession>>>,
}
impl ActiveLpSessions {
pub fn new() -> Self {
Self::default()
}
pub(crate) fn get_state_entry_mut(
&self,
receiver_index: LpReceiverIndex,
) -> Result<RefMut<'_, LpReceiverIndex, TimestampedState<LpTransportSession>>, LpHandlerError>
{
self.sessions
.get_mut(&receiver_index)
.ok_or_else(|| LpHandlerError::MissingLpSession { receiver_index })
}
pub(crate) fn insert_new_session(&self, session: LpTransportSession) {
let receiver_index = session.receiver_index();
self.sessions
.insert(receiver_index, TimestampedState::new(session));
}
}
/// Shared state for LP connection handlers
#[derive(Clone)]
pub struct SharedLpState {
@@ -57,8 +104,6 @@ pub struct SharedLpState {
/// LP configuration (for timestamp validation, etc.)
pub lp_config: LpConfig,
/// Established sessions keyed by receiver index
///
/// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions.
pub session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpTransportSession>>>,
/// Currently active LP sessions
pub session_states: ActiveLpSessions,
}
nym-node/src/node/mod.rs
+7 -1
View File
@@ -84,6 +84,7 @@ use tokio::sync::mpsc;
use tracing::{debug, info, trace};
use zeroize::Zeroizing;
use crate::node::lp::directory::LpNodes;
pub use nym_gateway::node::ActiveClientsStore;
pub use nym_gateway::node::GatewayStorage;
@@ -491,6 +492,7 @@ impl NymNode {
&self,
peer_registrator: Option<PeerRegistrator>,
mix_packet_sender: MixForwardingSender,
network_nodes: LpNodes,
) -> Result<LpSetup, NymNodeError> {
let lp_peer = LpLocalPeer::new(Ciphersuite::default(), self.x25519_lp_keys.clone())
.with_kem_keys(self.psq_kem_keys.clone());
@@ -500,6 +502,7 @@ impl NymNode {
self.config.lp,
self.metrics.clone(),
peer_registrator,
network_nodes,
mix_packet_sender,
self.shutdown_manager.shutdown_tracker().clone(),
)
@@ -683,6 +686,7 @@ impl NymNode {
async fn start_gateway_tasks(
&mut self,
cached_network: CachedNetwork,
lp_nodes: LpNodes,
metrics_sender: MetricEventsSender,
active_clients_store: ActiveClientsStore,
mix_packet_sender: MixForwardingSender,
@@ -766,7 +770,7 @@ impl NymNode {
self.config.lp.control_bind_address, self.config.lp.data_bind_address,
);
let lp_tasks = self
.build_lp_tasks(wg_peer_registrator.clone(), mix_packet_sender)
.build_lp_tasks(wg_peer_registrator.clone(), mix_packet_sender, lp_nodes)
.await?;
lp_tasks.start_tasks();
} else {
@@ -1355,6 +1359,7 @@ impl NymNode {
let network_refresher = self.build_network_refresher().await?;
let active_clients_store = ActiveClientsStore::new();
let lp_nodes = network_refresher.lp_nodes();
let bloomfilters_manager = self.setup_replay_detection().await?;
@@ -1381,6 +1386,7 @@ impl NymNode {
self.start_gateway_tasks(
network_refresher.cached_network(),
lp_nodes,
metrics_sender,
active_clients_store,
mix_packet_sender,
nym-node/src/node/shared_network.rs
+8
View File
@@ -3,6 +3,7 @@
use crate::error::NymNodeError;
use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
use crate::node::lp::directory::LpNodes;
use crate::node::routing_filter::network_filter::NetworkRoutingFilter;
use async_trait::async_trait;
use nym_crypto::asymmetric::ed25519;
@@ -212,6 +213,7 @@ pub struct NetworkRefresher {
network: CachedNetwork,
routing_filter: NetworkRoutingFilter,
noise_view: NoiseNetworkView,
lp_nodes: LpNodes,
}
impl NetworkRefresher {
@@ -240,6 +242,7 @@ impl NetworkRefresher {
network: CachedNetwork::new_empty(),
routing_filter: NetworkRoutingFilter::new_empty(testnet),
noise_view: NoiseNetworkView::new_empty(),
lp_nodes: Default::default(),
};
this.obtain_initial_network().await?;
@@ -334,6 +337,7 @@ impl NetworkRefresher {
})
.collect::<HashMap<_, _>>();
self.noise_view.swap_view(noise_nodes);
debug!("unimplemented: update LP nodes data");
let mut network_guard = self.network.inner.write().await;
network_guard.topology_metadata = metadata.to_topology_metadata();
@@ -373,6 +377,10 @@ impl NetworkRefresher {
self.noise_view.clone()
}
pub(crate) fn lp_nodes(&self) -> LpNodes {
self.lp_nodes.clone()
}
pub(crate) async fn run(&mut self) {
let mut full_refresh_interval = interval(self.full_refresh_interval);
full_refresh_interval.reset();
nym-registration-client/src/lp_client/client.rs
+3 -1
View File
@@ -14,6 +14,7 @@ use nym_crypto::asymmetric::{ed25519, x25519};
use nym_lp::LpTransportSession;
use nym_lp::peer::{DHKeyPair, LpLocalPeer, LpRemotePeer};
use nym_lp::peer_config::LpReceiverIndex;
use nym_lp::psq::initiator::HandshakeMode;
use nym_lp::transport::traits::LpTransportChannel;
use nym_lp::transport::{LpHandshakeChannel, LpTransportError};
use nym_lp::{Ciphersuite, packet::EncryptedLpPacket, packet::version};
@@ -396,7 +397,8 @@ where
local_peer,
remote_peer,
protocol_version,
)
HandshakeMode::OneWayEntry,
)?
.complete_handshake()
.await?;
nym-registration-client/src/lp_client/nested_session/mod.rs
+3 -1
View File
@@ -28,6 +28,7 @@ use nym_crypto::asymmetric::{ed25519, x25519};
use nym_lp::packet::version;
use nym_lp::packet::{EncryptedLpPacket, LpMessage};
use nym_lp::peer::{DHKeyPair, LpLocalPeer, LpRemotePeer};
use nym_lp::psq::initiator::HandshakeMode;
use nym_lp::transport::LpHandshakeChannel;
use nym_lp::transport::traits::LpTransportChannel;
use nym_lp::{Ciphersuite, KEM, LpTransportSession};
@@ -185,7 +186,8 @@ impl NestedLpSession {
local_peer,
remote_peer,
protocol_version,
)
HandshakeMode::OneWayExit,
)?
.complete_handshake()
.await?;
nym-wallet/Cargo.lock
Generated
+35 -14
View File
@@ -40,7 +40,7 @@ dependencies = [
"serde",
"serde_json",
"serde_repr",
"strum",
"strum 0.27.2",
"tap",
"tauri",
"tauri-build",
@@ -4506,8 +4506,8 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.10.9",
"strum",
"strum_macros",
"strum 0.28.0",
"strum_macros 0.28.0",
"tendermint",
"tendermint-rpc",
"thiserror 2.0.12",
@@ -4621,8 +4621,8 @@ dependencies = [
"nym-upgrade-mode-check",
"rand 0.8.5",
"serde",
"strum",
"strum_macros",
"strum 0.28.0",
"strum_macros 0.28.0",
"thiserror 2.0.12",
"time",
"utoipa",
@@ -4764,8 +4764,8 @@ version = "1.20.4"
dependencies = [
"num_enum",
"semver",
"strum",
"strum_macros",
"strum 0.28.0",
"strum_macros 0.28.0",
"thiserror 2.0.12",
]
@@ -4839,8 +4839,8 @@ dependencies = [
"schemars",
"serde",
"serde_json",
"strum",
"strum_macros",
"strum 0.28.0",
"strum_macros 0.28.0",
"thiserror 2.0.12",
"time",
"url",
@@ -4940,8 +4940,8 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.10.9",
"strum",
"strum_macros",
"strum 0.28.0",
"strum_macros 0.28.0",
"thiserror 2.0.12",
"ts-rs",
"url",
@@ -5055,8 +5055,8 @@ dependencies = [
"nym-vesting-contract-common",
"serde",
"serde_json",
"strum",
"strum_macros",
"strum 0.27.2",
"strum_macros 0.27.2",
"ts-rs",
]
@@ -7375,7 +7375,16 @@ version = "0.27.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf"
dependencies = [
"strum_macros",
"strum_macros 0.27.2",
]
[[package]]
name = "strum"
version = "0.28.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd"
dependencies = [
"strum_macros 0.28.0",
]
[[package]]
@@ -7390,6 +7399,18 @@ dependencies = [
"syn 2.0.100",
]
[[package]]
name = "strum_macros"
version = "0.28.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664"
dependencies = [
"heck 0.5.0",
"proc-macro2",
"quote",
"syn 2.0.100",
]
[[package]]
name = "subtle"
version = "2.6.1"