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

additional formatting

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2026-01-06 11:29:08 +00:00
parent 7e300654b8
commit cdf8e23af3
8 changed files with 42 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/src/node/lp_listener/data_handler.rs
+9 -16
View File
@@ -56,14 +56,14 @@ impl LpDataHandler {
shutdown: nym_task::ShutdownToken,
) -> Result<Self, GatewayError> {
let socket = UdpSocket::bind(bind_addr).await.map_err(|e| {
error!("Failed to bind LP data socket to {}: {}", bind_addr, e);
error!("Failed to bind LP data socket to {bind_addr}: {e}");
GatewayError::ListenerBindFailure {
address: bind_addr.to_string(),
source: Box::new(e),
}
})?;
info!("LP data handler listening on UDP {}", bind_addr);
info!("LP data handler listening on UDP {bind_addr}");
Ok(Self {
socket: Arc::new(socket),
@@ -129,14 +129,10 @@ impl LpDataHandler {
})?;
let receiver_idx = header.receiver_idx;
let counter = header.counter;
let len = packet.len();
trace!(
"LP data packet from {} (receiver_idx={}, counter={}, len={})",
src_addr,
receiver_idx,
header.counter,
packet.len()
);
trace!("LP data packet from {src_addr} (receiver_idx={receiver_idx}, counter={counter}, len={len})");
// Step 2: Look up session state machine by receiver_idx (mutable for state updates)
let mut state_entry = self
@@ -146,8 +142,7 @@ impl LpDataHandler {
.ok_or_else(|| {
inc!("lp_data_unknown_session");
GatewayError::LpProtocolError(format!(
"Unknown session for receiver_idx {}",
receiver_idx
"Unknown session for receiver_idx {receiver_idx}"
))
})?;
@@ -203,8 +198,7 @@ impl LpDataHandler {
// UDP is connectionless - we can't send responses back easily
// For subsession rekeying, the client should use TCP control plane
debug!(
"Ignoring SendPacket action on UDP (receiver_idx={}) - use TCP for rekeying",
receiver_idx
"Ignoring SendPacket action on UDP (receiver_idx={receiver_idx}) - use TCP for rekeying",
);
inc!("lp_data_ignored_send_actions");
Ok(())
@@ -234,7 +228,7 @@ impl LpDataHandler {
// Parse as MixPacket v2 format (packet_type || key_rotation || next_hop || packet)
let mix_packet = MixPacket::try_from_v2_bytes(sphinx_bytes).map_err(|e| {
inc!("lp_data_sphinx_parse_errors");
GatewayError::LpProtocolError(format!("Failed to parse MixPacket: {}", e))
GatewayError::LpProtocolError(format!("Failed to parse MixPacket: {e}"))
})?;
trace!(
@@ -248,8 +242,7 @@ impl LpDataHandler {
error!("Failed to forward Sphinx packet to mixnet: {}", e);
inc!("lp_data_forward_errors");
return Err(GatewayError::InternalError(format!(
"Mix packet forwarding failed: {}",
e
"Mix packet forwarding failed: {e}",
)));
}
gateway/src/node/lp_listener/handler.rs
+4 -8
View File
@@ -1005,8 +1005,7 @@ impl LpConnectionHandler {
// Return error to prevent silent behavior changes that could mask bugs
inc!("lp_forward_failed");
return Err(GatewayError::LpProtocolError(format!(
"Forward target mismatch: session bound to {}, got request for {}",
existing_addr, target_addr
"Forward target mismatch: session bound to {existing_addr}, got request for {target_addr}"
)));
}
None => true,
@@ -1032,8 +1031,7 @@ impl LpConnectionHandler {
Ok(Err(e)) => {
inc!("lp_forward_failed");
return Err(GatewayError::LpConnectionError(format!(
"Failed to connect to target gateway: {}",
e
"Failed to connect to target gateway: {e}",
)));
}
Err(_) => {
@@ -1044,14 +1042,12 @@ impl LpConnectionHandler {
}
};
debug!(
"Opened persistent exit connection to {} for forwarding",
target_addr
);
debug!("Opened persistent exit connection to {target_addr} for forwarding",);
self.exit_stream = Some((stream, target_addr));
}
// Get mutable reference to the exit stream
#[allow(clippy::unwrap_used)]
let (target_stream, _) = self.exit_stream.as_mut().unwrap();
debug!(
gateway/src/node/lp_listener/mod.rs
+4 -5
View File
@@ -488,8 +488,7 @@ impl LpListener {
}
debug!(
"Accepting LP connection from {} ({} active connections)",
remote_addr, active_connections
"Accepting LP connection from {remote_addr} ({active_connections} active connections)"
);
// Increment connection counter
@@ -507,7 +506,7 @@ impl LpListener {
// Handler emits lifecycle metrics internally on success
// For errors, we need to emit them here since handler is consumed
if let Err(e) = result {
warn!("LP handler error for {}: {}", remote_addr, e);
warn!("LP handler error for {remote_addr}: {e}");
// Note: metrics are emitted in handle() for graceful path
// On error path, handle() returns early without emitting
// So we track errors here
@@ -516,7 +515,7 @@ impl LpListener {
// Decrement connection counter on exit
metrics.network.lp_connection_closed();
},
&format!("LP::{}", remote_addr),
&format!("LP::{remote_addr}"),
);
}
@@ -546,7 +545,7 @@ impl LpListener {
let handle = self.shutdown.try_spawn_named(
async move {
if let Err(e) = data_handler.run().await {
error!("LP data handler error: {}", e);
error!("LP data handler error: {e}");
}
},
"LP::DataHandler",
gateway/src/node/lp_listener/registration.rs
+13 -38
View File
@@ -224,10 +224,7 @@ pub async fn process_registration(
// without wasting credentials
let wg_key_str = request.wg_public_key.to_string();
if let Some(existing_response) = check_existing_registration(&wg_key_str, state).await {
info!(
"LP dVPN re-registration for existing peer {} (idempotent)",
wg_key_str
);
info!("LP dVPN re-registration for existing peer {wg_key_str} (idempotent)",);
inc!("lp_registration_dvpn_idempotent");
return existing_response;
}
@@ -242,12 +239,11 @@ pub async fn process_registration(
{
Ok(result) => result,
Err(e) => {
error!("LP WireGuard peer registration failed: {}", e);
error!("LP WireGuard peer registration failed: {e}");
inc!("lp_registration_dvpn_failed");
inc!("lp_errors_wg_peer_registration");
return LpRegistrationResponse::error(format!(
"WireGuard peer registration failed: {}",
e
"WireGuard peer registration failed: {e}",
));
}
};
@@ -263,24 +259,17 @@ pub async fn process_registration(
Ok(bandwidth) => bandwidth,
Err(e) => {
// Credential verification failed, remove the peer
warn!(
"LP credential verification failed for client {}: {}",
client_id, e
);
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(format!(
"Credential verification failed: {}",
e
"Credential verification failed: {e}",
));
}
};
@@ -300,11 +289,10 @@ pub async fn process_registration(
let client_identity = match ed25519::PublicKey::from_bytes(&client_ed25519_pubkey) {
Ok(key) => key,
Err(e) => {
warn!("LP Mixnet registration failed: invalid ed25519 key: {}", e);
warn!("LP Mixnet registration failed: invalid ed25519 key: {e}");
inc!("lp_registration_mixnet_failed");
return LpRegistrationResponse::error(format!(
"Invalid client ed25519 key: {}",
e
"Invalid client ed25519 key: {e}",
));
}
};
@@ -320,10 +308,7 @@ pub async fn process_registration(
.expect("This cannot fail, since the key is 32 bytes long"),
);
info!(
"LP Mixnet registration for client {}, session {}",
client_identity, session_id
);
info!("LP Mixnet registration for client {client_identity}, session {session_id}",);
// Verify credential with CredentialVerifier
let allocated_bandwidth = match credential_verification(
@@ -335,14 +320,10 @@ pub async fn process_registration(
{
Ok(bandwidth) => bandwidth,
Err(e) => {
warn!(
"LP Mixnet credential verification failed for client {}: {}",
client_identity, e
);
warn!("LP Mixnet credential verification failed for client {client_identity}: {e}");
inc!("lp_registration_mixnet_failed");
return LpRegistrationResponse::error(format!(
"Credential verification failed: {}",
e
"Credential verification failed: {e}"
));
}
};
@@ -359,10 +340,7 @@ pub async fn process_registration(
is_active_request_sender,
OffsetDateTime::now_utc(),
) {
warn!(
"LP Mixnet registration failed: client {} already registered",
client_identity
);
warn!("LP Mixnet registration failed: client {client_identity} already registered",);
inc!("lp_registration_mixnet_failed");
return LpRegistrationResponse::error("Client already registered".to_string());
}
@@ -376,10 +354,7 @@ pub async fn process_registration(
.expect("valid ed25519 key should convert to x25519")
.to_bytes();
info!(
"LP Mixnet registration successful (client: {})",
client_identity
);
info!("LP Mixnet registration successful (client: {client_identity})",);
inc!("lp_registration_mixnet_success");
LpRegistrationResponse::success_mixnet(
nym-registration-client/src/lp_client/client.rs
+5 -5
View File
@@ -72,7 +72,6 @@ impl LpRegistrationClient {
///
/// # Note
/// This creates the client. Call `perform_handshake()` to establish the LP session.
/// Each packet exchange opens a new TCP connection (packet-per-connection model).
pub fn new(
local_ed25519_keypair: Arc<ed25519::KeyPair>,
gateway_ed25519_public_key: ed25519::PublicKey,
@@ -666,8 +665,10 @@ impl LpRegistrationClient {
/// Sends registration request and receives response in a single operation.
///
/// This is the primary registration method. It acquires a bandwidth credential,
/// sends the registration request, and receives the response using the
/// packet-per-connection model.
/// sends the registration request, and receives the response
/// on the same underlying connection.
/// Do note that this method does **not** perform retries on network failures,
/// for that please use [`Self::register_with_retry`] instead
///
/// # Arguments
/// * `wg_keypair` - Client's WireGuard x25519 keypair
@@ -701,8 +702,7 @@ impl LpRegistrationClient {
.await
.map_err(|e| {
LpClientError::SendRegistrationRequest(format!(
"Failed to acquire bandwidth credential: {}",
e
"Failed to acquire bandwidth credential: {e}",
))
})?
.data;
service-providers/ip-packet-router/src/mixnet_listener.rs
+3 -6
View File
@@ -477,9 +477,8 @@ impl MixnetListener {
reconstructed.sender_tag,
current_time_ms,
)
.map_err(|e| {
log::warn!("KCP processing error: {}", e);
e
.inspect_err(|e| {
log::warn!("KCP processing error: {e}");
})?;
log::debug!(
@@ -507,9 +506,7 @@ impl MixnetListener {
.await
{
log::warn!(
"Error sending KCP-wrapped response for conv_id={}: {}",
conv_id,
e
"Error sending KCP-wrapped response for conv_id={conv_id}: {e}",
);
}
}
tools/nym-lp-client/Cargo.toml
+2 -2
View File
@@ -10,9 +10,9 @@ name = "nym-lp-client"
path = "src/main.rs"
[dependencies]
anyhow = "1"
anyhow = { workspace = true }
bytes = { workspace = true }
clap = { version = "4", features = ["derive"] }
clap = { workspace = true, features = ["derive"] }
rand = { workspace = true }
rand_chacha = { workspace = true }
serde = { workspace = true, features = ["derive"] }
tools/nym-lp-client/src/client.rs
+2 -2
View File
@@ -114,7 +114,7 @@ impl SpeedtestClient {
self.gateway.lp_address
);
let client_ip = "0.0.0.0".parse().unwrap();
let client_ip = "0.0.0.0".parse()?;
let mut lp_client = LpRegistrationClient::new_with_default_psk(
self.identity_keypair.clone(),
@@ -218,7 +218,7 @@ impl SpeedtestClient {
let local_addr = socket.local_addr()?;
let conv_id = compute_conv_id(local_addr, self.gateway.mix_host);
debug!("UDP socket bound to {}, conv_id={}", local_addr, conv_id);
debug!("UDP socket bound to {local_addr}, conv_id={conv_id}");
let session = KcpSession::new(conv_id);
let driver = KcpDriver::new(session);