From de06f4a5c0cd680f39994d650b2113de0041e1a2 Mon Sep 17 00:00:00 2001 From: durch Date: Thu, 23 Oct 2025 20:14:57 +0200 Subject: [PATCH] fmt and metrics --- Cargo.lock | 1 + .../mixnet/operators/nymnode/bond_nymnode.rs | 4 +- .../nymnode/nymnode_bonding_sign_payload.rs | 4 +- .../nymnode/settings/update_config.rs | 5 +- common/crypto/src/asymmetric/ed25519/mod.rs | 19 +- common/crypto/src/kdf.rs | 10 +- common/nym-kcp/src/session.rs | 2 +- common/nym-lp/src/message.rs | 9 +- common/nym-lp/src/psk.rs | 24 +- common/nym-lp/src/session.rs | 7 +- common/registration/src/lp_messages.rs | 28 +- gateway/Cargo.toml | 4 + .../client_handling/websocket/common_state.rs | 3 +- .../authenticator/mod.rs | 4 +- gateway/src/node/lp_listener/handler.rs | 291 +++++++++++------- gateway/src/node/lp_listener/handshake.rs | 89 +++--- gateway/src/node/lp_listener/messages.rs | 2 +- gateway/src/node/lp_listener/mod.rs | 97 ++++-- gateway/src/node/lp_listener/registration.rs | 150 +++++++-- gateway/src/node/mod.rs | 26 +- nym-node/nym-node-metrics/src/network.rs | 9 +- nym-registration-client/src/builder/config.rs | 5 +- nym-registration-client/src/lib.rs | 31 +- .../src/lp_client/client.rs | 80 +++-- .../src/lp_client/transport.rs | 32 +- 25 files changed, 584 insertions(+), 352 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b9a7c806f8..7c0fa44f5a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5840,6 +5840,7 @@ dependencies = [ "nym-ip-packet-router", "nym-kcp", "nym-lp", + "nym-metrics", "nym-mixnet-client", "nym-network-defaults", "nym-network-requester", diff --git a/common/commands/src/validator/mixnet/operators/nymnode/bond_nymnode.rs b/common/commands/src/validator/mixnet/operators/nymnode/bond_nymnode.rs index 032528e91f..46c54417d3 100644 --- a/common/commands/src/validator/mixnet/operators/nymnode/bond_nymnode.rs +++ b/common/commands/src/validator/mixnet/operators/nymnode/bond_nymnode.rs @@ -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, diff --git a/common/commands/src/validator/mixnet/operators/nymnode/nymnode_bonding_sign_payload.rs b/common/commands/src/validator/mixnet/operators/nymnode/nymnode_bonding_sign_payload.rs index a7e7fae1b1..234c87b6bc 100644 --- a/common/commands/src/validator/mixnet/operators/nymnode/nymnode_bonding_sign_payload.rs +++ b/common/commands/src/validator/mixnet/operators/nymnode/nymnode_bonding_sign_payload.rs @@ -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, diff --git a/common/commands/src/validator/mixnet/operators/nymnode/settings/update_config.rs b/common/commands/src/validator/mixnet/operators/nymnode/settings/update_config.rs index 7aec6ebb0b..c2f863e35a 100644 --- a/common/commands/src/validator/mixnet/operators/nymnode/settings/update_config.rs +++ b/common/commands/src/validator/mixnet/operators/nymnode/settings/update_config.rs @@ -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, // equivalent to setting `lp_address` to `None` diff --git a/common/crypto/src/asymmetric/ed25519/mod.rs b/common/crypto/src/asymmetric/ed25519/mod.rs index 1191ac1b03..d072b4f598 100644 --- a/common/crypto/src/asymmetric/ed25519/mod.rs +++ b/common/crypto/src/asymmetric/ed25519/mod.rs @@ -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); diff --git a/common/crypto/src/kdf.rs b/common/crypto/src/kdf.rs index 6784def06f..3edb257299 100644 --- a/common/crypto/src/kdf.rs +++ b/common/crypto/src/kdf.rs @@ -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] diff --git a/common/nym-kcp/src/session.rs b/common/nym-kcp/src/session.rs index 7720d39300..2e56c9073f 100644 --- a/common/nym-kcp/src/session.rs +++ b/common/nym-kcp/src/session.rs @@ -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. diff --git a/common/nym-lp/src/message.rs b/common/nym-lp/src/message.rs index 6910cac1ea..27a83dc879 100644 --- a/common/nym-lp/src/message.rs +++ b/common/nym-lp/src/message.rs @@ -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); } } diff --git a/common/nym-lp/src/psk.rs b/common/nym-lp/src/psk.rs index 5fe3440c6c..676eb0e80a 100644 --- a/common/nym-lp/src/psk.rs +++ b/common/nym-lp/src/psk.rs @@ -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, diff --git a/common/nym-lp/src/session.rs b/common/nym-lp/src/session.rs index 2ba786022b..db512dad47 100644 --- a/common/nym-lp/src/session.rs +++ b/common/nym-lp/src/session.rs @@ -68,7 +68,7 @@ impl LpSession { remote_public_key: &[u8], psk: &[u8], ) -> Result { - // 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" diff --git a/common/registration/src/lp_messages.rs b/common/registration/src/lp_messages.rs index fa2d69abf0..8103e38314 100644 --- a/common/registration/src/lp_messages.rs +++ b/common/registration/src/lp_messages.rs @@ -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 = 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 ==================== diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index 5537e38399..c941b5c9da 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -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" } diff --git a/gateway/src/node/client_handling/websocket/common_state.rs b/gateway/src/node/client_handling/websocket/common_state.rs index 0543ffb12f..f129dfbe88 100644 --- a/gateway/src/node/client_handling/websocket/common_state.rs +++ b/gateway/src/node/client_handling/websocket/common_state.rs @@ -23,7 +23,8 @@ pub(crate) struct Config { #[derive(Clone)] pub(crate) struct CommonHandlerState { pub(crate) cfg: Config, - pub(crate) ecash_verifier: Arc, + pub(crate) ecash_verifier: + Arc, pub(crate) storage: GatewayStorage, pub(crate) local_identity: Arc, pub(crate) metrics: NymNodeMetrics, diff --git a/gateway/src/node/internal_service_providers/authenticator/mod.rs b/gateway/src/node/internal_service_providers/authenticator/mod.rs index b78b05b781..a98c31868e 100644 --- a/gateway/src/node/internal_service_providers/authenticator/mod.rs +++ b/gateway/src/node/internal_service_providers/authenticator/mod.rs @@ -51,7 +51,9 @@ impl Authenticator { upgrade_mode_state: UpgradeModeDetails, wireguard_gateway_data: WireguardGatewayData, used_private_network_ips: Vec, - ecash_verifier: Arc, + ecash_verifier: Arc< + dyn nym_credential_verification::ecash::traits::EcashManager + Send + Sync, + >, shutdown: ShutdownTracker, ) -> Self { Self { diff --git a/gateway/src/node/lp_listener/handler.rs b/gateway/src/node/lp_listener/handler.rs index 73f80bd8b7..71f08f930a 100644 --- a/gateway/src/node/lp_listener/handler.rs +++ b/gateway/src/node/lp_listener/handler.rs @@ -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, + ecash_verifier: Arc::new(ecash_verifier) + as Arc, 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"), } diff --git a/gateway/src/node/lp_listener/handshake.rs b/gateway/src/node/lp_listener/handshake.rs index f63c5a1e4f..935b501303 100644 --- a/gateway/src/node/lp_listener/handshake.rs +++ b/gateway/src/node/lp_listener/handshake.rs @@ -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 { + pub async fn complete(mut self, stream: &mut TcpStream) -> Result { 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 { + async fn receive_packet(&self, stream: &mut TcpStream) -> Result { 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) } -} \ No newline at end of file +} diff --git a/gateway/src/node/lp_listener/messages.rs b/gateway/src/node/lp_listener/messages.rs index 45c11d60c4..51e18a2805 100644 --- a/gateway/src/node/lp_listener/messages.rs +++ b/gateway/src/node/lp_listener/messages.rs @@ -7,4 +7,4 @@ pub use nym_registration_common::{ LpRegistrationRequest, LpRegistrationResponse, RegistrationMode, -}; \ No newline at end of file +}; diff --git a/gateway/src/node/lp_listener/mod.rs b/gateway/src/node/lp_listener/mod.rs index e06fb5007d..fef19c06dd 100644 --- a/gateway/src/node/lp_listener/mod.rs +++ b/gateway/src/node/lp_listener/mod.rs @@ -1,6 +1,55 @@ // Copyright 2025 - Nym Technologies SA // 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, + pub ecash_verifier: + Arc, /// 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() } -} \ No newline at end of file +} diff --git a/gateway/src/node/lp_listener/registration.rs b/gateway/src/node/lp_listener/registration.rs index 493d8b1d09..378fe5a86a 100644 --- a/gateway/src/node/lp_listener/registration.rs +++ b/gateway/src/node/lp_listener/registration.rs @@ -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, @@ -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::(); + 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, )) -} \ No newline at end of file +} diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 22703124fc..0f6e415159 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -249,13 +249,22 @@ impl GatewayTasksBuilder { Ok(Arc::new(ecash_manager)) } - async fn ecash_manager(&mut self) -> Result, GatewayError> { + async fn ecash_manager( + &mut self, + ) -> Result< + Arc, + GatewayError, + > { match self.ecash_manager.clone() { - Some(cached) => Ok(cached as Arc), + Some(cached) => Ok(cached + as Arc), None => { let manager = self.build_ecash_manager().await?; self.ecash_manager = Some(manager.clone()); - Ok(manager as Arc) + 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, diff --git a/nym-node/nym-node-metrics/src/network.rs b/nym-node/nym-node-metrics/src/network.rs index 6373d0ebfc..33fed5474b 100644 --- a/nym-node/nym-node-metrics/src/network.rs +++ b/nym-node/nym-node-metrics/src/network.rs @@ -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) } } diff --git a/nym-registration-client/src/builder/config.rs b/nym-registration-client/src/builder/config.rs index 9d71389f0d..9edb48adca 100644 --- a/nym-registration-client/src/builder/config.rs +++ b/nym-registration-client/src/builder/config.rs @@ -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() diff --git a/nym-registration-client/src/lib.rs b/nym-registration-client/src/lib.rs index 2cb62bc0a9..0f4c5fec1a 100644 --- a/nym-registration-client/src/lib.rs +++ b/nym-registration-client/src/lib.rs @@ -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)" diff --git a/nym-registration-client/src/lp_client/client.rs b/nym-registration-client/src/lp_client/client.rs index 2d5dab97cb..d59f60afb5 100644 --- a/nym-registration-client/src/lp_client/client.rs +++ b/nym-registration-client/src/lp_client/client.rs @@ -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 { // 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) diff --git a/nym-registration-client/src/lp_client/transport.rs b/nym-registration-client/src/lp_client/transport.rs index 638970202d..51f67e2406 100644 --- a/nym-registration-client/src/lp_client/transport.rs +++ b/nym-registration-client/src/lp_client/transport.rs @@ -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 { // 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) } }