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

fmt and metrics

Browse Source
This commit is contained in:
durch
2025-10-23 20:14:57 +02:00
parent ec90a218df
commit de06f4a5c0
25 changed files with 584 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -5840,6 +5840,7 @@ dependencies = [
"nym-ip-packet-router",
"nym-kcp",
"nym-lp",
"nym-metrics",
"nym-mixnet-client",
"nym-network-defaults",
"nym-network-requester",
common/commands/src/validator/mixnet/operators/nymnode/bond_nymnode.rs
+1 -3
View File
@@ -60,9 +60,7 @@ pub async fn bond_nymnode(args: Args, client: SigningClient) {
return;
}
let lp_address = args.lp_port.map(|port| {
format!("{}:{}", args.host, port)
});
let lp_address = args.lp_port.map(|port| format!("{}:{}", args.host, port));
let nymnode = nym_mixnet_contract_common::NymNode {
host: args.host,
common/commands/src/validator/mixnet/operators/nymnode/nymnode_bonding_sign_payload.rs
+1 -3
View File
@@ -50,9 +50,7 @@ pub struct Args {
pub async fn create_payload(args: Args, client: SigningClient) {
let denom = client.current_chain_details().mix_denom.base.as_str();
let lp_address = args.lp_port.map(|port| {
format!("{}:{}", args.host, port)
});
let lp_address = args.lp_port.map(|port| format!("{}:{}", args.host, port));
let mixnode = nym_mixnet_contract_common::NymNode {
host: args.host,
common/commands/src/validator/mixnet/operators/nymnode/settings/update_config.rs
+4 -1
View File
@@ -20,7 +20,10 @@ pub struct Args {
#[clap(long)]
pub restore_default_http_port: bool,
#[clap(long, help = "LP (Lewes Protocol) listener address (format: host:port)")]
#[clap(
long,
help = "LP (Lewes Protocol) listener address (format: host:port)"
)]
pub lp_address: Option<String>,
// equivalent to setting `lp_address` to `None`
common/crypto/src/asymmetric/ed25519/mod.rs
+10 -9
View File
@@ -228,20 +228,21 @@ impl PublicKey {
// Decompress the Ed25519 point
let compressed = CompressedEdwardsY((*self).to_bytes());
let edwards_point = compressed
.decompress()
.ok_or_else(|| Ed25519RecoveryError::MalformedBytes(
SignatureError::from_source("Failed to decompress Ed25519 point".to_string())
))?;
let edwards_point = compressed.decompress().ok_or_else(|| {
Ed25519RecoveryError::MalformedBytes(SignatureError::from_source(
"Failed to decompress Ed25519 point".to_string(),
))
})?;
// Convert to Montgomery form
let montgomery = edwards_point.to_montgomery();
// Create X25519 public key
crate::asymmetric::x25519::PublicKey::from_bytes(montgomery.as_bytes())
.map_err(|_| Ed25519RecoveryError::MalformedBytes(
SignatureError::from_source("Failed to convert to X25519".to_string())
crate::asymmetric::x25519::PublicKey::from_bytes(montgomery.as_bytes()).map_err(|_| {
Ed25519RecoveryError::MalformedBytes(SignatureError::from_source(
"Failed to convert to X25519".to_string(),
))
})
}
}
@@ -373,7 +374,7 @@ impl PrivateKey {
/// # Returns
/// The converted X25519 private key
pub fn to_x25519(&self) -> crate::asymmetric::x25519::PrivateKey {
use sha2::{Sha512, Digest};
use sha2::{Digest, Sha512};
// Hash the Ed25519 secret key with SHA-512
let hash = Sha512::digest(self.0);
common/crypto/src/kdf.rs
+8 -2
View File
@@ -52,7 +52,10 @@ mod tests {
let key1 = derive_key_blake3("context1", key_material, salt);
let key2 = derive_key_blake3("context2", key_material, salt);
assert_ne!(key1, key2, "Different contexts should produce different keys");
assert_ne!(
key1, key2,
"Different contexts should produce different keys"
);
}
#[test]
@@ -74,7 +77,10 @@ mod tests {
let key1 = derive_key_blake3(context, b"secret1", salt);
let key2 = derive_key_blake3(context, b"secret2", salt);
assert_ne!(key1, key2, "Different key material should produce different keys");
assert_ne!(
key1, key2,
"Different key material should produce different keys"
);
}
#[test]
common/nym-kcp/src/session.rs
+1 -1
View File
@@ -169,7 +169,7 @@ impl KcpSession {
let size = std::cmp::min(self.mtu, data.len());
let chunk = &data[..size];
// AIDEV-NOTE: KCP fragment numbering is REVERSED - last fragment has frg=0,
// KCP fragment numbering is REVERSED - last fragment has frg=0,
// first has frg=count-1. This allows receiver to know total count from first packet.
// In KCP, `frg` is set to the remaining fragments in reverse order.
// i.e., the last fragment has frg=0, the first has frg=count-1.
common/nym-lp/src/message.rs
+3 -6
View File
@@ -24,10 +24,7 @@ impl ClientHelloData {
/// # Arguments
/// * `client_lp_public_key` - Client's x25519 public key
/// * `protocol_version` - Protocol version number
pub fn new_with_fresh_salt(
client_lp_public_key: [u8; 32],
protocol_version: u8,
) -> Self {
pub fn new_with_fresh_salt(client_lp_public_key: [u8; 32], protocol_version: u8) -> Self {
use std::time::{SystemTime, UNIX_EPOCH};
// Generate salt: timestamp + nonce
@@ -159,8 +156,8 @@ impl LpMessage {
}
LpMessage::ClientHello(data) => {
// Serialize ClientHelloData using bincode
let serialized = bincode::serialize(data)
.expect("Failed to serialize ClientHelloData");
let serialized =
bincode::serialize(data).expect("Failed to serialize ClientHelloData");
dst.put_slice(&serialized);
}
}
common/nym-lp/src/psk.rs
+4 -20
View File
@@ -69,16 +69,8 @@ mod tests {
let salt = [1u8; 32];
// Derive PSK twice with same inputs
let psk1 = derive_psk(
keypair_1.private_key(),
keypair_2.public_key(),
&salt,
);
let psk2 = derive_psk(
keypair_1.private_key(),
keypair_2.public_key(),
&salt,
);
let psk1 = derive_psk(keypair_1.private_key(), keypair_2.public_key(), &salt);
let psk2 = derive_psk(keypair_1.private_key(), keypair_2.public_key(), &salt);
assert_eq!(psk1, psk2, "Same inputs should produce same PSK");
}
@@ -90,18 +82,10 @@ mod tests {
let salt = [2u8; 32];
// Client derives PSK
let client_psk = derive_psk(
keypair_1.private_key(),
keypair_2.public_key(),
&salt,
);
let client_psk = derive_psk(keypair_1.private_key(), keypair_2.public_key(), &salt);
// Gateway derives PSK from their perspective
let gateway_psk = derive_psk(
keypair_2.private_key(),
keypair_1.public_key(),
&salt,
);
let gateway_psk = derive_psk(keypair_2.private_key(), keypair_1.public_key(), &salt);
assert_eq!(
client_psk, gateway_psk,
common/nym-lp/src/session.rs
+4 -3
View File
@@ -68,7 +68,7 @@ impl LpSession {
remote_public_key: &[u8],
psk: &[u8],
) -> Result<Self, LpError> {
// AIDEV-NOTE: XKpsk3 pattern requires remote static key known upfront (XK)
// XKpsk3 pattern requires remote static key known upfront (XK)
// and PSK mixed at position 3. This provides forward secrecy with PSK authentication.
let pattern_name = "Noise_XKpsk3_25519_ChaChaPoly_SHA256";
let psk_index = 3;
@@ -125,7 +125,7 @@ impl LpSession {
/// * `Ok(())` if the counter is likely valid
/// * `Err(LpError::Replay)` if the counter is invalid or a potential replay
pub fn receiving_counter_quick_check(&self, counter: u64) -> Result<(), LpError> {
// AIDEV-NOTE: Branchless implementation uses SIMD when available for constant-time
// Branchless implementation uses SIMD when available for constant-time
// operations, preventing timing attacks. Check before crypto to save CPU cycles.
let counter_validator = self.receiving_counter.lock();
counter_validator
@@ -455,7 +455,8 @@ mod tests {
const MAX_ROUNDS: usize = 10; // Safety break for the loop
// Start by priming the initiator message
let mut initiator_to_responder_msg = initiator_session.prepare_handshake_message().unwrap().ok();
let mut initiator_to_responder_msg =
initiator_session.prepare_handshake_message().unwrap().ok();
assert!(
initiator_to_responder_msg.is_some(),
"Initiator did not produce initial message"
common/registration/src/lp_messages.rs
+17 -11
View File
@@ -100,11 +100,7 @@ impl LpRegistrationRequest {
impl LpRegistrationResponse {
/// Create a success response with GatewayData
pub fn success(
session_id: u32,
allocated_bandwidth: i64,
gateway_data: GatewayData,
) -> Self {
pub fn success(session_id: u32, allocated_bandwidth: i64, gateway_data: GatewayData) -> Self {
Self {
success: true,
error: None,
@@ -137,14 +133,15 @@ mod tests {
use std::net::Ipv6Addr;
GatewayData {
public_key: nym_crypto::asymmetric::x25519::PublicKey::from(nym_sphinx::PublicKey::from([1u8; 32])),
public_key: nym_crypto::asymmetric::x25519::PublicKey::from(
nym_sphinx::PublicKey::from([1u8; 32]),
),
private_ipv4: Ipv4Addr::new(10, 0, 0, 1),
private_ipv6: Ipv6Addr::new(0xfc00, 0, 0, 0, 0, 0, 0, 1),
endpoint: "192.168.1.1:8080".parse().expect("Valid test endpoint"),
}
}
// ==================== LpRegistrationRequest Tests ====================
// ==================== LpRegistrationResponse Tests ====================
@@ -155,7 +152,8 @@ mod tests {
let session_id = 12345;
let allocated_bandwidth = 1_000_000_000;
let response = LpRegistrationResponse::success(session_id, allocated_bandwidth, gateway_data.clone());
let response =
LpRegistrationResponse::success(session_id, allocated_bandwidth, gateway_data.clone());
assert!(response.success);
assert!(response.error.is_none());
@@ -163,7 +161,9 @@ mod tests {
assert_eq!(response.allocated_bandwidth, allocated_bandwidth);
assert_eq!(response.session_id, session_id);
let returned_gw_data = response.gateway_data.expect("Gateway data should be present in success response");
let returned_gw_data = response
.gateway_data
.expect("Gateway data should be present in success response");
assert_eq!(returned_gw_data.public_key, gateway_data.public_key);
assert_eq!(returned_gw_data.private_ipv4, gateway_data.private_ipv4);
assert_eq!(returned_gw_data.private_ipv6, gateway_data.private_ipv6);
@@ -198,7 +198,10 @@ mod tests {
assert_eq!(deserialized.success, original.success);
assert_eq!(deserialized.error, original.error);
assert_eq!(deserialized.allocated_bandwidth, original.allocated_bandwidth);
assert_eq!(
deserialized.allocated_bandwidth,
original.allocated_bandwidth
);
assert_eq!(deserialized.session_id, original.session_id);
assert!(deserialized.gateway_data.is_some());
}
@@ -229,7 +232,10 @@ mod tests {
// Attempt to deserialize
let result: Result<LpRegistrationResponse, _> = bincode::deserialize(&invalid_data);
assert!(result.is_err(), "Expected deserialization to fail for malformed data");
assert!(
result.is_err(),
"Expected deserialization to fail for malformed data"
);
}
// ==================== RegistrationMode Tests ====================
gateway/Cargo.toml
+4
View File
@@ -64,7 +64,11 @@ nym-topology = { path = "../common/topology" }
nym-validator-client = { path = "../common/client-libs/validator-client" }
nym-ip-packet-router = { path = "../service-providers/ip-packet-router" }
nym-node-metrics = { path = "../nym-node/nym-node-metrics" }
<<<<<<< HEAD
nym-upgrade-mode-check = { path = "../common/upgrade-mode-check" }
=======
nym-metrics = { path = "../common/nym-metrics" }
>>>>>>> 6ac28abef (fmt and metrics)
nym-wireguard = { path = "../common/wireguard" }
nym-wireguard-private-metadata-server = { path = "../common/wireguard-private-metadata/server" }
gateway/src/node/client_handling/websocket/common_state.rs
+2 -1
View File
@@ -23,7 +23,8 @@ pub(crate) struct Config {
#[derive(Clone)]
pub(crate) struct CommonHandlerState {
pub(crate) cfg: Config,
pub(crate) ecash_verifier: Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
pub(crate) ecash_verifier:
Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
pub(crate) storage: GatewayStorage,
pub(crate) local_identity: Arc<ed25519::KeyPair>,
pub(crate) metrics: NymNodeMetrics,
gateway/src/node/internal_service_providers/authenticator/mod.rs
+3 -1
View File
@@ -51,7 +51,9 @@ impl Authenticator {
upgrade_mode_state: UpgradeModeDetails,
wireguard_gateway_data: WireguardGatewayData,
used_private_network_ips: Vec<IpAddr>,
ecash_verifier: Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
ecash_verifier: Arc<
dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync,
>,
shutdown: ShutdownTracker,
) -> Self {
Self {
gateway/src/node/lp_listener/handler.rs
+184 -107
View File
@@ -10,11 +10,29 @@ use nym_lp::{
keypair::{Keypair, PublicKey},
LpMessage, LpPacket, LpSession,
};
use nym_metrics::{add_histogram_obs, inc};
use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tracing::*;
// Histogram buckets for LP operation duration tracking
// Covers typical LP operations from 10ms to 10 seconds
// - Most handshakes should complete in < 100ms
// - Registration with credential verification typically 100ms - 1s
// - Slow operations (network issues, DB contention) up to 10s
const LP_DURATION_BUCKETS: &[f64] = &[
0.01, // 10ms
0.05, // 50ms
0.1, // 100ms
0.25, // 250ms
0.5, // 500ms
1.0, // 1s
2.5, // 2.5s
5.0, // 5s
10.0, // 10s
];
pub struct LpConnectionHandler {
stream: TcpStream,
remote_addr: SocketAddr,
@@ -33,6 +51,9 @@ impl LpConnectionHandler {
pub async fn handle(mut self) -> Result<(), GatewayError> {
debug!("Handling LP connection from {}", self.remote_addr);
// Track total LP connections handled
inc!("lp_connections_total");
// For LP, we need:
// 1. Gateway's keypair (from local_identity)
// 2. Client's public key (will be received during handshake)
@@ -45,51 +66,80 @@ impl LpConnectionHandler {
// Receive client's public key and salt via ClientHello message
// The client initiates by sending ClientHello as first packet
let (client_pubkey, salt) = self.receive_client_hello().await?;
let (client_pubkey, salt) = match self.receive_client_hello().await {
Ok(result) => result,
Err(e) => {
// Track ClientHello failures (timestamp validation, protocol errors, etc.)
inc!("lp_client_hello_failed");
return Err(e);
}
};
// Derive PSK using ECDH + Blake3 KDF (nym-109)
// Both client and gateway derive the same PSK from their respective keys
let psk = nym_lp::derive_psk(
gateway_keypair.private_key(),
&client_pubkey,
&salt,
);
let psk = nym_lp::derive_psk(gateway_keypair.private_key(), &client_pubkey, &salt);
tracing::trace!("Derived PSK from LP keys and ClientHello salt");
// Create LP handshake as responder
let handshake = LpGatewayHandshake::new_responder(
&gateway_keypair,
&client_pubkey,
&psk,
)?;
let handshake = LpGatewayHandshake::new_responder(&gateway_keypair, &client_pubkey, &psk)?;
// Complete the LP handshake
let session = handshake.complete(&mut self.stream).await?;
// Complete the LP handshake with duration tracking
let handshake_start = std::time::Instant::now();
let session = match handshake.complete(&mut self.stream).await {
Ok(s) => {
let duration = handshake_start.elapsed().as_secs_f64();
add_histogram_obs!(
"lp_handshake_duration_seconds",
duration,
LP_DURATION_BUCKETS
);
inc!("lp_handshakes_success");
s
}
Err(e) => {
inc!("lp_handshakes_failed");
inc!("lp_errors_handshake");
return Err(e);
}
};
info!("LP handshake completed for {} (session {})",
self.remote_addr, session.id());
info!(
"LP handshake completed for {} (session {})",
self.remote_addr,
session.id()
);
// After handshake, receive registration request
let request = self.receive_registration_request(&session).await?;
debug!("LP registration request from {}: mode={:?}",
self.remote_addr, request.mode);
debug!(
"LP registration request from {}: mode={:?}",
self.remote_addr, request.mode
);
// Process registration (verify credentials, add peer, etc.)
let response = process_registration(request, &self.state).await;
// Send response
if let Err(e) = self.send_registration_response(&session, response.clone()).await {
if let Err(e) = self
.send_registration_response(&session, response.clone())
.await
{
warn!("Failed to send LP response to {}: {}", self.remote_addr, e);
inc!("lp_errors_send_response");
return Err(e);
}
if response.success {
info!("LP registration successful for {} (session {})",
self.remote_addr, response.session_id);
info!(
"LP registration successful for {} (session {})",
self.remote_addr, response.session_id
);
} else {
warn!("LP registration failed for {}: {:?}",
self.remote_addr, response.error);
warn!(
"LP registration failed for {}: {:?}",
self.remote_addr, response.error
);
}
Ok(())
@@ -131,12 +181,23 @@ impl LpConnectionHandler {
} else {
"future"
};
// Track timestamp validation failures
inc!("lp_timestamp_validation_rejected");
if now >= client_timestamp {
inc!("lp_errors_timestamp_too_old");
} else {
inc!("lp_errors_timestamp_too_far_future");
}
return Err(GatewayError::LpProtocolError(format!(
"ClientHello timestamp is too {} (age: {}s, tolerance: {}s)",
direction, age, tolerance_secs
)));
}
// Track successful timestamp validation
inc!("lp_timestamp_validation_accepted");
Ok(())
}
@@ -150,9 +211,10 @@ impl LpConnectionHandler {
LpMessage::ClientHello(hello_data) => {
// Validate protocol version (currently only v1)
if hello_data.protocol_version != 1 {
return Err(GatewayError::LpProtocolError(
format!("Unsupported protocol version: {}", hello_data.protocol_version)
));
return Err(GatewayError::LpProtocolError(format!(
"Unsupported protocol version: {}",
hello_data.protocol_version
)));
}
// Extract and validate timestamp (nym-110: replay protection)
@@ -179,20 +241,19 @@ impl LpConnectionHandler {
// Convert bytes to PublicKey
let client_pubkey = PublicKey::from_bytes(&hello_data.client_lp_public_key)
.map_err(|e| GatewayError::LpProtocolError(
format!("Invalid client public key: {}", e)
))?;
.map_err(|e| {
GatewayError::LpProtocolError(format!("Invalid client public key: {}", e))
})?;
// Extract salt for PSK derivation
let salt = hello_data.salt;
Ok((client_pubkey, salt))
}
other => {
Err(GatewayError::LpProtocolError(
format!("Expected ClientHello, got {}", other)
))
}
other => Err(GatewayError::LpProtocolError(format!(
"Expected ClientHello, got {}",
other
))),
}
}
@@ -206,26 +267,28 @@ impl LpConnectionHandler {
// Verify it's from the correct session
if packet.header().session_id != session.id() {
return Err(GatewayError::LpProtocolError(
format!("Session ID mismatch: expected {}, got {}",
session.id(), packet.header().session_id)
));
return Err(GatewayError::LpProtocolError(format!(
"Session ID mismatch: expected {}, got {}",
session.id(),
packet.header().session_id
)));
}
// Extract registration request from LP message
match packet.message() {
LpMessage::EncryptedData(data) => {
// Deserialize registration request
bincode::deserialize(&data)
.map_err(|e| GatewayError::LpProtocolError(
format!("Failed to deserialize registration request: {}", e)
bincode::deserialize(&data).map_err(|e| {
GatewayError::LpProtocolError(format!(
"Failed to deserialize registration request: {}",
e
))
})
}
other => {
Err(GatewayError::LpProtocolError(
format!("Expected EncryptedData message, got {:?}", other)
))
}
other => Err(GatewayError::LpProtocolError(format!(
"Expected EncryptedData message, got {:?}",
other
))),
}
}
@@ -236,16 +299,14 @@ impl LpConnectionHandler {
response: LpRegistrationResponse,
) -> Result<(), GatewayError> {
// Serialize response
let data = bincode::serialize(&response)
.map_err(|e| GatewayError::LpProtocolError(
format!("Failed to serialize response: {}", e)
))?;
let data = bincode::serialize(&response).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to serialize response: {}", e))
})?;
// Create LP packet with response
let packet = session.create_data_packet(data)
.map_err(|e| GatewayError::LpProtocolError(
format!("Failed to create data packet: {}", e)
))?;
let packet = session.create_data_packet(data).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to create data packet: {}", e))
})?;
// Send the packet
self.send_lp_packet(&packet).await
@@ -257,63 +318,59 @@ impl LpConnectionHandler {
// Read 4-byte length prefix (u32 big-endian)
let mut len_buf = [0u8; 4];
self.stream.read_exact(&mut len_buf).await
.map_err(|e| GatewayError::LpConnectionError(
format!("Failed to read packet length: {}", e)
))?;
self.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)
));
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];
self.stream.read_exact(&mut packet_buf).await
.map_err(|e| GatewayError::LpConnectionError(
format!("Failed to read packet data: {}", e)
))?;
self.stream.read_exact(&mut packet_buf).await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to read packet data: {}", e))
})?;
parse_lp_packet(&packet_buf)
.map_err(|e| GatewayError::LpProtocolError(
format!("Failed to parse LP packet: {}", e)
))
.map_err(|e| GatewayError::LpProtocolError(format!("Failed to parse LP packet: {}", e)))
}
/// Send an LP packet over the stream with proper length-prefixed framing
async fn send_lp_packet(&mut self, packet: &LpPacket) -> Result<(), GatewayError> {
use nym_lp::codec::serialize_lp_packet;
use bytes::BytesMut;
use nym_lp::codec::serialize_lp_packet;
// Serialize the packet first
let mut packet_buf = BytesMut::new();
serialize_lp_packet(packet, &mut packet_buf)
.map_err(|e| GatewayError::LpProtocolError(
format!("Failed to serialize packet: {}", e)
))?;
serialize_lp_packet(packet, &mut packet_buf).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;
self.stream.write_all(&len.to_be_bytes()).await
.map_err(|e| GatewayError::LpConnectionError(
format!("Failed to send packet length: {}", e)
))?;
self.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
self.stream.write_all(&packet_buf).await
.map_err(|e| GatewayError::LpConnectionError(
format!("Failed to send packet data: {}", e)
))?;
self.stream.write_all(&packet_buf).await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to send packet data: {}", e))
})?;
self.stream.flush().await
.map_err(|e| GatewayError::LpConnectionError(
format!("Failed to flush stream: {}", e)
))?;
self.stream.flush().await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to flush stream: {}", e))
})?;
Ok(())
}
@@ -344,15 +401,15 @@ impl LpSessionExt for LpSession {
#[cfg(test)]
mod tests {
use super::*;
use crate::node::lp_listener::LpConfig;
use crate::node::ActiveClientsStore;
use bytes::BytesMut;
use nym_lp::codec::{parse_lp_packet, serialize_lp_packet};
use nym_lp::keypair::Keypair;
use nym_lp::message::{ClientHelloData, LpMessage};
use nym_lp::packet::{LpHeader, LpPacket};
use nym_lp::codec::{serialize_lp_packet, parse_lp_packet};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use crate::node::ActiveClientsStore;
use crate::node::lp_listener::LpConfig;
// ==================== Test Helpers ====================
@@ -367,9 +424,8 @@ mod tests {
.expect("Failed to create test storage");
// Create mock ecash manager for testing
let ecash_verifier = nym_credential_verification::ecash::MockEcashManager::new(
Box::new(storage.clone())
);
let ecash_verifier =
nym_credential_verification::ecash::MockEcashManager::new(Box::new(storage.clone()));
LpHandlerState {
lp_config: LpConfig {
@@ -377,7 +433,8 @@ mod tests {
timestamp_tolerance_secs: 30,
..Default::default()
},
ecash_verifier: Arc::new(ecash_verifier) as Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
ecash_verifier: Arc::new(ecash_verifier)
as Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
storage,
local_identity: Arc::new(ed25519::KeyPair::new(&mut OsRng)),
metrics: nym_node_metrics::NymNodeMetrics::default(),
@@ -538,7 +595,9 @@ mod tests {
},
LpMessage::Busy,
);
write_lp_packet_to_stream(&mut client_stream, &packet).await.unwrap();
write_lp_packet_to_stream(&mut client_stream, &packet)
.await
.unwrap();
// Handler should receive and parse it correctly
let received = server_task.await.unwrap().unwrap();
@@ -565,7 +624,10 @@ mod tests {
// Send a packet size that exceeds MAX_PACKET_SIZE (64KB)
let oversized_len: u32 = 70000; // > 65536
client_stream.write_all(&oversized_len.to_be_bytes()).await.unwrap();
client_stream
.write_all(&oversized_len.to_be_bytes())
.await
.unwrap();
client_stream.flush().await.unwrap();
// Handler should reject it
@@ -604,7 +666,9 @@ mod tests {
server_task.await.unwrap().unwrap();
// Client should receive it correctly
let received = read_lp_packet_from_stream(&mut client_stream).await.unwrap();
let received = read_lp_packet_from_stream(&mut client_stream)
.await
.unwrap();
assert_eq!(received.header().session_id, 99);
assert_eq!(received.header().counter, 5);
}
@@ -638,7 +702,9 @@ mod tests {
let mut client_stream = TcpStream::connect(addr).await.unwrap();
server_task.await.unwrap().unwrap();
let received = read_lp_packet_from_stream(&mut client_stream).await.unwrap();
let received = read_lp_packet_from_stream(&mut client_stream)
.await
.unwrap();
assert_eq!(received.header().session_id, 100);
assert_eq!(received.header().counter, 10);
match received.message() {
@@ -676,7 +742,9 @@ mod tests {
let mut client_stream = TcpStream::connect(addr).await.unwrap();
server_task.await.unwrap().unwrap();
let received = read_lp_packet_from_stream(&mut client_stream).await.unwrap();
let received = read_lp_packet_from_stream(&mut client_stream)
.await
.unwrap();
assert_eq!(received.header().session_id, 200);
assert_eq!(received.header().counter, 20);
match received.message() {
@@ -687,8 +755,8 @@ mod tests {
#[tokio::test]
async fn test_send_receive_client_hello_message() {
use tokio::net::{TcpListener, TcpStream};
use nym_lp::message::ClientHelloData;
use tokio::net::{TcpListener, TcpStream};
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
@@ -716,7 +784,9 @@ mod tests {
let mut client_stream = TcpStream::connect(addr).await.unwrap();
server_task.await.unwrap().unwrap();
let received = read_lp_packet_from_stream(&mut client_stream).await.unwrap();
let received = read_lp_packet_from_stream(&mut client_stream)
.await
.unwrap();
assert_eq!(received.header().session_id, 300);
assert_eq!(received.header().counter, 30);
match received.message() {
@@ -761,7 +831,9 @@ mod tests {
},
LpMessage::ClientHello(hello_data.clone()),
);
write_lp_packet_to_stream(&mut client_stream, &packet).await.unwrap();
write_lp_packet_to_stream(&mut client_stream, &packet)
.await
.unwrap();
// Handler should receive and parse it
let result = server_task.await.unwrap();
@@ -774,8 +846,8 @@ mod tests {
#[tokio::test]
async fn test_receive_client_hello_timestamp_too_old() {
use tokio::net::{TcpListener, TcpStream};
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::net::{TcpListener, TcpStream};
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
@@ -791,16 +863,15 @@ mod tests {
// Create ClientHello with old timestamp
let client_keypair = Keypair::default();
let mut hello_data = ClientHelloData::new_with_fresh_salt(
client_keypair.public_key().to_bytes(),
1,
);
let mut hello_data =
ClientHelloData::new_with_fresh_salt(client_keypair.public_key().to_bytes(), 1);
// Manually set timestamp to be very old (100 seconds ago)
let old_timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs() - 100;
.as_secs()
- 100;
hello_data.salt[..8].copy_from_slice(&old_timestamp.to_le_bytes());
let packet = LpPacket::new(
@@ -811,7 +882,9 @@ mod tests {
},
LpMessage::ClientHello(hello_data),
);
write_lp_packet_to_stream(&mut client_stream, &packet).await.unwrap();
write_lp_packet_to_stream(&mut client_stream, &packet)
.await
.unwrap();
// Should fail with timestamp error
let result = server_task.await.unwrap();
@@ -821,7 +894,11 @@ mod tests {
match result {
Err(e) => {
let err_msg = format!("{}", e);
assert!(err_msg.contains("too old"), "Expected 'too old' in error, got: {}", err_msg);
assert!(
err_msg.contains("too old"),
"Expected 'too old' in error, got: {}",
err_msg
);
}
Ok(_) => panic!("Expected error but got success"),
}
gateway/src/node/lp_listener/handshake.rs
+50 -39
View File
@@ -28,16 +28,16 @@ impl LpGatewayHandshake {
local_keypair,
remote_public_key,
psk,
).map_err(|e| GatewayError::LpHandshakeError(format!("Failed to create state machine: {}", e)))?;
)
.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> {
pub async fn complete(mut self, stream: &mut TcpStream) -> Result<LpSession, GatewayError> {
debug!("Starting LP handshake as responder");
// Start the handshake
@@ -49,13 +49,14 @@ impl LpGatewayHandshake {
Ok(_) => {
// Unexpected action at this stage
return Err(GatewayError::LpHandshakeError(
"Unexpected action at handshake start".to_string()
"Unexpected action at handshake start".to_string(),
));
}
Err(e) => {
return Err(GatewayError::LpHandshakeError(
format!("Failed to start handshake: {}", e)
));
return Err(GatewayError::LpHandshakeError(format!(
"Failed to start handshake: {}",
e
)));
}
}
}
@@ -66,7 +67,10 @@ impl LpGatewayHandshake {
let packet = self.receive_packet(stream).await?;
// Process the received packet
if let Some(action) = self.state_machine.process_input(LpInput::ReceivePacket(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?;
@@ -79,19 +83,19 @@ impl LpGatewayHandshake {
debug!("Received action during handshake: {:?}", other);
}
Err(e) => {
return Err(GatewayError::LpHandshakeError(
format!("Handshake error: {}", 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)
))
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
@@ -100,56 +104,63 @@ impl LpGatewayHandshake {
stream: &mut TcpStream,
packet: &LpPacket,
) -> Result<(), GatewayError> {
use nym_lp::codec::serialize_lp_packet;
use bytes::BytesMut;
use nym_lp::codec::serialize_lp_packet;
// Serialize the packet first
let mut packet_buf = BytesMut::new();
serialize_lp_packet(packet, &mut packet_buf)
.map_err(|e| GatewayError::LpProtocolError(format!("Failed to serialize packet: {}", e)))?;
serialize_lp_packet(packet, &mut packet_buf).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)))?;
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.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)))?;
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());
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> {
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)))?;
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)
));
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)))?;
stream.read_exact(&mut packet_buf).await.map_err(|e| {
GatewayError::LpConnectionError(format!("Failed to read packet data: {}", e))
})?;
let packet = parse_lp_packet(&packet_buf)
.map_err(|e| GatewayError::LpProtocolError(format!("Failed to parse packet: {}", e)))?;
@@ -157,4 +168,4 @@ impl LpGatewayHandshake {
debug!("Received LP packet ({} bytes + 4 byte header)", packet_len);
Ok(packet)
}
}
}
gateway/src/node/lp_listener/messages.rs
+1 -1
View File
@@ -7,4 +7,4 @@
pub use nym_registration_common::{
LpRegistrationRequest, LpRegistrationResponse, RegistrationMode,
};
};
gateway/src/node/lp_listener/mod.rs
+76 -21
View File
@@ -1,6 +1,55 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
// LP (Lewes Protocol) Metrics Documentation
//
// This module implements comprehensive metrics collection for LP operations using nym-metrics macros.
// All metrics are automatically prefixed with the package name (nym_gateway) when registered.
//
// ## 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)
// - 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
// - lp_handshakes_failed: Counter for failed handshakes
// - lp_handshake_duration_seconds: Histogram of handshake durations (buckets: 10ms to 10s)
// - lp_timestamp_validation_accepted: Counter for timestamp validations that passed
// - lp_timestamp_validation_rejected: Counter for timestamp validations that failed
// - lp_errors_handshake: Counter for handshake errors
// - lp_errors_send_response: Counter for errors sending registration responses
// - lp_errors_timestamp_too_old: Counter for ClientHello timestamps that are too old
// - lp_errors_timestamp_too_far_future: Counter for ClientHello timestamps that are too far in the future
//
// ## Registration Metrics (in registration.rs)
// - lp_registration_attempts_total: Counter for all registration attempts
// - lp_registration_success_total: Counter for successful registrations (any mode)
// - lp_registration_failed_total: Counter for failed registrations (any mode)
// - lp_registration_failed_timestamp: Counter for registrations rejected due to invalid timestamp
// - lp_registration_duration_seconds: Histogram of registration durations (buckets: 100ms to 30s)
//
// ## Mode-Specific Registration Metrics (in registration.rs)
// - lp_registration_dvpn_attempts: Counter for dVPN mode registration attempts
// - lp_registration_dvpn_success: Counter for successful dVPN registrations
// - lp_registration_dvpn_failed: Counter for failed dVPN registrations
// - lp_registration_mixnet_attempts: Counter for Mixnet mode registration attempts
// - lp_registration_mixnet_success: Counter for successful Mixnet registrations
// - lp_registration_mixnet_failed: Counter for failed Mixnet registrations
//
// ## Credential Verification Metrics (in registration.rs)
// - lp_credential_verification_attempts: Counter for credential verification attempts
// - lp_credential_verification_success: Counter for successful credential verifications
// - lp_credential_verification_failed: Counter for failed credential verifications
// - lp_bandwidth_allocated_bytes_total: Counter for total bandwidth allocated (in bytes)
//
// ## Error Categorization Metrics
// - lp_errors_wg_peer_registration: Counter for WireGuard peer registration failures
//
// ## Usage Example
// To view metrics, the nym-metrics registry automatically collects all metrics.
// They can be exported via Prometheus format using the metrics endpoint.
use crate::error::GatewayError;
use crate::node::ActiveClientsStore;
use nym_crypto::asymmetric::ed25519;
@@ -91,7 +140,8 @@ fn default_timestamp_tolerance_secs() -> u64 {
#[derive(Clone)]
pub struct LpHandlerState {
/// Ecash verifier for bandwidth credentials
pub ecash_verifier: Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
pub ecash_verifier:
Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
/// Storage backend for persistence
pub storage: GatewayStorage,
@@ -151,18 +201,21 @@ impl LpListener {
}
pub async fn run(&mut self) -> Result<(), GatewayError> {
let listener = TcpListener::bind(self.control_address)
.await
.map_err(|e| {
error!("Failed to bind LP listener to {}: {}", self.control_address, e);
GatewayError::ListenerBindFailure {
address: self.control_address.to_string(),
source: Box::new(e),
}
})?;
let listener = TcpListener::bind(self.control_address).await.map_err(|e| {
error!(
"Failed to bind LP listener to {}: {}",
self.control_address, e
);
GatewayError::ListenerBindFailure {
address: self.control_address.to_string(),
source: Box::new(e),
}
})?;
info!("LP listener started on {} (data port reserved: {})",
self.control_address, self.data_port);
info!(
"LP listener started on {} (data port reserved: {})",
self.control_address, self.data_port
);
let shutdown_token = self.shutdown.clone_shutdown_token();
@@ -203,18 +256,17 @@ impl LpListener {
return;
}
debug!("Accepting LP connection from {} ({} active connections)",
remote_addr, active_connections);
debug!(
"Accepting LP connection from {} ({} active connections)",
remote_addr, active_connections
);
// Increment connection counter
self.handler_state.metrics.network.new_lp_connection();
// Spawn handler task
let handler = handler::LpConnectionHandler::new(
stream,
remote_addr,
self.handler_state.clone(),
);
let handler =
handler::LpConnectionHandler::new(stream, remote_addr, self.handler_state.clone());
let metrics = self.handler_state.metrics.clone();
self.shutdown.try_spawn_named(
@@ -230,6 +282,9 @@ impl LpListener {
}
fn active_lp_connections(&self) -> usize {
self.handler_state.metrics.network.active_lp_connections_count()
self.handler_state
.metrics
.network
.active_lp_connections_count()
}
}
}
gateway/src/node/lp_listener/registration.rs
+118 -32
View File
@@ -16,6 +16,7 @@ use nym_credentials_interface::CredentialSpendingData;
use nym_gateway_requests::models::CredentialSpendingRequest;
use nym_gateway_storage::models::PersistedBandwidth;
use nym_gateway_storage::traits::BandwidthGatewayStorage;
use nym_metrics::{add_histogram_obs, inc, inc_by};
use nym_registration_common::GatewayData;
use nym_wireguard::PeerControlRequest;
use rand::RngCore;
@@ -24,6 +25,20 @@ use std::str::FromStr;
use std::sync::Arc;
use tracing::*;
// Histogram buckets for LP registration duration tracking
// Registration includes credential verification, DB operations, and potentially WireGuard peer setup
// Expected durations: 100ms - 5s for normal operations, up to 30s for slow DB or network issues
const LP_REGISTRATION_DURATION_BUCKETS: &[f64] = &[
0.1, // 100ms
0.25, // 250ms
0.5, // 500ms
1.0, // 1s
2.5, // 2.5s
5.0, // 5s
10.0, // 10s
30.0, // 30s
];
/// Prepare bandwidth storage for a client
async fn credential_storage_preparation(
ecash_verifier: Arc<dyn EcashManager + Send + Sync>,
@@ -38,9 +53,7 @@ async fn credential_storage_preparation(
.get_available_bandwidth(client_id)
.await?
.ok_or_else(|| {
GatewayError::InternalError(
"bandwidth entry should have just been created".to_string(),
)
GatewayError::InternalError("bandwidth entry should have just been created".to_string())
})?;
Ok(bandwidth)
}
@@ -64,7 +77,22 @@ async fn credential_verification(
true,
),
);
Ok(verifier.verify().await?)
// Track credential verification attempts
inc!("lp_credential_verification_attempts");
match verifier.verify().await {
Ok(allocated) => {
inc!("lp_credential_verification_success");
// Track allocated bandwidth
inc_by!("lp_bandwidth_allocated_bytes_total", allocated);
Ok(allocated)
}
Err(e) => {
inc!("lp_credential_verification_failed");
Err(e.into())
}
}
}
/// Process an LP registration request
@@ -73,29 +101,37 @@ pub async fn process_registration(
state: &LpHandlerState,
) -> LpRegistrationResponse {
let session_id = rand::random::<u32>();
let registration_start = std::time::Instant::now();
// Track total registration attempts
inc!("lp_registration_attempts_total");
// 1. Validate timestamp for replay protection
if !request.validate_timestamp(30) {
warn!("LP registration failed: timestamp too old or too far in future");
return LpRegistrationResponse::error(
session_id,
"Invalid timestamp".to_string(),
);
inc!("lp_registration_failed_timestamp");
return LpRegistrationResponse::error(session_id, "Invalid timestamp".to_string());
}
// 2. Process based on mode
match request.mode {
let result = match request.mode {
RegistrationMode::Dvpn => {
// Track dVPN registration attempts
inc!("lp_registration_dvpn_attempts");
// Register as WireGuard peer first to get client_id
let (gateway_data, client_id) = match register_wg_peer(
request.wg_public_key.inner().as_ref(),
request.client_ip,
request.ticket_type,
state,
).await {
)
.await
{
Ok(result) => result,
Err(e) => {
error!("LP WireGuard peer registration failed: {}", e);
inc!("lp_registration_dvpn_failed");
inc!("lp_errors_wg_peer_registration");
return LpRegistrationResponse::error(
session_id,
format!("WireGuard peer registration failed: {}", e),
@@ -108,16 +144,26 @@ pub async fn process_registration(
state.ecash_verifier.clone(),
request.credential,
client_id,
).await {
)
.await
{
Ok(bandwidth) => bandwidth,
Err(e) => {
// Credential verification failed, remove the peer
warn!("LP credential verification failed for client {}: {}", client_id, e);
if let Err(remove_err) = state.storage
warn!(
"LP credential verification failed for client {}: {}",
client_id, e
);
inc!("lp_registration_dvpn_failed");
if let Err(remove_err) = state
.storage
.remove_wireguard_peer(&request.wg_public_key.to_string())
.await
{
error!("Failed to remove peer after credential verification failure: {}", remove_err);
error!(
"Failed to remove peer after credential verification failure: {}",
remove_err
);
}
return LpRegistrationResponse::error(
session_id,
@@ -126,28 +172,42 @@ pub async fn process_registration(
}
};
info!("LP dVPN registration successful for session {} (client_id: {})", session_id, client_id);
LpRegistrationResponse::success(
session_id,
allocated_bandwidth,
gateway_data,
)
info!(
"LP dVPN registration successful for session {} (client_id: {})",
session_id, client_id
);
inc!("lp_registration_dvpn_success");
LpRegistrationResponse::success(session_id, allocated_bandwidth, gateway_data)
}
RegistrationMode::Mixnet { client_id: client_id_bytes } => {
RegistrationMode::Mixnet {
client_id: client_id_bytes,
} => {
// Track mixnet registration attempts
inc!("lp_registration_mixnet_attempts");
// Generate i64 client_id from the [u8; 32] in the request
let client_id = i64::from_be_bytes(client_id_bytes[0..8].try_into().unwrap());
info!("LP Mixnet registration for client_id {}, session {}", client_id, session_id);
info!(
"LP Mixnet registration for client_id {}, session {}",
client_id, session_id
);
// Verify credential with CredentialVerifier
let allocated_bandwidth = match credential_verification(
state.ecash_verifier.clone(),
request.credential,
client_id,
).await {
)
.await
{
Ok(bandwidth) => bandwidth,
Err(e) => {
warn!("LP Mixnet credential verification failed for client {}: {}", client_id, e);
warn!(
"LP Mixnet credential verification failed for client {}: {}",
client_id, e
);
inc!("lp_registration_mixnet_failed");
return LpRegistrationResponse::error(
session_id,
format!("Credential verification failed: {}", e),
@@ -157,7 +217,11 @@ pub async fn process_registration(
// For mixnet mode, we don't have WireGuard data
// In the future, this would set up mixnet-specific state
info!("LP Mixnet registration successful for session {} (client_id: {})", session_id, client_id);
info!(
"LP Mixnet registration successful for session {} (client_id: {})",
session_id, client_id
);
inc!("lp_registration_mixnet_success");
LpRegistrationResponse {
success: true,
error: None,
@@ -166,7 +230,24 @@ pub async fn process_registration(
session_id,
}
}
};
// Track registration duration
let duration = registration_start.elapsed().as_secs_f64();
add_histogram_obs!(
"lp_registration_duration_seconds",
duration,
LP_REGISTRATION_DURATION_BUCKETS
);
// Track overall success/failure
if result.success {
inc!("lp_registration_success_total");
} else {
inc!("lp_registration_failed_total");
}
result
}
/// Register a WireGuard peer and return gateway data along with the client_id
@@ -192,7 +273,7 @@ async fn register_wg_peer(
let mut key_bytes = [0u8; 32];
if public_key_bytes.len() != 32 {
return Err(GatewayError::LpProtocolError(
"Invalid WireGuard public key length".to_string()
"Invalid WireGuard public key length".to_string(),
));
}
key_bytes.copy_from_slice(public_key_bytes);
@@ -211,9 +292,11 @@ async fn register_wg_peer(
// Create WireGuard peer
let mut peer = Peer::new(peer_key.clone());
peer.preshared_key = Some(Key::new(state.local_identity.public_key().to_bytes()));
peer.endpoint = Some(format!("{}:51820", client_ip).parse().unwrap_or_else(|_| {
SocketAddr::from_str("0.0.0.0:51820").unwrap()
}));
peer.endpoint = Some(
format!("{}:51820", client_ip)
.parse()
.unwrap_or_else(|_| SocketAddr::from_str("0.0.0.0:51820").unwrap()),
);
peer.allowed_ips = vec![
format!("{}/32", client_ipv4).parse().unwrap(),
format!("{}/128", client_ipv6).parse().unwrap(),
@@ -231,11 +314,14 @@ async fn register_wg_peer(
.map_err(|e| GatewayError::InternalError(format!("Failed to send peer request: {}", e)))?;
rx.await
.map_err(|e| GatewayError::InternalError(format!("Failed to receive peer response: {}", e)))?
.map_err(|e| {
GatewayError::InternalError(format!("Failed to receive peer response: {}", e))
})?
.map_err(|e| GatewayError::InternalError(format!("Failed to add peer: {:?}", e)))?;
// Store bandwidth allocation and get client_id
let client_id = state.storage
let client_id = state
.storage
.insert_wireguard_peer(&peer, ticket_type.into())
.await
.map_err(|e| {
@@ -259,4 +345,4 @@ async fn register_wg_peer(
},
client_id,
))
}
}
gateway/src/node/mod.rs
+18 -8
View File
@@ -249,13 +249,22 @@ impl GatewayTasksBuilder {
Ok(Arc::new(ecash_manager))
}
async fn ecash_manager(&mut self) -> Result<Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>, GatewayError> {
async fn ecash_manager(
&mut self,
) -> Result<
Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>,
GatewayError,
> {
match self.ecash_manager.clone() {
Some(cached) => Ok(cached as Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>),
Some(cached) => Ok(cached
as Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>),
None => {
let manager = self.build_ecash_manager().await?;
self.ecash_manager = Some(manager.clone());
Ok(manager as Arc<dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync>)
Ok(manager
as Arc<
dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync,
>)
}
}
}
@@ -312,11 +321,12 @@ impl GatewayTasksBuilder {
};
// Parse bind address from config
let bind_addr = format!("{}:{}",
self.config.lp.bind_address,
self.config.lp.control_port
).parse()
.map_err(|e| GatewayError::InternalError(format!("Invalid LP bind address: {}", e)))?;
let bind_addr = format!(
"{}:{}",
self.config.lp.bind_address, self.config.lp.control_port
)
.parse()
.map_err(|e| GatewayError::InternalError(format!("Invalid LP bind address: {}", e)))?;
Ok(lp_listener::LpListener::new(
bind_addr,
nym-node/nym-node-metrics/src/network.rs
+3 -6
View File
@@ -60,17 +60,14 @@ impl NetworkStats {
}
pub fn new_lp_connection(&self) {
self.active_lp_connections
.fetch_add(1, Ordering::Relaxed);
self.active_lp_connections.fetch_add(1, Ordering::Relaxed);
}
pub fn lp_connection_closed(&self) {
self.active_lp_connections
.fetch_sub(1, Ordering::Relaxed);
self.active_lp_connections.fetch_sub(1, Ordering::Relaxed);
}
pub fn active_lp_connections_count(&self) -> usize {
self.active_lp_connections
.load(Ordering::Relaxed)
self.active_lp_connections.load(Ordering::Relaxed)
}
}
nym-registration-client/src/builder/config.rs
+4 -1
View File
@@ -366,7 +366,10 @@ impl BuilderConfigBuilder {
/// Legacy method for backward compatibility
/// Use `wireguard_mode()` or `mixnet_mode()` instead
#[deprecated(since = "0.1.0", note = "Use `mode()`, `wireguard_mode()`, or `mixnet_mode()` instead")]
#[deprecated(
since = "0.1.0",
note = "Use `mode()`, `wireguard_mode()`, or `mixnet_mode()` instead"
)]
pub fn two_hops(self, two_hops: bool) -> Self {
if two_hops {
self.wireguard_mode()
nym-registration-client/src/lib.rs
+15 -16
View File
@@ -30,8 +30,7 @@ pub use config::RegistrationMode;
pub use error::RegistrationClientError;
pub use lp_client::LpConfig;
pub use types::{
LpRegistrationResult, MixnetRegistrationResult, RegistrationResult,
WireguardRegistrationResult,
LpRegistrationResult, MixnetRegistrationResult, RegistrationResult, WireguardRegistrationResult,
};
pub struct RegistrationClient {
@@ -175,24 +174,26 @@ impl RegistrationClient {
// For now, use gateway identities as LP public keys
// TODO(nym-87): Implement proper key derivation
let entry_gateway_lp_key =
LpPublicKey::from_bytes(&self.config.entry.node.identity.to_bytes())
.map_err(|e| RegistrationClientError::LpRegistrationNotPossible {
LpPublicKey::from_bytes(&self.config.entry.node.identity.to_bytes()).map_err(|e| {
RegistrationClientError::LpRegistrationNotPossible {
node_id: format!(
"{}: invalid LP key: {}",
self.config.entry.node.identity.to_base58_string(),
e
),
})?;
}
})?;
let exit_gateway_lp_key =
LpPublicKey::from_bytes(&self.config.exit.node.identity.to_bytes())
.map_err(|e| RegistrationClientError::LpRegistrationNotPossible {
LpPublicKey::from_bytes(&self.config.exit.node.identity.to_bytes()).map_err(|e| {
RegistrationClientError::LpRegistrationNotPossible {
node_id: format!(
"{}: invalid LP key: {}",
self.config.exit.node.identity.to_base58_string(),
e
),
})?;
}
})?;
// Generate LP keypairs for this connection
let client_lp_keypair = Arc::new(LpKeypair::default());
@@ -287,23 +288,21 @@ impl RegistrationClient {
// Handle entry gateway result
// Note: entry_transport is dropped here, closing the LP connection
let (_entry_transport, entry_gateway_data) = entry_result.map_err(|source| {
RegistrationClientError::EntryGatewayRegisterLp {
let (_entry_transport, entry_gateway_data) =
entry_result.map_err(|source| RegistrationClientError::EntryGatewayRegisterLp {
gateway_id: self.config.entry.node.identity.to_base58_string(),
lp_address: entry_lp_address,
source: Box::new(source),
}
})?;
})?;
// Handle exit gateway result
// Note: exit_transport is dropped here, closing the LP connection
let (_exit_transport, exit_gateway_data) = exit_result.map_err(|source| {
RegistrationClientError::ExitGatewayRegisterLp {
let (_exit_transport, exit_gateway_data) =
exit_result.map_err(|source| RegistrationClientError::ExitGatewayRegisterLp {
gateway_id: self.config.exit.node.identity.to_base58_string(),
lp_address: exit_lp_address,
source: Box::new(source),
}
})?;
})?;
tracing::info!(
"LP registration successful for both gateways (LP connections will be closed)"
nym-registration-client/src/lp_client/client.rs
+36 -44
View File
@@ -10,10 +10,10 @@ use bytes::BytesMut;
use nym_bandwidth_controller::{BandwidthTicketProvider, DEFAULT_TICKETS_TO_SPEND};
use nym_credentials_interface::TicketType;
use nym_crypto::asymmetric::{ed25519, x25519};
use nym_lp::LpPacket;
use nym_lp::codec::{parse_lp_packet, serialize_lp_packet};
use nym_lp::keypair::{Keypair, PublicKey};
use nym_lp::state_machine::{LpAction, LpInput, LpStateMachine};
use nym_lp::LpPacket;
use nym_registration_common::{GatewayData, LpRegistrationRequest, LpRegistrationResponse};
use nym_wireguard_types::PeerPublicKey;
use std::net::{IpAddr, SocketAddr};
@@ -141,10 +141,7 @@ impl LpRegistrationClient {
address: self.gateway_lp_address.to_string(),
source: std::io::Error::new(
std::io::ErrorKind::TimedOut,
format!(
"Connection timeout after {:?}",
self.config.connect_timeout
),
format!("Connection timeout after {:?}", self.config.connect_timeout),
),
})?
.map_err(|source| LpClientError::TcpConnection {
@@ -214,14 +211,17 @@ impl LpRegistrationClient {
/// Timeout applied in nym-102.
pub async fn perform_handshake(&mut self) -> Result<()> {
// Apply handshake timeout (nym-102)
tokio::time::timeout(self.config.handshake_timeout, self.perform_handshake_inner())
.await
.map_err(|_| {
LpClientError::Transport(format!(
"Handshake timeout after {:?}",
self.config.handshake_timeout
))
})?
tokio::time::timeout(
self.config.handshake_timeout,
self.perform_handshake_inner(),
)
.await
.map_err(|_| {
LpClientError::Transport(format!(
"Handshake timeout after {:?}",
self.config.handshake_timeout
))
})?
}
/// Internal handshake implementation without timeout.
@@ -239,7 +239,10 @@ impl LpRegistrationClient {
);
let salt = client_hello_data.salt;
tracing::trace!("Generated ClientHello with timestamp: {}", client_hello_data.extract_timestamp());
tracing::trace!(
"Generated ClientHello with timestamp: {}",
client_hello_data.extract_timestamp()
);
// Step 2: Send ClientHello as first packet (before Noise handshake)
let client_hello_header = nym_lp::packet::LpHeader::new(
@@ -328,10 +331,9 @@ impl LpRegistrationClient {
// 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| LpClientError::Transport(format!("Failed to send packet length: {}", e)))?;
stream.write_all(&len.to_be_bytes()).await.map_err(|e| {
LpClientError::Transport(format!("Failed to send packet length: {}", e))
})?;
// Send the actual packet data
stream
@@ -345,7 +347,10 @@ impl LpRegistrationClient {
.await
.map_err(|e| LpClientError::Transport(format!("Failed to flush stream: {}", e)))?;
tracing::trace!("Sent LP packet ({} bytes + 4 byte header)", packet_buf.len());
tracing::trace!(
"Sent LP packet ({} bytes + 4 byte header)",
packet_buf.len()
);
Ok(())
}
@@ -361,10 +366,9 @@ impl LpRegistrationClient {
async fn receive_packet(stream: &mut TcpStream) -> Result<LpPacket> {
// Read 4-byte length prefix (u32 big-endian)
let mut len_buf = [0u8; 4];
stream
.read_exact(&mut len_buf)
.await
.map_err(|e| LpClientError::Transport(format!("Failed to read packet length: {}", e)))?;
stream.read_exact(&mut len_buf).await.map_err(|e| {
LpClientError::Transport(format!("Failed to read packet length: {}", e))
})?;
let packet_len = u32::from_be_bytes(len_buf) as usize;
@@ -388,10 +392,7 @@ impl LpRegistrationClient {
let packet = parse_lp_packet(&packet_buf)
.map_err(|e| LpClientError::Transport(format!("Failed to parse packet: {}", e)))?;
tracing::trace!(
"Received LP packet ({} bytes + 4 byte header)",
packet_len
);
tracing::trace!("Received LP packet ({} bytes + 4 byte header)", packet_len);
Ok(packet)
}
@@ -457,12 +458,8 @@ impl LpRegistrationClient {
// 2. 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,
self.client_ip,
);
let request =
LpRegistrationRequest::new_dvpn(wg_public_key, credential, ticket_type, self.client_ip);
tracing::trace!("Built registration request: {:?}", request);
@@ -592,19 +589,18 @@ impl LpRegistrationClient {
return Err(LpClientError::Transport(format!(
"Unexpected action when receiving registration response: {:?}",
other
)))
)));
}
};
// 4. Deserialize the response
let response: LpRegistrationResponse = bincode::deserialize(&response_data).map_err(
|e| {
let response: LpRegistrationResponse =
bincode::deserialize(&response_data).map_err(|e| {
LpClientError::ReceiveRegistrationResponse(format!(
"Failed to deserialize registration response: {}",
e
))
},
)?;
})?;
tracing::debug!(
"Received registration response: success={}, session_id={}",
@@ -618,9 +614,7 @@ impl LpRegistrationClient {
.error
.unwrap_or_else(|| "Unknown error".to_string());
tracing::warn!("Gateway rejected registration: {}", error_msg);
return Err(LpClientError::RegistrationRejected {
reason: error_msg,
});
return Err(LpClientError::RegistrationRejected { reason: error_msg });
}
// Extract gateway_data
@@ -676,9 +670,7 @@ impl LpRegistrationClient {
// Ensure handshake completed
let state_machine = self.state_machine.ok_or_else(|| {
LpClientError::Transport(
"Cannot create transport: handshake not completed".to_string(),
)
LpClientError::Transport("Cannot create transport: handshake not completed".to_string())
})?;
// Create and return transport (validates state is Transport)
nym-registration-client/src/lp_client/transport.rs
+16 -16
View File
@@ -8,9 +8,9 @@
use super::error::{LpClientError, Result};
use bytes::BytesMut;
use nym_lp::LpPacket;
use nym_lp::codec::{parse_lp_packet, serialize_lp_packet};
use nym_lp::state_machine::{LpAction, LpInput, LpStateBare, LpStateMachine};
use nym_lp::LpPacket;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
@@ -88,7 +88,9 @@ impl LpTransport {
.state_machine
.process_input(LpInput::SendData(data.to_vec()))
.ok_or_else(|| {
LpClientError::Transport("State machine returned no action for SendData".to_string())
LpClientError::Transport(
"State machine returned no action for SendData".to_string(),
)
})?
.map_err(|e| LpClientError::Transport(format!("Failed to encrypt data: {}", e)))?;
@@ -166,10 +168,7 @@ impl LpTransport {
tracing::debug!("LP connection closed by state machine");
}
Ok(other) => {
tracing::warn!(
"Unexpected action when closing connection: {:?}",
other
);
tracing::warn!("Unexpected action when closing connection: {:?}", other);
}
Err(e) => {
tracing::warn!("Error closing LP connection: {}", e);
@@ -207,7 +206,9 @@ impl LpTransport {
self.stream
.write_all(&len.to_be_bytes())
.await
.map_err(|e| LpClientError::Transport(format!("Failed to send packet length: {}", e)))?;
.map_err(|e| {
LpClientError::Transport(format!("Failed to send packet length: {}", e))
})?;
// Send the actual packet data
self.stream
@@ -221,7 +222,10 @@ impl LpTransport {
.await
.map_err(|e| LpClientError::Transport(format!("Failed to flush stream: {}", e)))?;
tracing::trace!("Sent LP packet ({} bytes + 4 byte header)", packet_buf.len());
tracing::trace!(
"Sent LP packet ({} bytes + 4 byte header)",
packet_buf.len()
);
Ok(())
}
@@ -231,10 +235,9 @@ impl LpTransport {
async fn receive_packet(&mut self) -> Result<LpPacket> {
// Read 4-byte length prefix (u32 big-endian)
let mut len_buf = [0u8; 4];
self.stream
.read_exact(&mut len_buf)
.await
.map_err(|e| LpClientError::Transport(format!("Failed to read packet length: {}", e)))?;
self.stream.read_exact(&mut len_buf).await.map_err(|e| {
LpClientError::Transport(format!("Failed to read packet length: {}", e))
})?;
let packet_len = u32::from_be_bytes(len_buf) as usize;
@@ -258,10 +261,7 @@ impl LpTransport {
let packet = parse_lp_packet(&packet_buf)
.map_err(|e| LpClientError::Transport(format!("Failed to parse packet: {}", e)))?;
tracing::trace!(
"Received LP packet ({} bytes + 4 byte header)",
packet_len
);
tracing::trace!("Received LP packet ({} bytes + 4 byte header)", packet_len);
Ok(packet)
}
}