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

Cleanup and fmt

Browse Source
This commit is contained in:
durch
2025-12-05 13:31:19 +01:00
parent 8fe791c4b4
commit bd85a53079
19 changed files with 656 additions and 604 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nym-lp/src/codec.rs
+11 -11
View File
@@ -16,8 +16,8 @@ pub const OUTER_HEADER_SIZE: usize = OuterHeader::SIZE; // 12 bytes
/// Size of inner prefix (proto + reserved) - cleartext or encrypted depending on mode
const INNER_PREFIX_SIZE: usize = 4; // proto(1) + reserved(3)
use chacha20poly1305::{
aead::{AeadInPlace, KeyInit},
ChaCha20Poly1305, Key, Nonce, Tag,
aead::{AeadInPlace, KeyInit},
};
use zeroize::{Zeroize, ZeroizeOnDrop};
@@ -32,7 +32,7 @@ use zeroize::{Zeroize, ZeroizeOnDrop};
/// ChaCha20-Poly1305 requires unique nonces per key. The counter starts at 0
/// for each session, which is safe because:
///
/// 1. **PSK is always fresh**: Each handshake uses PSQ
/// 1. **PSK is always fresh**: Each handshake uses PSQ
/// with a client-generated random salt. This ensures a unique
/// PSK for every session, even between the same client-gateway pair.
///
@@ -106,9 +106,9 @@ fn parse_message_from_type_and_content(
Ok(LpMessage::Busy)
}
MessageType::Handshake => Ok(LpMessage::Handshake(HandshakeData(content.to_vec()))),
MessageType::EncryptedData => {
Ok(LpMessage::EncryptedData(EncryptedDataPayload(content.to_vec())))
}
MessageType::EncryptedData => Ok(LpMessage::EncryptedData(EncryptedDataPayload(
content.to_vec(),
))),
MessageType::ClientHello => {
let data: ClientHelloData = bincode::deserialize(content)
.map_err(|e| LpError::DeserializationError(e.to_string()))?;
@@ -205,10 +205,7 @@ pub fn parse_lp_header_only(src: &[u8]) -> Result<OuterHeader, LpError> {
/// # Errors
/// * `LpError::AeadTagMismatch` - Tag verification failed (when outer_key provided)
/// * `LpError::InsufficientBufferSize` - Packet too small
pub fn parse_lp_packet(
src: &[u8],
outer_key: Option<&OuterAeadKey>,
) -> Result<LpPacket, LpError> {
pub fn parse_lp_packet(src: &[u8], outer_key: Option<&OuterAeadKey>) -> Result<LpPacket, LpError> {
// Minimum size check: OuterHeader + InnerPrefix + MsgType + Trailer (for 0-payload message)
// 12 + 4 + 2 + 16 = 34 bytes
let min_size = OUTER_HEADER_SIZE + INNER_PREFIX_SIZE + 2 + TRAILER_LEN;
@@ -391,7 +388,7 @@ impl LpError {
#[cfg(test)]
mod tests {
// Import standalone functions
use super::{parse_lp_packet, serialize_lp_packet, OuterAeadKey};
use super::{OuterAeadKey, parse_lp_packet, serialize_lp_packet};
// Keep necessary imports
use crate::LpError;
use crate::message::{EncryptedDataPayload, HandshakeData, LpMessage, MessageType};
@@ -1306,7 +1303,10 @@ mod tests {
let result = parse_lp_packet(&buf, None);
assert!(matches!(
result,
Err(LpError::InvalidPayloadSize { expected: 0, actual: 1 })
Err(LpError::InvalidPayloadSize {
expected: 0,
actual: 1
})
));
}
common/nym-lp/src/lib.rs
+3 -3
View File
@@ -16,7 +16,7 @@ pub mod session_manager;
pub use error::LpError;
pub use message::{ClientHelloData, LpMessage};
pub use packet::{LpPacket, OuterHeader, BOOTSTRAP_RECEIVER_IDX};
pub use packet::{BOOTSTRAP_RECEIVER_IDX, LpPacket, OuterHeader};
pub use replay::{ReceivingKeyCounterValidator, ReplayError};
pub use session::{LpSession, generate_fresh_salt};
pub use session_manager::SessionManager;
@@ -315,8 +315,8 @@ mod tests {
let parsed_packet3 = parse_lp_packet(&buf3, None).unwrap();
// Perform replay check (should fail)
let replay_result =
local_manager.receiving_counter_quick_check(receiver_index, parsed_packet3.header.counter);
let replay_result = local_manager
.receiving_counter_quick_check(receiver_index, parsed_packet3.header.counter);
assert!(replay_result.is_err());
match replay_result.unwrap_err() {
LpError::Replay(e) => {
common/nym-lp/src/message.rs
+1 -1
View File
@@ -233,7 +233,7 @@ impl LpMessage {
LpMessage::SubsessionKK1(_) => false, // Always has payload
LpMessage::SubsessionKK2(_) => false, // Always has payload
LpMessage::SubsessionReady(_) => false, // Always has receiver_index
LpMessage::SubsessionAbort => true, // Empty signal
LpMessage::SubsessionAbort => true, // Empty signal
}
}
common/nym-lp/src/session.rs
+196 -88
View File
@@ -11,7 +11,9 @@ use crate::keypair::{PrivateKey, PublicKey};
use crate::message::{EncryptedDataPayload, HandshakeData};
use crate::noise_protocol::{NoiseError, NoiseProtocol, ReadResult};
use crate::packet::LpHeader;
use crate::psk::{derive_subsession_psk, psq_initiator_create_message, psq_responder_process_message};
use crate::psk::{
derive_subsession_psk, psq_initiator_create_message, psq_responder_process_message,
};
use crate::replay::ReceivingKeyCounterValidator;
use crate::{LpError, LpMessage, LpPacket};
use nym_crypto::asymmetric::ed25519;
@@ -908,7 +910,8 @@ impl LpSession {
let psk = psq_result.psk;
// Store PQ shared secret for subsession PSK derivation
*self.pq_shared_secret.lock() = Some(PqSharedSecret::new(psq_result.pq_shared_secret));
*self.pq_shared_secret.lock() =
Some(PqSharedSecret::new(psq_result.pq_shared_secret));
// Store the PSK handle (ctxt_B) for transmission in next message
{
@@ -938,39 +941,39 @@ impl LpSession {
}
// Check if initiator should extract PSK handle from message 2
if let PSQState::InitiatorWaiting { psk } = *psq_state {
if self.is_initiator {
// Extract PSK handle: [u16 handle_len][handle_bytes][noise_msg]
if payload.len() >= 2 {
let handle_len = u16::from_le_bytes([payload[0], payload[1]]) as usize;
if let PSQState::InitiatorWaiting { psk } = *psq_state
&& self.is_initiator
{
// Extract PSK handle: [u16 handle_len][handle_bytes][noise_msg]
if payload.len() >= 2 {
let handle_len = u16::from_le_bytes([payload[0], payload[1]]) as usize;
if handle_len > 0 && payload.len() >= 2 + handle_len {
// Extract and store the PSK handle
let handle_bytes = &payload[2..2 + handle_len];
let noise_payload = &payload[2 + handle_len..];
if handle_len > 0 && payload.len() >= 2 + handle_len {
// Extract and store the PSK handle
let handle_bytes = &payload[2..2 + handle_len];
let noise_payload = &payload[2 + handle_len..];
tracing::debug!(
"Extracted PSK handle ({} bytes) from message 2",
handle_len
);
tracing::debug!(
"Extracted PSK handle ({} bytes) from message 2",
handle_len
);
{
let mut psk_handle = self.psk_handle.lock();
*psk_handle = Some(handle_bytes.to_vec());
}
// Transition to Completed - we've received confirmation from responder
*psq_state = PSQState::Completed { psk };
drop(psq_state);
// Process only the Noise message part
return noise_state
.read_message(noise_payload)
.map_err(LpError::NoiseError);
{
let mut psk_handle = self.psk_handle.lock();
*psk_handle = Some(handle_bytes.to_vec());
}
// Transition to Completed - we've received confirmation from responder
*psq_state = PSQState::Completed { psk };
drop(psq_state);
// Process only the Noise message part
return noise_state
.read_message(noise_payload)
.map_err(LpError::NoiseError);
}
// If no valid handle found, fall through to normal processing
}
// If no valid handle found, fall through to normal processing
}
// The sans-io NoiseProtocol::read_message expects only the payload.
@@ -1132,9 +1135,7 @@ impl LpSession {
) -> Result<SubsessionHandshake, LpError> {
// Verify parent handshake is complete
if !self.is_handshake_complete() {
return Err(LpError::Internal(
"Parent handshake not complete".into(),
));
return Err(LpError::Internal("Parent handshake not complete".into()));
}
// Get PQ shared secret
@@ -1172,11 +1173,16 @@ impl LpSession {
// Copy key material from parent for into_session() conversion
local_ed25519_private: ed25519::PrivateKey::from_bytes(
&self.local_ed25519_private.to_bytes(),
).expect("Valid Ed25519 private key from parent"),
local_ed25519_public: ed25519::PublicKey::from_bytes(&self.local_ed25519_public.to_bytes())
.expect("Valid Ed25519 public key from parent"),
remote_ed25519_public: ed25519::PublicKey::from_bytes(&self.remote_ed25519_public.to_bytes())
.expect("Valid Ed25519 public key from parent"),
)
.expect("Valid Ed25519 private key from parent"),
local_ed25519_public: ed25519::PublicKey::from_bytes(
&self.local_ed25519_public.to_bytes(),
)
.expect("Valid Ed25519 public key from parent"),
remote_ed25519_public: ed25519::PublicKey::from_bytes(
&self.remote_ed25519_public.to_bytes(),
)
.expect("Valid Ed25519 public key from parent"),
local_x25519_private: self.local_x25519_private.clone(),
remote_x25519_public: self.remote_x25519_public.clone(),
pq_shared_secret: PqSharedSecret::new(pq_secret),
@@ -1305,7 +1311,9 @@ impl SubsessionHandshake {
// KKT: subsession inherits from parent, mark as processed
kkt_state: Mutex::new(KKTState::ResponderProcessed),
// PSQ: subsession uses PSK derived from parent's PQ secret
psq_state: Mutex::new(PSQState::Completed { psk: self.subsession_psk }),
psq_state: Mutex::new(PSQState::Completed {
psk: self.subsession_psk,
}),
psk_handle: Mutex::new(None), // Subsession doesn't have its own handle
sending_counter: AtomicU64::new(0),
receiving_counter: Mutex::new(ReceivingKeyCounterValidator::new(0)),
@@ -1471,8 +1479,12 @@ mod tests {
let responder_keys = generate_keypair();
let receiver_index = 12345u32;
let initiator_session =
create_handshake_test_session(receiver_index, true, &initiator_keys, responder_keys.public_key());
let initiator_session = create_handshake_test_session(
receiver_index,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
receiver_index,
false,
@@ -1499,10 +1511,18 @@ mod tests {
let responder_keys = generate_keypair();
let receiver_index = 12345u32;
let initiator_session =
create_handshake_test_session(receiver_index, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(receiver_index, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
receiver_index,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
receiver_index,
false,
&responder_keys,
initiator_keys.public_key(),
);
// 1. Initiator prepares the first message (-> e)
let initiator_msg_result = initiator_session.prepare_handshake_message();
@@ -1536,10 +1556,18 @@ mod tests {
let initiator_keys = generate_keypair();
let responder_keys = generate_keypair();
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
let mut responder_to_initiator_msg = None;
let mut rounds = 0;
@@ -1623,10 +1651,18 @@ mod tests {
let initiator_keys = generate_keypair();
let responder_keys = generate_keypair();
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Drive handshake to completion (simplified loop from previous test)
let mut i_msg = initiator_session
@@ -1684,8 +1720,12 @@ mod tests {
let initiator_keys = generate_keypair();
let responder_keys = generate_keypair();
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
assert!(!initiator_session.is_handshake_complete());
@@ -1756,10 +1796,18 @@ mod tests {
let initiator_keys = generate_keypair();
let responder_keys = generate_keypair();
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Drive the handshake
let mut i_msg = initiator_session
@@ -1850,10 +1898,18 @@ mod tests {
let responder_keys = generate_keypair();
// Create sessions - they start with dummy PSK [0u8; 32]
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Prepare first message (initiator runs PSQ and injects PSK)
let i_msg = initiator_session
@@ -1915,10 +1971,18 @@ mod tests {
let initiator_keys = generate_keypair();
let responder_keys = generate_keypair();
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Verify initial state
assert!(!initiator_session.is_handshake_complete());
@@ -1994,10 +2058,18 @@ mod tests {
let responder_keys = generate_keypair();
// Create sessions with explicit Ed25519 keys
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Verify sessions store Ed25519 keys
// (Internal verification - keys are used in PSQ calls)
@@ -2039,8 +2111,12 @@ mod tests {
let responder_keys = generate_keypair();
let initiator_keys = generate_keypair();
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Create a handshake message with corrupted PSQ payload
let corrupted_psq_data = vec![0xFF; 128]; // Random garbage
@@ -2204,8 +2280,12 @@ mod tests {
let responder_keys = generate_keypair();
let initiator_keys = generate_keypair();
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Capture initial PSQ state (should be ResponderWaiting)
// (We can't directly access psq_state, but we can verify behavior)
@@ -2222,8 +2302,12 @@ mod tests {
// Session should still be functional - can process valid messages
// Create a proper initiator to send valid message
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let valid_msg = initiator_session
.prepare_handshake_message()
@@ -2249,8 +2333,12 @@ mod tests {
let responder_keys = generate_keypair();
// Create session but don't complete handshake (no PSK injection will occur)
let session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
// Verify session was created successfully
assert!(!session.is_handshake_complete());
@@ -2293,8 +2381,12 @@ mod tests {
let initiator_keys = generate_keypair();
let responder_keys = generate_keypair();
let session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
// Initially not read-only
assert!(!session.is_read_only());
@@ -2314,10 +2406,18 @@ mod tests {
let initiator_keys = generate_keypair();
let responder_keys = generate_keypair();
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Drive handshake to completion
let i_msg = initiator_session
@@ -2362,10 +2462,18 @@ mod tests {
let initiator_keys = generate_keypair();
let responder_keys = generate_keypair();
let initiator_session =
create_handshake_test_session(12345u32, true, &initiator_keys, responder_keys.public_key());
let responder_session =
create_handshake_test_session(12345u32, false, &responder_keys, initiator_keys.public_key());
let initiator_session = create_handshake_test_session(
12345u32,
true,
&initiator_keys,
responder_keys.public_key(),
);
let responder_session = create_handshake_test_session(
12345u32,
false,
&responder_keys,
initiator_keys.public_key(),
);
// Drive handshake to completion
let i_msg = initiator_session
common/nym-lp/src/session_integration/mod.rs
+25 -29
View File
@@ -149,7 +149,8 @@ mod tests {
let counter = session_manager_1.next_counter(receiver_index).unwrap();
let message_a_to_b = create_test_packet(1, receiver_index, counter, payload);
let mut encoded_msg = BytesMut::new();
serialize_lp_packet(&message_a_to_b, &mut encoded_msg, None).expect("A serialize failed");
serialize_lp_packet(&message_a_to_b, &mut encoded_msg, None)
.expect("A serialize failed");
// B parses packet and checks replay
let decoded_packet = parse_lp_packet(&encoded_msg, None).expect("B parse failed");
@@ -200,7 +201,8 @@ mod tests {
let counter = session_manager_2.next_counter(peer_b_sm).unwrap();
let message_b_to_a = create_test_packet(1, receiver_index, counter, payload);
let mut encoded_msg = BytesMut::new();
serialize_lp_packet(&message_b_to_a, &mut encoded_msg, None).expect("B serialize failed");
serialize_lp_packet(&message_b_to_a, &mut encoded_msg, None)
.expect("B serialize failed");
// A parses packet and checks replay
let decoded_packet = parse_lp_packet(&encoded_msg, None).expect("A parse failed");
@@ -289,7 +291,8 @@ mod tests {
.expect("A serialize data failed");
// B parses packet and checks replay
let decoded_packet_b = parse_lp_packet(&encoded_data_a_to_b, None).expect("B parse data failed");
let decoded_packet_b =
parse_lp_packet(&encoded_data_a_to_b, None).expect("B parse data failed");
assert_eq!(decoded_packet_b.header.counter, counter_a);
// Check replay before decrypting
@@ -323,7 +326,8 @@ mod tests {
.expect("B serialize data failed");
// A parses packet and checks replay
let decoded_packet_a = parse_lp_packet(&encoded_data_b_to_a, None).expect("A parse data failed");
let decoded_packet_a =
parse_lp_packet(&encoded_data_b_to_a, None).expect("A parse data failed");
assert_eq!(decoded_packet_a.header.counter, counter_b);
// Check replay before decrypting
@@ -360,8 +364,12 @@ mod tests {
)), // Using plaintext here, but content doesn't matter for replay check
);
let mut encoded_data_b_to_a_replay = BytesMut::new();
serialize_lp_packet(&message_b_to_a_replay, &mut encoded_data_b_to_a_replay, None)
.expect("B serialize replay failed");
serialize_lp_packet(
&message_b_to_a_replay,
&mut encoded_data_b_to_a_replay,
None,
)
.expect("B serialize replay failed");
let parsed_replay_packet =
parse_lp_packet(&encoded_data_b_to_a_replay, None).expect("A parse replay failed");
@@ -398,7 +406,8 @@ mod tests {
.expect("Failed to serialize skip message");
// B parses skip message and checks replay
let decoded_packet_skip = parse_lp_packet(&encoded_skip, None).expect("B parse skip failed");
let decoded_packet_skip =
parse_lp_packet(&encoded_skip, None).expect("B parse skip failed");
session_manager_2
.receiving_counter_quick_check(peer_b_sm, decoded_packet_skip.header.counter)
.expect("B replay check skip failed");
@@ -840,22 +849,6 @@ mod tests {
let ed25519_keypair_a = ed25519::KeyPair::from_secret([6u8; 32], 0);
let ed25519_keypair_b = ed25519::KeyPair::from_secret([7u8; 32], 1);
// Derive X25519 keys from Ed25519 (same as state machine does internally)
let x25519_pub_a = ed25519_keypair_a
.public_key()
.to_x25519()
.expect("Failed to derive X25519 from Ed25519");
let x25519_pub_b = ed25519_keypair_b
.public_key()
.to_x25519()
.expect("Failed to derive X25519 from Ed25519");
// Convert to LP keypair types (needed for init_kkt_for_test if used)
let lp_pub_a = PublicKey::from_bytes(x25519_pub_a.as_bytes())
.expect("Failed to create PublicKey from bytes");
let lp_pub_b = PublicKey::from_bytes(x25519_pub_b.as_bytes())
.expect("Failed to create PublicKey from bytes");
// Use fixed receiver_index for test
let receiver_index: u32 = 100005;
@@ -940,7 +933,8 @@ mod tests {
" Round {}: Responder explicitly enters KKTExchange state",
rounds
);
let action_b_start = session_manager_2.process_input(receiver_index, LpInput::StartHandshake);
let action_b_start =
session_manager_2.process_input(receiver_index, LpInput::StartHandshake);
// Responder's StartHandshake should not produce an action to send
assert!(
action_b_start.as_ref().unwrap().is_none(),
@@ -1030,7 +1024,9 @@ mod tests {
// KKT completed, now need to explicitly trigger handshake message
// This might be the case if KKT completion doesn't automatically send the first Noise message
// Let's try to prepare the handshake message
if let Some(msg_result) = session_manager_1.prepare_handshake_message(receiver_index) {
if let Some(msg_result) =
session_manager_1.prepare_handshake_message(receiver_index)
{
let msg = msg_result.expect("Failed to prepare handshake message after KKT");
// Create a packet from the message
let packet = create_test_packet(1, receiver_index, 0, msg);
@@ -1252,8 +1248,8 @@ mod tests {
// --- 6. Replay Protection Test ---
println!("Testing data packet replay protection via process_input...");
let replay_result =
session_manager_1.process_input(receiver_index, LpInput::ReceivePacket(data_packet_b_replay)); // Use cloned packet
let replay_result = session_manager_1
.process_input(receiver_index, LpInput::ReceivePacket(data_packet_b_replay)); // Use cloned packet
assert!(replay_result.is_err(), "Replay should produce Err(...)");
let error = replay_result.err().unwrap();
@@ -1314,8 +1310,8 @@ mod tests {
// B tries to replay N (should fail)
println!(" B tries to replay N");
let replay_n_result =
session_manager_2.process_input(receiver_index, LpInput::ReceivePacket(packet_n_replay));
let replay_n_result = session_manager_2
.process_input(receiver_index, LpInput::ReceivePacket(packet_n_replay));
assert!(replay_n_result.is_err(), "Replay N should produce Err");
assert!(
matches!(replay_n_result.err().unwrap(), LpError::Replay(_)),
common/nym-lp/src/session_manager.rs
+3 -1
View File
@@ -42,7 +42,9 @@ impl SessionManager {
pub fn add(&self, session: LpSession) -> Result<(), LpError> {
let sm = LpStateMachine {
state: LpState::ReadyToHandshake { session },
state: LpState::ReadyToHandshake {
session: Box::new(session),
},
};
self.state_machines.insert(sm.id()?, sm);
Ok(())
common/nym-lp/src/state_machine.rs
+26 -22
View File
@@ -21,29 +21,29 @@ use tracing::debug;
pub enum LpState {
/// Initial state: Ready to start the handshake.
/// State machine is created with keys, lp_id is derived, session is ready.
ReadyToHandshake { session: LpSession },
ReadyToHandshake { session: Box<LpSession> },
/// Performing KKT (KEM Key Transfer) exchange before Noise handshake.
/// Initiator requests responder's KEM public key, responder provides signed key.
KKTExchange { session: LpSession },
KKTExchange { session: Box<LpSession> },
/// Actively performing the Noise handshake.
/// (We might be able to merge this with ReadyToHandshake if the first step always happens)
Handshaking { session: LpSession }, // Kept for now, logic might merge later
Handshaking { session: Box<LpSession> }, // Kept for now, logic might merge later
/// Handshake complete, ready for data transport.
Transport { session: LpSession },
Transport { session: Box<LpSession> },
/// Performing subsession KK handshake while parent remains active.
/// Parent can still send/receive; subsession messages tunneled through parent.
SubsessionHandshaking {
session: LpSession,
subsession: SubsessionHandshake,
session: Box<LpSession>,
subsession: Box<SubsessionHandshake>,
},
/// Parent session demoted after subsession promoted.
/// Can only receive (drain in-flight), cannot send.
ReadOnlyTransport { session: LpSession },
ReadOnlyTransport { session: Box<LpSession> },
/// An error occurred, or the connection was intentionally closed.
Closed { reason: String },
@@ -119,7 +119,7 @@ pub enum LpAction {
/// the completed SubsessionHandshake for into_session(), and the new receiver_index.
SubsessionComplete {
packet: Option<LpPacket>,
subsession: SubsessionHandshake,
subsession: Box<SubsessionHandshake>,
new_receiver_index: u32,
},
}
@@ -157,7 +157,7 @@ impl LpStateMachine {
| LpState::Handshaking { session }
| LpState::Transport { session }
| LpState::SubsessionHandshaking { session, .. }
| LpState::ReadOnlyTransport { session } => Ok(session),
| LpState::ReadOnlyTransport { session } => Ok(*session),
LpState::Closed { .. } => Err(LpError::LpSessionClosed),
LpState::Processing => Err(LpError::LpSessionProcessing),
}
@@ -234,7 +234,9 @@ impl LpStateMachine {
)?;
Ok(LpStateMachine {
state: LpState::ReadyToHandshake { session },
state: LpState::ReadyToHandshake {
session: Box::new(session),
},
})
}
@@ -257,7 +259,9 @@ impl LpStateMachine {
) -> Result<Self, LpError> {
let session = subsession.into_session(receiver_index)?;
Ok(LpStateMachine {
state: LpState::Transport { session },
state: LpState::Transport {
session: Box::new(session),
},
})
}
@@ -538,7 +542,7 @@ impl LpStateMachine {
Ok(response_packet) => {
result_action = Some(Ok(LpAction::SendPacket(response_packet)));
// Stay in SubsessionHandshaking, wait for SubsessionReady
LpState::SubsessionHandshaking { session, subsession }
LpState::SubsessionHandshaking { session, subsession: Box::new(subsession) }
}
Err(e) => {
let reason = e.to_string();
@@ -659,7 +663,7 @@ impl LpStateMachine {
packet,
subsession_index,
}));
LpState::SubsessionHandshaking { session, subsession }
LpState::SubsessionHandshaking { session, subsession: Box::new(subsession) }
}
Err(e) => {
let reason = e.to_string();
@@ -753,7 +757,7 @@ impl LpStateMachine {
Ok(response_packet) => {
result_action = Some(Ok(LpAction::SendPacket(response_packet)));
// Replace old initiator subsession with new responder subsession
LpState::SubsessionHandshaking { session, subsession: new_subsession }
LpState::SubsessionHandshaking { session, subsession: Box::new(new_subsession) }
}
Err(e) => {
let reason = e.to_string();
@@ -955,8 +959,7 @@ impl LpStateMachine {
if packet.header.receiver_idx() != session.id() {
result_action = Some(Err(LpError::UnknownSessionId(packet.header.receiver_idx())));
LpState::ReadOnlyTransport { session }
} else {
if let Err(e) = session.receiving_counter_quick_check(packet.header.counter) {
} else if let Err(e) = session.receiving_counter_quick_check(packet.header.counter) {
result_action = Some(Err(e));
LpState::ReadOnlyTransport { session }
} else {
@@ -976,7 +979,6 @@ impl LpStateMachine {
LpState::Closed { reason }
}
}
}
}
}
@@ -1613,7 +1615,12 @@ mod tests {
// --- Complete Noise Handshake ---
// Alice prepares first Noise message
let noise1_msg = alice.session().unwrap().prepare_handshake_message().unwrap().unwrap();
let noise1_msg = alice
.session()
.unwrap()
.prepare_handshake_message()
.unwrap()
.unwrap();
let noise1_packet = alice.session().unwrap().next_packet(noise1_msg).unwrap();
// Bob receives noise1, sends noise2
@@ -1666,10 +1673,7 @@ mod tests {
} else {
panic!("Alice should initiate subsession with KK1");
};
assert!(matches!(
alice.state,
LpState::SubsessionHandshaking { .. }
));
assert!(matches!(alice.state, LpState::SubsessionHandshaking { .. }));
// Bob initiates subsession (simultaneously)
let bob_kk1_packet = if let Some(Ok(LpAction::SubsessionInitiated { packet, .. })) =
common/registration/src/lp_messages.rs
+1 -3
View File
@@ -144,11 +144,9 @@ mod tests {
#[test]
fn test_lp_registration_response_success() {
let gateway_data = create_test_gateway_data();
let session_id = 12345;
let allocated_bandwidth = 1_000_000_000;
let response =
LpRegistrationResponse::success(allocated_bandwidth, gateway_data.clone());
let response = LpRegistrationResponse::success(allocated_bandwidth, gateway_data.clone());
assert!(response.success);
assert!(response.error.is_none());
gateway/src/node/lp_listener/handler.rs
+152 -95
View File
@@ -135,11 +135,12 @@ impl LpConnectionHandler {
};
// Step 3: Parse full packet with outer AEAD key
let packet = nym_lp::codec::parse_lp_packet(&raw_bytes, outer_key.as_ref()).map_err(|e| {
inc!("lp_errors_parse_packet");
self.emit_lifecycle_metrics(false);
GatewayError::LpProtocolError(format!("Failed to parse LP packet: {}", e))
})?;
let packet =
nym_lp::codec::parse_lp_packet(&raw_bytes, outer_key.as_ref()).map_err(|e| {
inc!("lp_errors_parse_packet");
self.emit_lifecycle_metrics(false);
GatewayError::LpProtocolError(format!("Failed to parse LP packet: {}", e))
})?;
trace!(
"Received packet from {} (receiver_idx={}, counter={}, encrypted={})",
@@ -179,8 +180,8 @@ impl LpConnectionHandler {
/// Handle ClientHello packet (receiver_idx=0, first packet)
async fn handle_client_hello(&mut self, packet: LpPacket) -> Result<(), GatewayError> {
use nym_lp::state_machine::{LpInput, LpStateMachine};
use nym_lp::packet::LpHeader;
use nym_lp::state_machine::{LpInput, LpStateMachine};
// Extract ClientHello data
let (receiver_index, client_ed25519_pubkey, salt) = match packet.message() {
@@ -195,7 +196,12 @@ impl LpConnectionHandler {
let client_ed25519_pubkey = nym_crypto::asymmetric::ed25519::PublicKey::from_bytes(
&hello_data.client_ed25519_public_key,
)
.map_err(|e| GatewayError::LpProtocolError(format!("Invalid client Ed25519 public key: {}", e)))?;
.map_err(|e| {
GatewayError::LpProtocolError(format!(
"Invalid client Ed25519 public key: {}",
e
))
})?;
(receiver_index, client_ed25519_pubkey, hello_data.salt)
}
@@ -209,22 +215,26 @@ impl LpConnectionHandler {
}
};
debug!("Processing ClientHello from {} (proposed receiver_index={})", self.remote_addr, receiver_index);
debug!(
"Processing ClientHello from {} (proposed receiver_index={})",
self.remote_addr, receiver_index
);
// Collision check for client-proposed receiver_index
// Check both handshake_states (in-progress) and session_states (established)
if self.state.handshake_states.contains_key(&receiver_index)
|| self.state.session_states.contains_key(&receiver_index)
{
warn!("Receiver index collision: {} from {}", receiver_index, self.remote_addr);
warn!(
"Receiver index collision: {} from {}",
receiver_index, self.remote_addr
);
inc!("lp_receiver_index_collision");
// Send Collision response to tell client to retry with new receiver_index
// No outer key - this is before PSK derivation
let collision_packet = LpPacket::new(
LpHeader::new(receiver_index, 0),
LpMessage::Collision,
);
let collision_packet =
LpPacket::new(LpHeader::new(receiver_index, 0), LpMessage::Collision);
self.send_lp_packet(&collision_packet, None).await?;
self.emit_lifecycle_metrics(true);
@@ -255,19 +265,19 @@ impl LpConnectionHandler {
// Transition state machine to KKTExchange (responder waits for client's KKT request)
// For responder, StartHandshake returns None (just transitions state)
// For initiator, StartHandshake returns SendPacket (KKT request)
if let Some(action) = state_machine.process_input(LpInput::StartHandshake) {
if let Err(e) = action {
inc!("lp_client_hello_failed");
return Err(GatewayError::LpHandshakeError(format!(
"StartHandshake failed: {}",
e
)));
}
if let Some(Err(e)) = state_machine.process_input(LpInput::StartHandshake) {
inc!("lp_client_hello_failed");
return Err(GatewayError::LpHandshakeError(format!(
"StartHandshake failed: {}",
e
)));
// Responder (gateway) gets Ok but no packet to send - we just wait for client's next packet
}
// Store state machine for subsequent handshake packets (KKT request with receiver_index=X)
self.state.handshake_states.insert(receiver_index, super::TimestampedState::new(state_machine));
self.state
.handshake_states
.insert(receiver_index, super::TimestampedState::new(state_machine));
debug!(
"Stored handshake state for {} (receiver_index={}) - waiting for KKT request",
@@ -276,10 +286,7 @@ impl LpConnectionHandler {
// Send Ack to confirm ClientHello received (packet-per-connection model)
// No outer key - this is before PSK derivation
let ack_packet = LpPacket::new(
LpHeader::new(receiver_index, 0),
LpMessage::Ack,
);
let ack_packet = LpPacket::new(LpHeader::new(receiver_index, 0), LpMessage::Ack);
self.send_lp_packet(&ack_packet, None).await?;
self.emit_lifecycle_metrics(true);
@@ -292,7 +299,7 @@ impl LpConnectionHandler {
receiver_idx: u32,
packet: LpPacket,
) -> Result<(), GatewayError> {
use nym_lp::state_machine::{LpInput, LpAction};
use nym_lp::state_machine::{LpAction, LpInput};
debug!(
"Processing handshake packet from {} (receiver_idx={})",
@@ -300,9 +307,16 @@ impl LpConnectionHandler {
);
// Get mutable reference to state machine
let mut state_entry = self.state.handshake_states.get_mut(&receiver_idx).ok_or_else(|| {
GatewayError::LpProtocolError(format!("Handshake state not found for session {}", receiver_idx))
})?;
let mut state_entry = self
.state
.handshake_states
.get_mut(&receiver_idx)
.ok_or_else(|| {
GatewayError::LpProtocolError(format!(
"Handshake state not found for session {}",
receiver_idx
))
})?;
let state_machine = &mut state_entry.value_mut().state;
@@ -345,19 +359,28 @@ impl LpConnectionHandler {
// subsession/rekeying support during transport phase
drop(state_entry); // Release mutable borrow
let (_receiver_idx, timestamped_state) = self.state.handshake_states.remove(&receiver_idx)
.ok_or_else(|| GatewayError::LpHandshakeError("Failed to remove handshake state".to_string()))?;
let (_receiver_idx, timestamped_state) = self
.state
.handshake_states
.remove(&receiver_idx)
.ok_or_else(|| {
GatewayError::LpHandshakeError(
"Failed to remove handshake state".to_string(),
)
})?;
self.state.session_states.insert(receiver_idx, timestamped_state);
self.state
.session_states
.insert(receiver_idx, timestamped_state);
inc!("lp_handshakes_success");
// Send Ack to confirm handshake completion to the client
let ack_packet = LpPacket::new(
LpHeader::new(receiver_idx, 0),
LpMessage::Ack,
let ack_packet = LpPacket::new(LpHeader::new(receiver_idx, 0), LpMessage::Ack);
trace!(
"Moved session {} to transport mode, sending Ack",
receiver_idx
);
trace!("Moved session {} to transport mode, sending Ack", receiver_idx);
Some((ack_packet, outer_key))
}
other => {
@@ -370,7 +393,11 @@ impl LpConnectionHandler {
// Send response packet if needed
if let Some((packet, outer_key)) = should_send {
self.send_lp_packet(&packet, outer_key.as_ref()).await?;
trace!("Sent handshake response to {} (encrypted={})", self.remote_addr, outer_key.is_some());
trace!(
"Sent handshake response to {} (encrypted={})",
self.remote_addr,
outer_key.is_some()
);
}
self.emit_lifecycle_metrics(true);
@@ -402,9 +429,13 @@ impl LpConnectionHandler {
);
// Get state machine and process packet
let mut state_entry = self.state.session_states.get_mut(&receiver_idx).ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx))
})?;
let mut state_entry = self
.state
.session_states
.get_mut(&receiver_idx)
.ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx))
})?;
// Update last activity timestamp
state_entry.value().touch();
@@ -420,7 +451,10 @@ impl LpConnectionHandler {
.map_err(|e| GatewayError::LpProtocolError(format!("State machine error: {}", e)))?;
// Get outer key before releasing borrow
let outer_key = state_machine.session().ok().and_then(|s| s.outer_aead_key());
let outer_key = state_machine
.session()
.ok()
.and_then(|s| s.outer_aead_key());
drop(state_entry);
match action {
@@ -432,13 +466,15 @@ impl LpConnectionHandler {
self.remote_addr, receiver_idx
);
inc!("lp_subsession_kk2_sent");
self.send_lp_packet(&response_packet, outer_key.as_ref()).await?;
self.send_lp_packet(&response_packet, outer_key.as_ref())
.await?;
self.emit_lifecycle_metrics(true);
Ok(())
}
LpAction::DeliverData(data) => {
// Decrypted application data - process as registration/forwarding
self.handle_decrypted_payload(receiver_idx, data.to_vec()).await
self.handle_decrypted_payload(receiver_idx, data.to_vec())
.await
}
LpAction::SubsessionComplete {
packet: ready_packet,
@@ -449,7 +485,7 @@ impl LpConnectionHandler {
self.handle_subsession_complete(
receiver_idx,
ready_packet,
subsession,
*subsession,
new_receiver_index,
outer_key,
)
@@ -481,7 +517,9 @@ impl LpConnectionHandler {
"LP registration request from {} (receiver_idx={}): mode={:?}",
self.remote_addr, receiver_idx, request.mode
);
return self.handle_registration_request(receiver_idx, request).await;
return self
.handle_registration_request(receiver_idx, request)
.await;
}
// Try to deserialize as ForwardPacketData (entry gateway forwarding to exit)
@@ -490,7 +528,9 @@ impl LpConnectionHandler {
"LP forward request from {} (receiver_idx={}) to {}",
self.remote_addr, receiver_idx, forward_data.target_lp_address
);
return self.handle_forwarding_request(receiver_idx, forward_data).await;
return self
.handle_forwarding_request(receiver_idx, forward_data)
.await;
}
// Neither registration nor forwarding - unknown payload type
@@ -535,14 +575,20 @@ impl LpConnectionHandler {
// Create new state machine from completed subsession
let new_state_machine = LpStateMachine::from_subsession(subsession, new_receiver_index)
.map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to create session from subsession: {}", e))
GatewayError::LpProtocolError(format!(
"Failed to create session from subsession: {}",
e
))
})?;
// Check for receiver_index collision before inserting
// new_receiver_index is client-generated (rand::random() in state machine).
// Collisions are statistically unlikely (1 in 4 billion) but could cause DoS if exploited.
if self.state.session_states.contains_key(&new_receiver_index)
|| self.state.handshake_states.contains_key(&new_receiver_index)
|| self
.state
.handshake_states
.contains_key(&new_receiver_index)
{
warn!(
"Subsession receiver_index collision: {} from {}",
@@ -556,9 +602,10 @@ impl LpConnectionHandler {
}
// Store new session under new_receiver_index
self.state
.session_states
.insert(new_receiver_index, super::TimestampedState::new(new_state_machine));
self.state.session_states.insert(
new_receiver_index,
super::TimestampedState::new(new_state_machine),
);
// Old session is now in ReadOnlyTransport state (handled by state machine)
// It will be cleaned up by TTL-based cleanup task
@@ -579,9 +626,13 @@ impl LpConnectionHandler {
// Acquire session lock for encryption and get outer AEAD key
let (response_packet, outer_key) = {
let session_entry = self.state.session_states.get(&receiver_idx).ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx))
})?;
let session_entry = self
.state
.session_states
.get(&receiver_idx)
.ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx))
})?;
// Access session via state machine for subsession support
let session = session_entry
.value()
@@ -608,13 +659,11 @@ impl LpConnectionHandler {
};
// Send response (encrypted with outer AEAD)
self.send_lp_packet(&response_packet, outer_key.as_ref()).await?;
self.send_lp_packet(&response_packet, outer_key.as_ref())
.await?;
if response.success {
info!(
"LP registration successful for {})",
self.remote_addr
);
info!("LP registration successful for {})", self.remote_addr);
} else {
warn!(
"LP registration failed for {}: {:?}",
@@ -641,9 +690,13 @@ impl LpConnectionHandler {
// Encrypt response for client and get outer AEAD key
let (response_packet, outer_key) = {
let session_entry = self.state.session_states.get(&receiver_idx).ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx))
})?;
let session_entry = self
.state
.session_states
.get(&receiver_idx)
.ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx))
})?;
// Access session via state machine for subsession support
let session = session_entry
.value()
@@ -665,7 +718,8 @@ impl LpConnectionHandler {
};
// Send encrypted response to client (encrypted with outer AEAD)
self.send_lp_packet(&response_packet, outer_key.as_ref()).await?;
self.send_lp_packet(&response_packet, outer_key.as_ref())
.await?;
debug!(
"LP forwarding completed for {} (receiver_idx={})",
@@ -810,8 +864,8 @@ impl LpConnectionHandler {
&mut self,
forward_data: ForwardPacketData,
) -> Result<Vec<u8>, GatewayError> {
use tokio::time::timeout;
use std::time::Duration;
use tokio::time::timeout;
inc!("lp_forward_total");
let start = std::time::Instant::now();
@@ -823,22 +877,23 @@ impl LpConnectionHandler {
})?;
// Connect to target gateway with timeout
let mut target_stream = match timeout(Duration::from_secs(5), TcpStream::connect(target_addr)).await {
Ok(Ok(stream)) => stream,
Ok(Err(e)) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(format!(
"Failed to connect to target gateway: {}",
e
)));
}
Err(_) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(
"Target gateway connection timeout".to_string(),
));
}
};
let mut target_stream =
match timeout(Duration::from_secs(5), TcpStream::connect(target_addr)).await {
Ok(Ok(stream)) => stream,
Ok(Err(e)) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(format!(
"Failed to connect to target gateway: {}",
e
)));
}
Err(_) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(
"Target gateway connection timeout".to_string(),
));
}
};
debug!(
"Forwarding packet to {} (target: {})",
@@ -872,7 +927,10 @@ impl LpConnectionHandler {
let mut len_buf = [0u8; 4];
target_stream.read_exact(&mut len_buf).await.map_err(|e| {
inc!("lp_forward_failed");
GatewayError::LpConnectionError(format!("Failed to read response length from target: {}", e))
GatewayError::LpConnectionError(format!(
"Failed to read response length from target: {}",
e
))
})?;
let response_len = u32::from_be_bytes(len_buf) as usize;
@@ -893,23 +951,20 @@ impl LpConnectionHandler {
.await
.map_err(|e| {
inc!("lp_forward_failed");
GatewayError::LpConnectionError(format!("Failed to read response from target: {}", e))
GatewayError::LpConnectionError(format!(
"Failed to read response from target: {}",
e
))
})?;
// Record metrics
let duration = start.elapsed().as_secs_f64();
add_histogram_obs!(
"lp_forward_duration_seconds",
duration,
LP_DURATION_BUCKETS
);
add_histogram_obs!("lp_forward_duration_seconds", duration, LP_DURATION_BUCKETS);
inc!("lp_forward_success");
debug!(
"Forwarding successful to {} ({} bytes response, {:.3}s)",
target_addr,
response_len,
duration
target_addr, response_len, duration
);
Ok(response_buf)
@@ -950,8 +1005,9 @@ impl LpConnectionHandler {
self.stats.record_bytes_received(4 + packet_len);
// Parse header only (for routing - header is always cleartext)
let header = parse_lp_header_only(&packet_buf)
.map_err(|e| GatewayError::LpProtocolError(format!("Failed to parse LP header: {}", e)))?;
let header = parse_lp_header_only(&packet_buf).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to parse LP header: {}", e))
})?;
Ok((packet_buf, header))
}
@@ -1214,8 +1270,9 @@ mod tests {
let mut handler = LpConnectionHandler::new(stream, remote_addr, state);
// Two-phase: receive raw bytes + header, then parse full packet
let (raw_bytes, header) = handler.receive_raw_packet().await?;
let packet = parse_lp_packet(&raw_bytes, None)
.map_err(|e| GatewayError::LpProtocolError(format!("Failed to parse packet: {}", e)))?;
let packet = parse_lp_packet(&raw_bytes, None).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to parse packet: {}", e))
})?;
Ok::<_, GatewayError>((header, packet))
});
gateway/src/node/lp_listener/handshake.rs
-182
View File
@@ -1,182 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use crate::error::GatewayError;
use nym_lp::{
state_machine::{LpAction, LpInput, LpStateMachine},
LpPacket, LpSession,
};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tracing::*;
/// Wrapper around the nym-lp state machine for gateway-side LP connections
pub struct LpGatewayHandshake {
state_machine: LpStateMachine,
}
impl LpGatewayHandshake {
/// Create a new responder (gateway side) handshake
///
/// # Arguments
/// * `receiver_index` - Client-proposed receiver_index (from ClientHello)
/// * `gateway_ed25519_keypair` - Gateway's Ed25519 identity keypair (for PSQ auth and X25519 derivation)
/// * `client_ed25519_public_key` - Client's Ed25519 public key (from ClientHello)
/// * `salt` - Salt from ClientHello (for PSK derivation)
pub fn new_responder(
receiver_index: u32,
gateway_ed25519_keypair: (
&nym_crypto::asymmetric::ed25519::PrivateKey,
&nym_crypto::asymmetric::ed25519::PublicKey,
),
client_ed25519_public_key: &nym_crypto::asymmetric::ed25519::PublicKey,
salt: &[u8; 32],
) -> Result<Self, GatewayError> {
let state_machine = LpStateMachine::new(
receiver_index,
false, // responder
gateway_ed25519_keypair,
client_ed25519_public_key,
salt,
)
.map_err(|e| {
GatewayError::LpHandshakeError(format!("Failed to create state machine: {}", e))
})?;
Ok(Self { state_machine })
}
/// Complete the handshake and return the established session
pub async fn complete(mut self, stream: &mut TcpStream) -> Result<LpSession, GatewayError> {
debug!("Starting LP handshake as responder");
// Start the handshake
if let Some(action) = self.state_machine.process_input(LpInput::StartHandshake) {
match action {
Ok(LpAction::SendPacket(packet)) => {
self.send_packet(stream, &packet).await?;
}
Ok(_) => {
// Unexpected action at this stage
return Err(GatewayError::LpHandshakeError(
"Unexpected action at handshake start".to_string(),
));
}
Err(e) => {
return Err(GatewayError::LpHandshakeError(format!(
"Failed to start handshake: {}",
e
)));
}
}
}
// Continue handshake until complete
loop {
// Read incoming packet
let packet = self.receive_packet(stream).await?;
// Process the received packet
if let Some(action) = self
.state_machine
.process_input(LpInput::ReceivePacket(packet))
{
match action {
Ok(LpAction::SendPacket(response_packet)) => {
self.send_packet(stream, &response_packet).await?;
}
Ok(LpAction::HandshakeComplete) => {
info!("LP handshake completed successfully");
break;
}
Ok(other) => {
debug!("Received action during handshake: {:?}", other);
}
Err(e) => {
return Err(GatewayError::LpHandshakeError(format!(
"Handshake error: {}",
e
)));
}
}
}
}
// Extract the session from the state machine
self.state_machine.into_session().map_err(|e| {
GatewayError::LpHandshakeError(format!("Failed to get session after handshake: {}", e))
})
}
/// Send an LP packet over the stream with proper length-prefixed framing
async fn send_packet(
&self,
stream: &mut TcpStream,
packet: &LpPacket,
) -> Result<(), GatewayError> {
use bytes::BytesMut;
use nym_lp::codec::serialize_lp_packet;
// Serialize the packet first (None key during handshake phase)
let mut packet_buf = BytesMut::new();
serialize_lp_packet(packet, &mut packet_buf, None).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to serialize packet: {}", e))
})?;
// Send 4-byte length prefix (u32 big-endian)
let len = packet_buf.len() as u32;
stream.write_all(&len.to_be_bytes()).await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to send packet length: {}", e))
})?;
// Send the actual packet data
stream.write_all(&packet_buf).await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to send packet data: {}", e))
})?;
stream.flush().await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to flush stream: {}", e))
})?;
debug!(
"Sent LP packet ({} bytes + 4 byte header)",
packet_buf.len()
);
Ok(())
}
/// Receive an LP packet from the stream with proper length-prefixed framing
async fn receive_packet(&self, stream: &mut TcpStream) -> Result<LpPacket, GatewayError> {
use nym_lp::codec::parse_lp_packet;
// Read 4-byte length prefix (u32 big-endian)
let mut len_buf = [0u8; 4];
stream.read_exact(&mut len_buf).await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to read packet length: {}", e))
})?;
let packet_len = u32::from_be_bytes(len_buf) as usize;
// Sanity check to prevent huge allocations
const MAX_PACKET_SIZE: usize = 65536; // 64KB max
if packet_len > MAX_PACKET_SIZE {
return Err(GatewayError::LpProtocolError(format!(
"Packet size {} exceeds maximum {}",
packet_len, MAX_PACKET_SIZE
)));
}
// Read the actual packet data
let mut packet_buf = vec![0u8; packet_len];
stream.read_exact(&mut packet_buf).await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to read packet data: {}", e))
})?;
// Parse packet (None key during handshake phase)
let packet = parse_lp_packet(&packet_buf, None)
.map_err(|e| GatewayError::LpProtocolError(format!("Failed to parse packet: {}", e)))?;
debug!("Received LP packet ({} bytes + 4 byte header)", packet_len);
Ok(packet)
}
}
gateway/src/node/lp_listener/mod.rs
+6 -3
View File
@@ -83,7 +83,6 @@ use tokio::sync::mpsc;
use tracing::*;
mod handler;
mod handshake;
mod messages;
mod registration;
@@ -269,7 +268,8 @@ impl<T> TimestampedState<T> {
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
self.last_activity.store(now, std::sync::atomic::Ordering::Relaxed);
self.last_activity
.store(now, std::sync::atomic::Ordering::Relaxed);
}
/// Get age since creation
@@ -283,7 +283,9 @@ impl<T> TimestampedState<T> {
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let last = self.last_activity.load(std::sync::atomic::Ordering::Relaxed);
let last = self
.last_activity
.load(std::sync::atomic::Ordering::Relaxed);
now.saturating_sub(last)
}
}
@@ -511,6 +513,7 @@ impl LpListener {
///
/// Demoted sessions (ReadOnlyTransport) use shorter TTL since they
/// only need to drain in-flight packets after subsession promotion.
#[allow(clippy::too_many_arguments)]
async fn cleanup_loop(
handshake_states: Arc<DashMap<u32, TimestampedState<LpStateMachine>>>,
session_states: Arc<DashMap<u32, TimestampedState<LpStateMachine>>>,
nym-gateway-probe/src/common/mod.rs
+1 -1
View File
@@ -9,5 +9,5 @@
pub mod wireguard;
pub use wireguard::{
run_tunnel_tests, run_two_hop_tunnel_tests, TwoHopWgTunnelConfig, WgTunnelConfig,
TwoHopWgTunnelConfig, WgTunnelConfig, run_tunnel_tests, run_two_hop_tunnel_tests,
};
nym-gateway-probe/src/common/wireguard.rs
+4 -3
View File
@@ -9,9 +9,11 @@
use nym_config::defaults::{WG_METADATA_PORT, WG_TUN_DEVICE_IP_ADDRESS_V4};
use tracing::{error, info};
use crate::netstack::{NetstackRequest, NetstackRequestGo, NetstackResult, TwoHopNetstackRequestGo};
use crate::types::WgProbeResults;
use crate::NetstackArgs;
use crate::netstack::{
NetstackRequest, NetstackRequestGo, NetstackResult, TwoHopNetstackRequestGo,
};
use crate::types::WgProbeResults;
/// Safe division that returns 0.0 when divisor is 0 (instead of NaN/Inf)
fn safe_ratio(received: u16, sent: u16) -> f32 {
@@ -80,7 +82,6 @@ pub fn run_tunnel_tests(
awg_args: &str,
wg_outcome: &mut WgProbeResults,
) {
// Build the netstack request
let netstack_request = NetstackRequest::new(
&config.private_ipv4,
nym-gateway-probe/src/lib.rs
+55 -30
View File
@@ -51,7 +51,6 @@ use crate::{
types::Exit,
};
mod bandwidth_helpers;
mod common;
mod icmp;
@@ -62,8 +61,8 @@ mod types;
use crate::bandwidth_helpers::{acquire_bandwidth, import_bandwidth};
use crate::nodes::{DirectoryNode, NymApiDirectory};
use nym_node_status_client::models::AttachedTicketMaterials;
pub use mode::TestMode;
use nym_node_status_client::models::AttachedTicketMaterials;
pub use types::{IpPingReplies, ProbeOutcome, ProbeResult};
#[derive(Args, Clone)]
@@ -283,6 +282,7 @@ impl Probe {
self
}
#[allow(clippy::too_many_arguments)]
pub async fn probe(
self,
directory: Option<NymApiDirectory>,
@@ -320,7 +320,8 @@ impl Probe {
// Convert legacy flags to TestMode
let has_exit = self.exit_gateway_node.is_some() || self.localnet_exit.is_some();
let test_mode = TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, has_exit);
let test_mode =
TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, has_exit);
self.do_probe_test(
Some(mixnet_client),
@@ -375,7 +376,8 @@ impl Probe {
// Convert legacy flags to TestMode
let has_exit = self.localnet_exit.is_some();
let test_mode = TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, has_exit);
let test_mode =
TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, has_exit);
return self
.do_probe_test(
@@ -396,8 +398,16 @@ impl Probe {
// If both gateways are pre-queried via --gateway-ip and --exit-gateway-ip,
// skip mixnet setup entirely - we have all the data we need
if self.direct_gateway_node.is_some() && self.exit_gateway_node.is_some() {
let entry_node = self.direct_gateway_node.as_ref().unwrap();
let exit_node = self.exit_gateway_node.as_ref().unwrap();
let entry_node = if let Some(entry_node) = self.direct_gateway_node.as_ref() {
entry_node
} else {
return Err(anyhow::anyhow!("Entry gateway node is missing"));
};
let exit_node = if let Some(exit_node) = self.exit_gateway_node.as_ref() {
exit_node
} else {
return Err(anyhow::anyhow!("Exit gateway node is missing"));
};
// Initialize storage (needed for credentials)
if !config_dir.exists() {
@@ -413,7 +423,8 @@ impl Probe {
let node_info = exit_node.to_testable_node()?;
// Convert legacy flags to TestMode (has_exit = true since we have exit_gateway_node)
let test_mode = TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, true);
let test_mode =
TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, true);
return self
.do_probe_test(
@@ -504,7 +515,8 @@ impl Probe {
// Convert legacy flags to TestMode
let has_exit = self.exit_gateway_node.is_some() || self.localnet_exit.is_some();
let test_mode = TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, has_exit);
let test_mode =
TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, has_exit);
self.do_probe_test(
Some(mixnet_client),
@@ -732,24 +744,30 @@ impl Probe {
} else if test_mode.uses_lp() && test_mode.tests_wireguard() {
// LP modes (SingleHop/TwoHop) don't need mixnet client
// Create default outcome and continue to LP-WG test below
(Ok(ProbeOutcome {
as_entry: Entry::NotTested,
as_exit: None,
wg: None,
lp: None,
}), None)
(
Ok(ProbeOutcome {
as_entry: Entry::NotTested,
as_exit: None,
wg: None,
lp: None,
}),
None,
)
} else {
// For Mixnet mode, missing mixnet client is a failure
(Ok(ProbeOutcome {
as_entry: if tested_entry {
Entry::fail_to_connect()
} else {
Entry::EntryFailure
},
as_exit: None,
wg: None,
lp: None,
}), None)
(
Ok(ProbeOutcome {
as_entry: if tested_entry {
Entry::fail_to_connect()
} else {
Entry::EntryFailure
},
as_exit: None,
wg: None,
lp: None,
}),
None,
)
};
let wg_outcome = if !test_mode.tests_wireguard() {
@@ -790,10 +808,9 @@ impl Probe {
} else if let Some(exit_localnet) = &self.localnet_exit {
// Localnet mode: use CLI-provided identities and LP addresses
info!("Using localnet entry and exit gateways for LP forwarding test");
let entry_localnet = self
.localnet_entry
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Entry gateway not available in localnet mode"))?;
let entry_localnet = self.localnet_entry.as_ref().ok_or_else(|| {
anyhow::anyhow!("Entry gateway not available in localnet mode")
})?;
(entry_localnet.clone(), exit_localnet.clone())
} else {
@@ -1120,7 +1137,10 @@ where
ticket_type,
);
match client.register_with_credential(&wg_keypair, credential, ticket_type).await {
match client
.register_with_credential(&wg_keypair, credential, ticket_type)
.await
{
Ok(data) => data,
Err(e) => {
let error_msg = format!("LP registration failed (mock ecash): {}", e);
@@ -1132,7 +1152,12 @@ where
} else {
info!("Using real bandwidth controller for LP registration");
match client
.register(&wg_keypair, &gateway_ed25519_pubkey, bandwidth_controller, ticket_type)
.register(
&wg_keypair,
&gateway_ed25519_pubkey,
bandwidth_controller,
ticket_type,
)
.await
{
Ok(data) => data,
nym-gateway-probe/src/mode/mod.rs
+30 -7
View File
@@ -58,12 +58,18 @@ impl TestMode {
/// Whether this mode uses LP registration
pub fn uses_lp(&self) -> bool {
matches!(self, TestMode::SingleHop | TestMode::TwoHop | TestMode::LpOnly)
matches!(
self,
TestMode::SingleHop | TestMode::TwoHop | TestMode::LpOnly
)
}
/// Whether this mode tests WireGuard tunnels
pub fn tests_wireguard(&self) -> bool {
matches!(self, TestMode::Mixnet | TestMode::SingleHop | TestMode::TwoHop)
matches!(
self,
TestMode::Mixnet | TestMode::SingleHop | TestMode::TwoHop
)
}
/// Whether this mode requires an exit gateway
@@ -214,7 +220,10 @@ mod tests {
#[test]
fn test_from_str_canonical() {
assert_eq!("mixnet".parse::<TestMode>().unwrap(), TestMode::Mixnet);
assert_eq!("single-hop".parse::<TestMode>().unwrap(), TestMode::SingleHop);
assert_eq!(
"single-hop".parse::<TestMode>().unwrap(),
TestMode::SingleHop
);
assert_eq!("two-hop".parse::<TestMode>().unwrap(), TestMode::TwoHop);
assert_eq!("lp-only".parse::<TestMode>().unwrap(), TestMode::LpOnly);
}
@@ -222,12 +231,18 @@ mod tests {
#[test]
fn test_from_str_alternate_formats() {
// snake_case
assert_eq!("single_hop".parse::<TestMode>().unwrap(), TestMode::SingleHop);
assert_eq!(
"single_hop".parse::<TestMode>().unwrap(),
TestMode::SingleHop
);
assert_eq!("two_hop".parse::<TestMode>().unwrap(), TestMode::TwoHop);
assert_eq!("lp_only".parse::<TestMode>().unwrap(), TestMode::LpOnly);
// no separator
assert_eq!("singlehop".parse::<TestMode>().unwrap(), TestMode::SingleHop);
assert_eq!(
"singlehop".parse::<TestMode>().unwrap(),
TestMode::SingleHop
);
assert_eq!("twohop".parse::<TestMode>().unwrap(), TestMode::TwoHop);
assert_eq!("lponly".parse::<TestMode>().unwrap(), TestMode::LpOnly);
}
@@ -235,7 +250,10 @@ mod tests {
#[test]
fn test_from_str_case_insensitive() {
assert_eq!("MIXNET".parse::<TestMode>().unwrap(), TestMode::Mixnet);
assert_eq!("Single-Hop".parse::<TestMode>().unwrap(), TestMode::SingleHop);
assert_eq!(
"Single-Hop".parse::<TestMode>().unwrap(),
TestMode::SingleHop
);
assert_eq!("TWO_HOP".parse::<TestMode>().unwrap(), TestMode::TwoHop);
assert_eq!("LpOnly".parse::<TestMode>().unwrap(), TestMode::LpOnly);
}
@@ -251,7 +269,12 @@ mod tests {
#[test]
fn test_display_fromstr_roundtrip() {
for mode in [TestMode::Mixnet, TestMode::SingleHop, TestMode::TwoHop, TestMode::LpOnly] {
for mode in [
TestMode::Mixnet,
TestMode::SingleHop,
TestMode::TwoHop,
TestMode::LpOnly,
] {
let s = mode.to_string();
let parsed: TestMode = s.parse().unwrap();
assert_eq!(mode, parsed);
nym-gateway-probe/src/run.rs
+70 -48
View File
@@ -6,9 +6,11 @@ use clap::{Parser, Subcommand};
use nym_bin_common::bin_info;
use nym_config::defaults::setup_env;
use nym_gateway_probe::nodes::{NymApiDirectory, query_gateway_by_ip};
use nym_gateway_probe::{CredentialArgs, NetstackArgs, ProbeResult, TestedNode, TestedNodeDetails, TestMode};
use std::net::SocketAddr;
use nym_gateway_probe::{
CredentialArgs, NetstackArgs, ProbeResult, TestMode, TestedNode, TestedNodeDetails,
};
use nym_sdk::mixnet::NodeIdentity;
use std::net::SocketAddr;
use std::path::Path;
use std::{path::PathBuf, sync::OnceLock};
use tracing::*;
@@ -176,11 +178,17 @@ fn resolve_test_mode(
) -> anyhow::Result<TestMode> {
if let Some(mode_str) = mode_arg {
// Explicit --mode takes priority
mode_str.parse::<TestMode>()
mode_str
.parse::<TestMode>()
.map_err(|e| anyhow::anyhow!("{}", e))
} else {
// Infer from legacy flags
Ok(TestMode::from_flags(only_wireguard, only_lp_registration, test_lp_wg, has_exit_gateway))
Ok(TestMode::from_flags(
only_wireguard,
only_lp_registration,
test_lp_wg,
has_exit_gateway,
))
}
}
@@ -223,13 +231,19 @@ pub(crate) async fn run() -> anyhow::Result<ProbeResult> {
// Entry LP address: explicit or derived from gateway_ip + lp_port
let entry_lp_addr: SocketAddr = if let Some(lp_addr) = &args.entry_lp_address {
lp_addr.parse().map_err(|e| anyhow::anyhow!("Invalid entry-lp-address '{}': {}", lp_addr, e))?
lp_addr
.parse()
.map_err(|e| anyhow::anyhow!("Invalid entry-lp-address '{}': {}", lp_addr, e))?
} else if let Some(gw_ip) = &args.gateway_ip {
// Derive LP address from gateway IP
let ip: std::net::IpAddr = gw_ip.parse().map_err(|e| anyhow::anyhow!("Invalid gateway-ip '{}': {}", gw_ip, e))?;
let ip: std::net::IpAddr = gw_ip
.parse()
.map_err(|e| anyhow::anyhow!("Invalid gateway-ip '{}': {}", gw_ip, e))?;
SocketAddr::new(ip, args.lp_port)
} else {
anyhow::bail!("--entry-lp-address or --gateway-ip required with --entry-gateway-identity");
anyhow::bail!(
"--entry-lp-address or --gateway-ip required with --entry-gateway-identity"
);
};
let entry_details = TestedNodeDetails::from_cli(entry_identity, entry_lp_addr);
@@ -237,9 +251,12 @@ pub(crate) async fn run() -> anyhow::Result<ProbeResult> {
// Parse exit gateway if provided
let exit_details = if let Some(exit_identity_str) = &args.exit_gateway_identity {
let exit_identity = NodeIdentity::from_base58_string(exit_identity_str)?;
let exit_lp_addr: SocketAddr = args.exit_lp_address
let exit_lp_addr: SocketAddr = args
.exit_lp_address
.as_ref()
.ok_or_else(|| anyhow::anyhow!("--exit-lp-address required with --exit-gateway-identity"))?
.ok_or_else(|| {
anyhow::anyhow!("--exit-lp-address required with --exit-gateway-identity")
})?
.parse()
.map_err(|e| anyhow::anyhow!("Invalid exit-lp-address: {}", e))?;
Some(TestedNodeDetails::from_cli(exit_identity, exit_lp_addr))
@@ -330,48 +347,52 @@ pub(crate) async fn run() -> anyhow::Result<ProbeResult> {
}
// If gateway IP is provided, query it directly without using the directory
let (entry, directory, gateway_node, exit_gateway_node) = if let Some(gateway_ip) = args.gateway_ip.clone() {
info!("Using direct IP query mode for gateway: {}", gateway_ip);
let gateway_node = query_gateway_by_ip(gateway_ip).await?;
let identity = gateway_node.identity();
let (entry, directory, gateway_node, exit_gateway_node) =
if let Some(gateway_ip) = args.gateway_ip.clone() {
info!("Using direct IP query mode for gateway: {}", gateway_ip);
let gateway_node = query_gateway_by_ip(gateway_ip).await?;
let identity = gateway_node.identity();
// Query exit gateway if provided (for LP forwarding tests)
let exit_node = if let Some(exit_gateway_ip) = args.exit_gateway_ip {
info!("Using direct IP query mode for exit gateway: {}", exit_gateway_ip);
Some(query_gateway_by_ip(exit_gateway_ip).await?)
// Query exit gateway if provided (for LP forwarding tests)
let exit_node = if let Some(exit_gateway_ip) = args.exit_gateway_ip {
info!(
"Using direct IP query mode for exit gateway: {}",
exit_gateway_ip
);
Some(query_gateway_by_ip(exit_gateway_ip).await?)
} else {
None
};
// Still create the directory for potential secondary lookups,
// but only if API URL is available
let directory =
if let Some(api_url) = network.endpoints.first().and_then(|ep| ep.api_url()) {
Some(NymApiDirectory::new(api_url).await?)
} else {
None
};
(identity, directory, Some(gateway_node), exit_node)
} else {
None
// Original behavior: use directory service
let api_url = network
.endpoints
.first()
.and_then(|ep| ep.api_url())
.ok_or(anyhow::anyhow!("missing api url"))?;
let directory = NymApiDirectory::new(api_url).await?;
let entry = if let Some(gateway) = &args.entry_gateway {
NodeIdentity::from_base58_string(gateway)?
} else {
directory.random_exit_with_ipr()?
};
(entry, Some(directory), None, None)
};
// Still create the directory for potential secondary lookups,
// but only if API URL is available
let directory = if let Some(api_url) = network.endpoints.first().and_then(|ep| ep.api_url())
{
Some(NymApiDirectory::new(api_url).await?)
} else {
None
};
(identity, directory, Some(gateway_node), exit_node)
} else {
// Original behavior: use directory service
let api_url = network
.endpoints
.first()
.and_then(|ep| ep.api_url())
.ok_or(anyhow::anyhow!("missing api url"))?;
let directory = NymApiDirectory::new(api_url).await?;
let entry = if let Some(gateway) = &args.entry_gateway {
NodeIdentity::from_base58_string(gateway)?
} else {
directory.random_exit_with_ipr()?
};
(entry, Some(directory), None, None)
};
let test_point = if let Some(node) = args.node {
TestedNode::Custom { identity: node }
} else {
@@ -394,7 +415,8 @@ pub(crate) async fn run() -> anyhow::Result<ProbeResult> {
let (_, only_lp_registration, test_lp_wg) = mode_to_flags(test_mode);
let only_wireguard = args.only_wireguard;
let mut trial = if let (Some(entry_node), Some(exit_node)) = (&gateway_node, &exit_gateway_node) {
let mut trial = if let (Some(entry_node), Some(exit_node)) = (&gateway_node, &exit_gateway_node)
{
// Both entry and exit gateways provided (for LP telescoping tests)
info!("Using both entry and exit gateways for LP forwarding test");
nym_gateway_probe::Probe::new_with_gateways(
nym-registration-client/src/lib.rs
+4 -5
View File
@@ -189,14 +189,13 @@ impl RegistrationClient {
);
// Perform handshake with entry gateway (outer session now established)
entry_client
.perform_handshake()
.await
.map_err(|source| RegistrationClientError::EntryGatewayRegisterLp {
entry_client.perform_handshake().await.map_err(|source| {
RegistrationClientError::EntryGatewayRegisterLp {
gateway_id: self.config.entry.node.identity.to_base58_string(),
lp_address: entry_lp_address,
source: Box::new(source),
})?;
}
})?;
tracing::info!("Outer session with entry gateway established");
nym-registration-client/src/lp_client/client.rs
+12 -12
View File
@@ -10,7 +10,7 @@ use nym_bandwidth_controller::{BandwidthTicketProvider, DEFAULT_TICKETS_TO_SPEND
use nym_credentials_interface::{CredentialSpendingData, TicketType};
use nym_crypto::asymmetric::{ed25519, x25519};
use nym_lp::LpPacket;
use nym_lp::codec::{parse_lp_packet, serialize_lp_packet, OuterAeadKey};
use nym_lp::codec::{OuterAeadKey, parse_lp_packet, serialize_lp_packet};
use nym_lp::message::ForwardPacketData;
use nym_lp::state_machine::{LpAction, LpInput, LpStateMachine};
use nym_registration_common::{GatewayData, LpRegistrationRequest, LpRegistrationResponse};
@@ -271,8 +271,11 @@ impl LpRegistrationClient {
.ok()
.and_then(|s| s.outer_aead_key());
tracing::trace!("Sending handshake packet (send_key={}, recv_key={})",
send_key.is_some(), recv_key.is_some());
tracing::trace!(
"Sending handshake packet (send_key={}, recv_key={})",
send_key.is_some(),
recv_key.is_some()
);
let response = Self::connect_send_receive(
self.gateway_lp_address,
&packet,
@@ -284,8 +287,7 @@ impl LpRegistrationClient {
tracing::trace!("Received handshake response");
// Process the received packet
if let Some(action) =
state_machine.process_input(LpInput::ReceivePacket(response))
if let Some(action) = state_machine.process_input(LpInput::ReceivePacket(response))
{
match action? {
LpAction::SendPacket(response_packet) => {
@@ -344,10 +346,10 @@ impl LpRegistrationClient {
.session()?
.prepare_handshake_message()
.ok_or_else(|| {
LpClientError::Transport(
"No handshake message available after KKT".to_string(),
)
})??;
LpClientError::Transport(
"No handshake message available after KKT".to_string(),
)
})??;
let noise_packet = state_machine.session()?.next_packet(noise_msg)?;
pending_packet = Some(noise_packet);
}
@@ -596,9 +598,7 @@ impl LpRegistrationClient {
) -> Result<GatewayData> {
// Ensure handshake is complete (state machine exists)
let state_machine = self.state_machine.as_mut().ok_or_else(|| {
LpClientError::Transport(
"Cannot register: handshake not completed".to_string(),
)
LpClientError::Transport("Cannot register: handshake not completed".to_string())
})?;
tracing::debug!("Sending registration request (packet-per-connection)");
nym-registration-client/src/lp_client/nested_session.rs
+56 -60
View File
@@ -24,7 +24,7 @@ use bytes::BytesMut;
use nym_bandwidth_controller::BandwidthTicketProvider;
use nym_credentials_interface::TicketType;
use nym_crypto::asymmetric::{ed25519, x25519};
use nym_lp::codec::{parse_lp_packet, serialize_lp_packet, OuterAeadKey};
use nym_lp::codec::{OuterAeadKey, parse_lp_packet, serialize_lp_packet};
use nym_lp::state_machine::{LpAction, LpInput, LpStateMachine};
use nym_lp::{LpMessage, LpPacket};
use nym_registration_common::{GatewayData, LpRegistrationRequest, LpRegistrationResponse};
@@ -110,23 +110,16 @@ impl NestedLpSession {
/// - Forwarding through entry gateway fails
/// - Exit gateway handshake fails
/// - Cryptographic operations fail
async fn perform_handshake(
&mut self,
outer_client: &mut LpRegistrationClient,
) -> Result<()> {
async fn perform_handshake(&mut self, outer_client: &mut LpRegistrationClient) -> Result<()> {
tracing::debug!(
"Starting nested LP handshake with exit gateway {}",
self.exit_address
);
// Step 1: Derive X25519 keys from Ed25519 for Noise protocol
let client_x25519_public = self
.client_keypair
.public_key()
.to_x25519()
.map_err(|e| {
LpClientError::Crypto(format!("Failed to derive X25519 public key: {}", e))
})?;
let client_x25519_public = self.client_keypair.public_key().to_x25519().map_err(|e| {
LpClientError::Crypto(format!("Failed to derive X25519 public key: {}", e))
})?;
// Step 2: Generate ClientHello for exit gateway
let client_hello_data = nym_lp::ClientHelloData::new_with_fresh_salt(
@@ -144,7 +137,7 @@ impl NestedLpSession {
// Step 3: Send ClientHello to exit gateway via forwarding
let client_hello_header = nym_lp::packet::LpHeader::new(
nym_lp::BOOTSTRAP_RECEIVER_IDX, // Use constant for bootstrap session
0, // counter starts at 0
0, // counter starts at 0
);
let client_hello_packet = nym_lp::LpPacket::new(
client_hello_header,
@@ -219,8 +212,7 @@ impl NestedLpSession {
tracing::trace!("Received handshake response from exit");
// Process the received packet
if let Some(action) =
state_machine.process_input(LpInput::ReceivePacket(response))
if let Some(action) = state_machine.process_input(LpInput::ReceivePacket(response))
{
match action? {
LpAction::SendPacket(response_packet) => {
@@ -238,9 +230,7 @@ impl NestedLpSession {
)
.await?;
}
tracing::info!(
"Nested LP handshake completed with exit gateway"
);
tracing::info!("Nested LP handshake completed with exit gateway");
break;
}
}
@@ -255,10 +245,10 @@ impl NestedLpSession {
.session()?
.prepare_handshake_message()
.ok_or_else(|| {
LpClientError::Transport(
"No handshake message available after KKT".to_string(),
)
})??;
LpClientError::Transport(
"No handshake message available after KKT".to_string(),
)
})??;
let noise_packet = state_machine.session()?.next_packet(noise_msg)?;
pending_packet = Some(noise_packet);
}
@@ -311,11 +301,14 @@ impl NestedLpSession {
LpClientError::Transport("State machine missing after handshake".to_string())
})?;
tracing::debug!("Building registration request for exit gateway (with pre-made credential)");
tracing::debug!(
"Building registration request for exit gateway (with pre-made credential)"
);
// Step 3: Build registration request (credential already provided)
let wg_public_key = PeerPublicKey::new(wg_keypair.public_key().to_bytes().into());
let request = LpRegistrationRequest::new_dvpn(wg_public_key, credential, ticket_type, client_ip);
let request =
LpRegistrationRequest::new_dvpn(wg_public_key, credential, ticket_type, client_ip);
tracing::trace!("Built registration request: {:?}", request);
@@ -336,15 +329,15 @@ impl NestedLpSession {
LpClientError::Transport("State machine returned no action".to_string())
})?
.map_err(|e| {
LpClientError::Transport(format!(
"Failed to encrypt registration request: {}",
e
))
LpClientError::Transport(format!("Failed to encrypt registration request: {}", e))
})?;
// Step 6: Send the encrypted packet via forwarding
// Get outer key for AEAD encryption (PSK is available after handshake)
let outer_key = state_machine.session().ok().and_then(|s| s.outer_aead_key_for_sending());
let outer_key = state_machine
.session()
.ok()
.and_then(|s| s.outer_aead_key_for_sending());
let response_bytes = match action {
LpAction::SendPacket(packet) => {
let packet_bytes = Self::serialize_packet(&packet, outer_key.as_ref())?;
@@ -367,7 +360,10 @@ impl NestedLpSession {
tracing::trace!("Received registration response from exit gateway");
// Step 7: Parse response bytes to LP packet
let outer_key = state_machine.session().ok().and_then(|s| s.outer_aead_key());
let outer_key = state_machine
.session()
.ok()
.and_then(|s| s.outer_aead_key());
let response_packet = Self::parse_packet(&response_bytes, outer_key.as_ref())?;
// Step 8: Decrypt via state machine
@@ -377,10 +373,7 @@ impl NestedLpSession {
LpClientError::Transport("State machine returned no action".to_string())
})?
.map_err(|e| {
LpClientError::Transport(format!(
"Failed to decrypt registration response: {}",
e
))
LpClientError::Transport(format!("Failed to decrypt registration response: {}", e))
})?;
// Step 9: Extract decrypted data
@@ -480,19 +473,21 @@ impl NestedLpSession {
// Step 3: Acquire bandwidth credential
let credential = bandwidth_controller
.get_ecash_ticket(ticket_type, *gateway_identity, nym_bandwidth_controller::DEFAULT_TICKETS_TO_SPEND)
.get_ecash_ticket(
ticket_type,
*gateway_identity,
nym_bandwidth_controller::DEFAULT_TICKETS_TO_SPEND,
)
.await
.map_err(|e| {
LpClientError::Transport(format!(
"Failed to acquire bandwidth credential: {}",
e
))
LpClientError::Transport(format!("Failed to acquire bandwidth credential: {}", e))
})?
.data;
// Step 4: Build registration request
let wg_public_key = PeerPublicKey::new(wg_keypair.public_key().to_bytes().into());
let request = LpRegistrationRequest::new_dvpn(wg_public_key, credential, ticket_type, client_ip);
let request =
LpRegistrationRequest::new_dvpn(wg_public_key, credential, ticket_type, client_ip);
tracing::trace!("Built registration request: {:?}", request);
@@ -513,15 +508,15 @@ impl NestedLpSession {
LpClientError::Transport("State machine returned no action".to_string())
})?
.map_err(|e| {
LpClientError::Transport(format!(
"Failed to encrypt registration request: {}",
e
))
LpClientError::Transport(format!("Failed to encrypt registration request: {}", e))
})?;
// Step 7: Send the encrypted packet via forwarding
// Get outer key for AEAD encryption (PSK is available after handshake)
let outer_key = state_machine.session().ok().and_then(|s| s.outer_aead_key_for_sending());
let outer_key = state_machine
.session()
.ok()
.and_then(|s| s.outer_aead_key_for_sending());
let response_bytes = match action {
LpAction::SendPacket(packet) => {
let packet_bytes = Self::serialize_packet(&packet, outer_key.as_ref())?;
@@ -544,7 +539,10 @@ impl NestedLpSession {
tracing::trace!("Received registration response from exit gateway");
// Step 8: Parse response bytes to LP packet
let outer_key = state_machine.session().ok().and_then(|s| s.outer_aead_key());
let outer_key = state_machine
.session()
.ok()
.and_then(|s| s.outer_aead_key());
let response_packet = Self::parse_packet(&response_bytes, outer_key.as_ref())?;
// Step 9: Decrypt via state machine
@@ -554,10 +552,7 @@ impl NestedLpSession {
LpClientError::Transport("State machine returned no action".to_string())
})?
.map_err(|e| {
LpClientError::Transport(format!(
"Failed to decrypt registration response: {}",
e
))
LpClientError::Transport(format!("Failed to decrypt registration response: {}", e))
})?;
// Step 10: Extract decrypted data
@@ -623,16 +618,18 @@ impl NestedLpSession {
packet: &LpPacket,
) -> Result<LpPacket> {
// Use outer_aead_key_for_sending() for send, outer_aead_key() for receive
let send_key = state_machine.session().ok().and_then(|s| s.outer_aead_key_for_sending());
let send_key = state_machine
.session()
.ok()
.and_then(|s| s.outer_aead_key_for_sending());
let packet_bytes = Self::serialize_packet(packet, send_key.as_ref())?;
let response_bytes = outer_client
.send_forward_packet(
self.exit_identity,
self.exit_address.clone(),
packet_bytes,
)
.send_forward_packet(self.exit_identity, self.exit_address.clone(), packet_bytes)
.await?;
let recv_key = state_machine.session().ok().and_then(|s| s.outer_aead_key());
let recv_key = state_machine
.session()
.ok()
.and_then(|s| s.outer_aead_key());
Self::parse_packet(&response_bytes, recv_key.as_ref())
}
@@ -667,8 +664,7 @@ impl NestedLpSession {
/// Returns an error if parsing fails
fn parse_packet(bytes: &[u8], outer_key: Option<&OuterAeadKey>) -> Result<LpPacket> {
// Use outer AEAD key when available (after PSK derivation)
parse_lp_packet(bytes, outer_key).map_err(|e| {
LpClientError::Transport(format!("Failed to parse LP packet: {}", e))
})
parse_lp_packet(bytes, outer_key)
.map_err(|e| LpClientError::Transport(format!("Failed to parse LP packet: {}", e)))
}
}