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

LP: mixnet reg fixes (#6356)

Browse Source 
* removed x25519 key used within LP mixnet registration

* use Vec<u8> rather than BytesMut for LpAction::DeliverData

* introduced an explicit kind prefix for raw data sent and received within LP

* review nits
This commit is contained in:
Jędrzej Stuczyński
2026-01-23 13:21:52 +00:00
committed by GitHub
parent a63a1e745e
commit e2be2b0b34
20 changed files with 558 additions and 535 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/src/node/lp_listener/data_handler.rs
+1 -1
View File
@@ -191,7 +191,7 @@ impl LpDataHandler {
match action {
LpAction::DeliverData(data) => {
// Decrypted application data - forward as Sphinx packet
self.forward_sphinx_packet(&data).await?;
self.forward_sphinx_packet(&data.content).await?;
inc!("lp_data_packets_forwarded");
Ok(())
}
gateway/src/node/lp_listener/handler.rs
+86 -108
View File
@@ -1,18 +1,18 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use super::messages::LpRegistrationRequest;
use super::registration::process_registration;
use super::LpHandlerState;
use crate::error::GatewayError;
use nym_crypto::asymmetric::{ed25519, x25519};
use nym_lp::state_machine::{LpAction, LpInput};
use nym_lp::state_machine::{LpAction, LpData, LpDataKind, LpInput};
use nym_lp::{
codec::OuterAeadKey, message::ForwardPacketData, packet::LpHeader, LpMessage, LpPacket,
OuterHeader,
};
use nym_lp_transport::traits::LpTransport;
use nym_metrics::{add_histogram_obs, inc};
use nym_registration_common::LpRegistrationRequest;
use std::net::SocketAddr;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -535,16 +535,12 @@ where
})?
.map_err(|e| GatewayError::LpProtocolError(format!("State machine error: {}", e)))?;
let lp_session = state_machine.session().map_err(|e| {
GatewayError::LpProtocolError(format!("Session unavailable after processing: {}", e))
})?;
// Get outer key before releasing borrow
let outer_key = state_machine
.session()
.map_err(|e| {
GatewayError::LpProtocolError(format!(
"Session unavailable after processing: {}",
e
))
})?
.outer_aead_key();
let outer_key = lp_session.outer_aead_key();
drop(state_entry);
match action {
@@ -562,8 +558,7 @@ where
}
LpAction::DeliverData(data) => {
// Decrypted application data - process as registration/forwarding
self.handle_decrypted_payload(receiver_idx, data.to_vec())
.await
self.handle_decrypted_payload(receiver_idx, data).await
}
LpAction::SubsessionComplete {
packet: ready_packet,
@@ -597,38 +592,41 @@ where
async fn handle_decrypted_payload(
&mut self,
receiver_idx: u32,
decrypted_bytes: Vec<u8>,
decrypted_data: LpData,
) -> Result<(), GatewayError> {
let remote = self.remote_addr;
// Try to deserialize as LpRegistrationRequest first (most common case after handshake)
if let Ok(request) = LpRegistrationRequest::try_deserialise(&decrypted_bytes) {
debug!(
"LP registration request from {remote} (receiver_idx={receiver_idx}): mode={:?}",
request.mode
);
return self
.handle_registration_request(receiver_idx, request)
.await;
}
let bytes = decrypted_data.content;
match decrypted_data.kind {
LpDataKind::Registration => {
let request = LpRegistrationRequest::try_deserialise(&bytes).map_err(|err| {
GatewayError::LpProtocolError(format!("malformed LpRegistrationRequest: {err}"))
})?;
// Try to deserialize as ForwardPacketData (entry gateway forwarding to exit)
if let Ok(forward_data) = ForwardPacketData::decode(&decrypted_bytes) {
debug!(
"LP forward request from {remote} (receiver_idx={receiver_idx}) to {}",
forward_data.target_lp_address
);
return self
.handle_forwarding_request(receiver_idx, forward_data)
.await;
}
debug!(
"LP registration request from {remote} (receiver_idx={receiver_idx}): mode={:?}",
request.mode());
// Neither registration nor forwarding - unknown payload type
warn!("Unknown transport payload type from {remote} (receiver_idx={receiver_idx})");
inc!("lp_errors_unknown_payload_type");
Err(GatewayError::LpProtocolError(
"Unknown transport payload type (not registration or forwarding)".to_string(),
))
self.handle_registration_request(receiver_idx, request)
.await
}
LpDataKind::Forward => {
let forward_data = ForwardPacketData::decode(&bytes).map_err(|err| {
GatewayError::LpProtocolError(format!("malformed ForwardPacketData: {err}"))
})?;
self.handle_forwarding_request(receiver_idx, forward_data)
.await
}
LpDataKind::Opaque => {
// Neither registration nor forwarding - unknown payload type
warn!("Unknown transport payload type from {remote} (receiver_idx={receiver_idx}). dropping {} bytes", bytes.len());
inc!("lp_errors_unknown_payload_type");
Err(GatewayError::LpProtocolError(
"Unknown transport payload type (not registration or forwarding)".to_string(),
))
}
}
}
/// Handle subsession completion - promote subsession to new session
@@ -699,6 +697,48 @@ where
Ok(())
}
/// Attempt to wrap and send specified response back to the client
async fn send_response_packet(
&mut self,
receiver_idx: u32,
serialised_response: Vec<u8>,
response_kind: LpDataKind,
) -> Result<(), GatewayError> {
let session_entry = self
.state
.session_states
.get(&receiver_idx)
.ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {receiver_idx}"))
})?;
// Access session via state machine for subsession support
let session = session_entry
.value()
.state
.session()
.map_err(|e| GatewayError::LpProtocolError(format!("Session error: {e}")))?;
let wrapped_lp_data = LpData::new(response_kind, serialised_response);
let data_bytes = wrapped_lp_data.to_vec();
let encrypted_message = session.encrypt_data(&data_bytes).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to encrypt response: {e}"))
})?;
let response_packet = session.next_packet(encrypted_message).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to create response packet: {e}"))
})?;
let outer_key = session.outer_aead_key();
drop(session_entry);
// Send response (encrypted with outer AEAD)
self.send_lp_packet(response_packet, outer_key.as_ref())
.await?;
Ok(())
}
/// Handle registration request on an established session
async fn handle_registration_request(
&mut self,
@@ -707,43 +747,11 @@ where
) -> Result<(), GatewayError> {
// Process registration (might modify state)
let response = process_registration(request, &self.state).await;
let response_bytes = response.serialise().map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to serialize response: {e}"))
})?;
// Acquire session lock for encryption and get outer AEAD key
let (response_packet, outer_key) = {
let session_entry = self
.state
.session_states
.get(&receiver_idx)
.ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx))
})?;
// Access session via state machine for subsession support
let session = session_entry
.value()
.state
.session()
.map_err(|e| GatewayError::LpProtocolError(format!("Session error: {}", e)))?;
// Serialize and encrypt response
let response_bytes = response.serialise().map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to serialize response: {}", e))
})?;
let encrypted_message = session.encrypt_data(&response_bytes).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to encrypt response: {}", e))
})?;
let packet = session.next_packet(encrypted_message).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to create response packet: {}", e))
})?;
// Get outer AEAD key for packet encryption
let outer_key = session.outer_aead_key();
(packet, outer_key)
};
// Send response (encrypted with outer AEAD)
self.send_lp_packet(response_packet, outer_key.as_ref())
self.send_response_packet(receiver_idx, response_bytes, LpDataKind::Registration)
.await?;
if response.success {
@@ -767,40 +775,10 @@ where
receiver_idx: u32,
forward_data: ForwardPacketData,
) -> Result<(), GatewayError> {
// Forward the packet to the target gateway
// Forward the packet to the target gateway and retrieve its response
let response_bytes = self.handle_forward_packet(forward_data).await?;
// Encrypt response for client and get outer AEAD key
let (response_packet, outer_key) = {
let session_entry = self
.state
.session_states
.get(&receiver_idx)
.ok_or_else(|| {
GatewayError::LpProtocolError(format!("Session not found: {}", receiver_idx))
})?;
// Access session via state machine for subsession support
let session = session_entry
.value()
.state
.session()
.map_err(|e| GatewayError::LpProtocolError(format!("Session error: {}", e)))?;
let encrypted_message = session.encrypt_data(&response_bytes).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to encrypt forward response: {}", e))
})?;
let packet = session.next_packet(encrypted_message).map_err(|e| {
GatewayError::LpProtocolError(format!("Failed to create response packet: {}", e))
})?;
// Get outer AEAD key for packet encryption
let outer_key = session.outer_aead_key();
(packet, outer_key)
};
// Send encrypted response to client (encrypted with outer AEAD)
self.send_lp_packet(response_packet, outer_key.as_ref())
self.send_response_packet(receiver_idx, response_bytes, LpDataKind::Forward)
.await?;
debug!(
gateway/src/node/lp_listener/messages.rs
-10
View File
@@ -1,10 +0,0 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
//! LP registration message types.
//!
//! Re-exports shared message types from nym-registration-common.
pub use nym_registration_common::{
LpGatewayData, LpRegistrationRequest, LpRegistrationResponse, RegistrationMode,
};
gateway/src/node/lp_listener/mod.rs
+1 -2
View File
@@ -78,7 +78,6 @@ pub use nym_mixnet_client::forwarder::{
};
use nym_node_metrics::NymNodeMetrics;
use nym_task::ShutdownTracker;
pub use nym_wireguard::{PeerControlRequest, WireguardGatewayData};
use std::net::{IpAddr, Ipv6Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
@@ -87,10 +86,10 @@ use tokio::sync::{mpsc, Semaphore};
use tracing::*;
pub use nym_lp::peer::LpLocalPeer;
pub use nym_wireguard::{PeerControlRequest, WireguardGatewayData};
mod data_handler;
pub mod handler;
mod messages;
mod registration;
/// Configuration for LP listener
gateway/src/node/lp_listener/registration.rs
+117 -159
View File
@@ -1,9 +1,6 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use super::messages::{
LpGatewayData, LpRegistrationRequest, LpRegistrationResponse, RegistrationMode,
};
use super::LpHandlerState;
use crate::error::GatewayError;
use crate::node::client_handling::websocket::message_receiver::IsActive;
@@ -16,12 +13,14 @@ use nym_credential_verification::{
ClientBandwidth, CredentialVerifier,
};
use nym_credentials_interface::CredentialSpendingData;
use nym_crypto::asymmetric::ed25519;
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_registration_common::{
GatewayData, LpDvpnRegistrationRequest, LpMixnetGatewayData, LpMixnetRegistrationRequest,
LpRegistrationData, LpRegistrationRequest, LpRegistrationResponse,
};
use nym_wireguard::PeerControlRequest;
use std::sync::Arc;
use time::OffsetDateTime;
@@ -195,12 +194,121 @@ async fn check_existing_registration(
))
}
async fn process_dvpn_registration(
request: Box<LpDvpnRegistrationRequest>,
state: &LpHandlerState,
) -> LpRegistrationResponse {
// Track dVPN registration attempts
inc!("lp_registration_dvpn_attempts");
// Check for idempotent re-registration (same WG key already registered)
// This allows clients to retry registration after network failures
// 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 {wg_key_str} (idempotent)",);
inc!("lp_registration_dvpn_idempotent");
return existing_response;
}
// 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.ticket_type,
state,
)
.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(format!(
"WireGuard peer registration failed: {e}",
));
}
};
// Verify credential with CredentialVerifier (handles double-spend, storage, etc.)
let allocated_bandwidth =
match credential_verification(state.ecash_verifier.clone(), request.credential, client_id)
.await
{
Ok(bandwidth) => bandwidth,
Err(e) => {
// Credential verification failed, remove the peer
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}"
);
}
return LpRegistrationResponse::error(format!(
"Credential verification failed: {e}",
));
}
};
info!("LP dVPN registration successful (client_id: {client_id})");
inc!("lp_registration_dvpn_success");
LpRegistrationResponse::success(allocated_bandwidth, gateway_data)
}
async fn process_mixnet_registration(
request: LpMixnetRegistrationRequest,
state: &LpHandlerState,
) -> LpRegistrationResponse {
let session_id = rand::random::<u32>();
// Track mixnet registration attempts
inc!("lp_registration_mixnet_attempts");
// Derive destination address for ActiveClientsStore lookup
let client_identity = request.client_ed25519_pubkey;
let client_address = client_identity.derive_destination_address();
info!("LP Mixnet registration for client {client_identity}, session {session_id}");
warn!("unimplemented: LP mixnet registration initial bandwidth allocation");
// (the old implementation was wrong - it wasn't creating correct db entries)
// Create channels for client message delivery
let (mix_sender, _mix_receiver) = mpsc::unbounded();
let (is_active_request_sender, _is_active_request_receiver) =
mpsc::unbounded::<oneshot::Sender<IsActive>>();
// Insert client into ActiveClientsStore for SURB reply delivery
if !state.active_clients_store.insert_remote(
client_address,
mix_sender,
is_active_request_sender,
OffsetDateTime::now_utc(),
) {
warn!("LP Mixnet registration failed: client {client_identity} already registered",);
inc!("lp_registration_mixnet_failed");
return LpRegistrationResponse::error("Client already registered".to_string());
}
// Get gateway identity and derive sphinx key
let gateway_identity = *state.local_lp_peer.ed25519().public_key();
info!("LP Mixnet registration successful (client: {client_identity})",);
inc!("lp_registration_mixnet_success");
LpRegistrationResponse::success_mixnet(0, LpMixnetGatewayData { gateway_identity })
}
/// Process an LP registration request
pub async fn process_registration(
request: LpRegistrationRequest,
state: &LpHandlerState,
) -> LpRegistrationResponse {
let session_id = rand::random::<u32>();
let registration_start = std::time::Instant::now();
// Track total registration attempts
@@ -214,159 +322,9 @@ pub async fn process_registration(
}
// 2. Process based on mode
let result = match request.mode {
RegistrationMode::Dvpn => {
// Track dVPN registration attempts
inc!("lp_registration_dvpn_attempts");
// Check for idempotent re-registration (same WG key already registered)
// This allows clients to retry registration after network failures
// 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 {wg_key_str} (idempotent)",);
inc!("lp_registration_dvpn_idempotent");
return existing_response;
}
// 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.ticket_type,
state,
)
.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(format!(
"WireGuard peer registration failed: {e}",
));
}
};
// Verify credential with CredentialVerifier (handles double-spend, storage, etc.)
let allocated_bandwidth = match credential_verification(
state.ecash_verifier.clone(),
request.credential,
client_id,
)
.await
{
Ok(bandwidth) => bandwidth,
Err(e) => {
// Credential verification failed, remove the peer
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}");
}
return LpRegistrationResponse::error(format!(
"Credential verification failed: {e}",
));
}
};
info!("LP dVPN registration successful (client_id: {})", client_id);
inc!("lp_registration_dvpn_success");
LpRegistrationResponse::success(allocated_bandwidth, gateway_data)
}
RegistrationMode::Mixnet {
client_ed25519_pubkey,
client_x25519_pubkey: _,
} => {
// Track mixnet registration attempts
inc!("lp_registration_mixnet_attempts");
// Parse client's ed25519 public key
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}");
inc!("lp_registration_mixnet_failed");
return LpRegistrationResponse::error(format!(
"Invalid client ed25519 key: {e}",
));
}
};
// Derive destination address for ActiveClientsStore lookup
let client_address = client_identity.derive_destination_address();
// Generate client_id for credential verification (first 8 bytes of ed25519 key)
#[allow(clippy::expect_used)]
let client_id = i64::from_be_bytes(
client_ed25519_pubkey[0..8]
.try_into()
.expect("This cannot fail, since the key is 32 bytes long"),
);
info!("LP Mixnet registration for client {client_identity}, session {session_id}",);
// Verify credential with CredentialVerifier
let allocated_bandwidth = match credential_verification(
state.ecash_verifier.clone(),
request.credential,
client_id,
)
.await
{
Ok(bandwidth) => bandwidth,
Err(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}"
));
}
};
// Create channels for client message delivery
let (mix_sender, _mix_receiver) = mpsc::unbounded();
let (is_active_request_sender, _is_active_request_receiver) =
mpsc::unbounded::<oneshot::Sender<IsActive>>();
// Insert client into ActiveClientsStore for SURB reply delivery
if !state.active_clients_store.insert_remote(
client_address,
mix_sender,
is_active_request_sender,
OffsetDateTime::now_utc(),
) {
warn!("LP Mixnet registration failed: client {client_identity} already registered",);
inc!("lp_registration_mixnet_failed");
return LpRegistrationResponse::error("Client already registered".to_string());
}
// Get gateway identity and derive sphinx key
let ed25519_key = state.local_lp_peer.ed25519().public_key();
let gateway_identity = ed25519_key.to_bytes();
warn!("TEMPORARY ed25519 -> x25519 conversion");
#[allow(clippy::expect_used)]
let gateway_sphinx_key = ed25519_key
.to_x25519()
.expect("valid ed25519 key should convert to x25519")
.to_bytes();
info!("LP Mixnet registration successful (client: {client_identity})",);
inc!("lp_registration_mixnet_success");
LpRegistrationResponse::success_mixnet(
allocated_bandwidth,
LpGatewayData {
gateway_identity,
gateway_sphinx_key,
},
)
}
let result = match request.registration_data {
LpRegistrationData::Dvpn { data } => process_dvpn_registration(data, state).await,
LpRegistrationData::Mixnet { data } => process_mixnet_registration(data, state).await,
};
// Track registration duration