feat: Lewes Protocol with PSQv2 (#6491)
* merging georgio/lp-psqv2-integration * use authenicator on the responder's side * nym-lp crate compiling * moved the e2e test to nym-lp * move key generation to peer * moved principal generation * update KKTResponder * encapsulation key parsing * Adding concrete types within KKT exchange * initiator side of the full handshake * responder side of the handshake and full e2e test * fixed unit-tests within nym-kkt * LpSession cleanup * helpers for Transport * revamp of the transport traits and initial work on client-side transport * compiling nym-crypto * 'working' client-entry dvpn reg * Fix key conversion * Slightly reduce use of rand08 * reverted back to libcrux repo refs * intial telescoping reg * removing dead code * wip * moved data encryption into the state machine * restoring nym-lp tests * update lp api model * Add receiver index derivation * Add receiver index derivation * use derived receiver index * feat: add kem key generation to nodes * generate fresh x25519, mlkem768 and mceliece keys on config migration * add lp peer config * nym-node startup cleanup * removed dependency on pre-rand09 from nym-lp * re-expose LP information on the http API * fixed tests compilation * add peer config happy path tests * formatting * add more tests and fix bug * better docs * clippy and formatting issues * return error on mceliece within NestedSession * wasm fixes * removed legacy nym-vpn-lib-wasm * fixing wasm for real this time * additional fixes * add payload to kkt * make clippy happy * moved LP to nym-node crate * cargo fmt * integrate lpconfig payload * fix response size trait impl * Migrate receiver index * Change receiver index to u32 and regorganize crates * clippy * hopefully final wasm fixes * simple conversion method from semver to ciphersuite * updated nym-node config template * chore: remove duplicated code --------- Co-authored-by: Georgio Nicolas <me@georgio.xyz>
This commit is contained in:
Jędrzej Stuczyński
committed by
GitHub
GitHub
parent
e5c3f39a57
commit
f6bd511599
@@ -6,6 +6,10 @@ use crate::error::{KeyIOFailure, NymNodeError};
|
||||
use crate::node::key_rotation::key::{SphinxPrivateKey, SphinxPublicKey};
|
||||
use crate::node::nym_apis_client::NymApisClient;
|
||||
use nym_crypto::asymmetric::{ed25519, x25519};
|
||||
use nym_kkt::keys::storage_wrappers::StorableKey;
|
||||
use nym_kkt::keys::{
|
||||
DHKeyPair, DHPrivateKey, MlKem768KeyPair, MlKem768PrivateKey, MlKem768PublicKey, mceliece,
|
||||
};
|
||||
use nym_node_requests::api::v1::node::models::NodeDescription;
|
||||
use nym_pemstore::KeyPairPath;
|
||||
use nym_pemstore::traits::{PemStorableKey, PemStorableKeyPair};
|
||||
@@ -169,6 +173,34 @@ pub(crate) fn load_x25519_wireguard_keypair(
|
||||
Ok(load_keypair(paths, "x25519-wireguard")?)
|
||||
}
|
||||
|
||||
fn load_lp_key<S, P>(path: P, name: &'static str) -> Result<S, NymNodeError>
|
||||
where
|
||||
P: AsRef<Path>,
|
||||
S: StorableKey,
|
||||
{
|
||||
let repr = load_key::<<S as StorableKey>::StorableRepresentation<'_>, _>(path, name)?;
|
||||
Ok(S::from_storable(repr)?)
|
||||
}
|
||||
|
||||
pub(crate) fn load_x25519_lp_keypair(paths: &KeyPairPath) -> Result<DHKeyPair, NymNodeError> {
|
||||
let pk: DHPrivateKey = load_lp_key(&paths.private_key_path, "x25519-lp-private-key")?;
|
||||
Ok(DHKeyPair::from(pk))
|
||||
}
|
||||
|
||||
pub(crate) fn load_mlkem768_keypair(paths: &KeyPairPath) -> Result<MlKem768KeyPair, NymNodeError> {
|
||||
let sk: MlKem768PrivateKey = load_lp_key(&paths.private_key_path, "mlkem768-private-key")?;
|
||||
let pk: MlKem768PublicKey = load_lp_key(&paths.public_key_path, "mlkem768-public-key")?;
|
||||
Ok(MlKem768KeyPair::from(sk, pk))
|
||||
}
|
||||
|
||||
pub(crate) fn load_mceliece_keypair(
|
||||
paths: &KeyPairPath,
|
||||
) -> Result<mceliece::KeyPair, NymNodeError> {
|
||||
let sk: mceliece::SecretKey = load_lp_key(&paths.private_key_path, "mceliece-private-key")?;
|
||||
let pk: mceliece::PublicKey = load_lp_key(&paths.public_key_path, "mceliece-public-key")?;
|
||||
Ok(mceliece::KeyPair { sk, pk })
|
||||
}
|
||||
|
||||
pub(crate) fn store_ed25519_identity_keypair(
|
||||
keys: &ed25519::KeyPair,
|
||||
paths: &KeyPairPath,
|
||||
@@ -183,6 +215,57 @@ pub(crate) fn store_x25519_noise_keypair(
|
||||
Ok(store_keypair(keys, paths, "x25519-noise")?)
|
||||
}
|
||||
|
||||
pub(crate) fn store_x25519_lp_keypair(
|
||||
keys: &DHKeyPair,
|
||||
paths: &KeyPairPath,
|
||||
) -> Result<(), NymNodeError> {
|
||||
store_key(
|
||||
&keys.pk.to_storable(),
|
||||
&paths.public_key_path,
|
||||
"x25519-lp-public-key",
|
||||
)?;
|
||||
store_key(
|
||||
&keys.sk().to_storable(),
|
||||
&paths.private_key_path,
|
||||
"x25519-lp-private-key",
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn store_mlkem768_keypair(
|
||||
keys: &MlKem768KeyPair,
|
||||
paths: &KeyPairPath,
|
||||
) -> Result<(), NymNodeError> {
|
||||
store_key(
|
||||
&keys.public_key().to_storable(),
|
||||
&paths.public_key_path,
|
||||
"mlkem768-public-key",
|
||||
)?;
|
||||
store_key(
|
||||
&keys.private_key().to_storable(),
|
||||
&paths.private_key_path,
|
||||
"mlkem768-private-key",
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn store_mceliece_keypair(
|
||||
keys: &mceliece::KeyPair,
|
||||
paths: &KeyPairPath,
|
||||
) -> Result<(), NymNodeError> {
|
||||
store_key(
|
||||
&keys.pk.to_storable(),
|
||||
&paths.public_key_path,
|
||||
"mceliece-public-key",
|
||||
)?;
|
||||
store_key(
|
||||
&keys.sk.to_storable(),
|
||||
&paths.private_key_path,
|
||||
"mceliece-private-key",
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn get_current_rotation_id(
|
||||
nym_apis: &[Url],
|
||||
fallback_nyxd: &[Url],
|
||||
|
||||
@@ -3,13 +3,13 @@
|
||||
|
||||
use axum::Router;
|
||||
use axum::routing::get;
|
||||
use nym_node_requests::api::v1::lewes_protocol::models;
|
||||
use nym_node_requests::api::SignedLewesProtocol;
|
||||
|
||||
pub mod root;
|
||||
|
||||
#[derive(Debug, Default, Clone)]
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Config {
|
||||
pub details: Option<models::LewesProtocol>,
|
||||
pub details: SignedLewesProtocol,
|
||||
}
|
||||
|
||||
pub(crate) fn routes<S: Send + Sync + 'static + Clone>(config: Config) -> Router<S> {
|
||||
|
||||
@@ -4,30 +4,29 @@
|
||||
use axum::extract::Query;
|
||||
use axum::http::StatusCode;
|
||||
use nym_http_api_common::{FormattedResponse, OutputParams};
|
||||
use nym_node_requests::api::v1::lewes_protocol::models::LewesProtocol;
|
||||
use nym_node_requests::api::{SignedLewesProtocol, SignedLewesProtocolInfo};
|
||||
|
||||
/// Returns root Lewes Protocol information
|
||||
#[utoipa::path(
|
||||
get,
|
||||
path = "",
|
||||
context_path = "/api/v1/lewes-protocol",
|
||||
path = "/lewes-protocol",
|
||||
context_path = "/api/v1",
|
||||
tag = "Lewes Protocol",
|
||||
responses(
|
||||
(status = 501, description = "the endpoint hasn't been implemented yet"),
|
||||
(status = 200, content(
|
||||
(LewesProtocol = "application/json"),
|
||||
(LewesProtocol = "application/yaml"),
|
||||
(LewesProtocol = "application/bincode")
|
||||
(SignedLewesProtocolInfo = "application/json"),
|
||||
(SignedLewesProtocolInfo = "application/yaml"),
|
||||
(SignedLewesProtocolInfo = "application/bincode")
|
||||
))
|
||||
),
|
||||
params(OutputParams)
|
||||
)]
|
||||
pub(crate) async fn root_lewes_protocol(
|
||||
config: Option<LewesProtocol>,
|
||||
config: SignedLewesProtocol,
|
||||
Query(output): Query<OutputParams>,
|
||||
) -> Result<LewesProtocolResponse, StatusCode> {
|
||||
let config = config.ok_or(StatusCode::NOT_IMPLEMENTED)?;
|
||||
Ok(output.to_response(config))
|
||||
}
|
||||
|
||||
pub type LewesProtocolResponse = FormattedResponse<LewesProtocol>;
|
||||
pub type LewesProtocolResponse = FormattedResponse<SignedLewesProtocol>;
|
||||
|
||||
@@ -16,7 +16,8 @@ use nym_node_requests::api::{SignedDataHostInfo, v1::node::models::SignedHostInf
|
||||
responses(
|
||||
(status = 200, content(
|
||||
(SignedDataHostInfo = "application/json"),
|
||||
(SignedDataHostInfo = "application/yaml")
|
||||
(SignedDataHostInfo = "application/yaml"),
|
||||
(SignedDataHostInfo = "application/bincode")
|
||||
))
|
||||
),
|
||||
params(OutputParams)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
// SPDX-License-Identifier: GPL-3.0-only
|
||||
|
||||
use crate::node::http::NymNodeHttpServer;
|
||||
use crate::node::http::api::v1::lewes_protocol;
|
||||
use crate::node::http::error::NymNodeHttpError;
|
||||
use crate::node::http::state::AppState;
|
||||
use axum::Router;
|
||||
@@ -9,6 +10,7 @@ use axum::response::Redirect;
|
||||
use axum::routing::get;
|
||||
use nym_bin_common::bin_info_owned;
|
||||
use nym_http_api_common::middleware::logging;
|
||||
use nym_node_requests::api::SignedLewesProtocol;
|
||||
use nym_node_requests::api::v1::authenticator::models::Authenticator;
|
||||
use nym_node_requests::api::v1::gateway::models::{Bridges, Gateway};
|
||||
use nym_node_requests::api::v1::ip_packet_router::models::IpPacketRouter;
|
||||
@@ -33,7 +35,7 @@ pub struct HttpServerConfig {
|
||||
}
|
||||
|
||||
impl HttpServerConfig {
|
||||
pub fn new() -> Self {
|
||||
pub fn new(signed_lewes_protocol: SignedLewesProtocol) -> Self {
|
||||
HttpServerConfig {
|
||||
landing: Default::default(),
|
||||
api: api::Config {
|
||||
@@ -52,7 +54,9 @@ impl HttpServerConfig {
|
||||
network_requester: Default::default(),
|
||||
ip_packet_router: Default::default(),
|
||||
authenticator: Default::default(),
|
||||
lewes_protocol: Default::default(),
|
||||
lewes_protocol: lewes_protocol::Config {
|
||||
details: signed_lewes_protocol,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@@ -113,14 +113,11 @@ impl PemStorableKey for SphinxPrivateKey {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use rand::SeedableRng;
|
||||
use rand_chacha::ChaCha20Rng;
|
||||
use nym_test_utils::helpers::deterministic_rng;
|
||||
|
||||
#[test]
|
||||
fn private_key_bytes_convertion() {
|
||||
// Set up a deterministic RNG.
|
||||
let seed = [42u8; 32];
|
||||
let mut rng = ChaCha20Rng::from_seed(seed);
|
||||
let mut rng = deterministic_rng();
|
||||
|
||||
let key = SphinxPrivateKey {
|
||||
rotation_id: 42,
|
||||
|
||||
@@ -0,0 +1,237 @@
|
||||
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: GPL-3.0-only
|
||||
|
||||
//! LP Data Handler - UDP listener for LP data plane (port 51264)
|
||||
//!
|
||||
//! This module handles the data plane for LP clients that have completed registration
|
||||
//! via the control plane (TCP:41264). LP-wrapped Sphinx packets arrive here, get
|
||||
//! decrypted, and are forwarded into the mixnet.
|
||||
//!
|
||||
//! # Packet Flow
|
||||
//!
|
||||
//! ```text
|
||||
//! LP Client → UDP:51264 → LP Data Handler → Mixnet Entry
|
||||
//! LP(Sphinx) decrypt LP forward Sphinx
|
||||
//! ```
|
||||
//!
|
||||
|
||||
use super::LpHandlerState;
|
||||
use crate::error::NymNodeError;
|
||||
use crate::node::lp::error::LpHandlerError;
|
||||
use nym_lp::packet::OuterHeader;
|
||||
use nym_metrics::inc;
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
use tokio::net::UdpSocket;
|
||||
use tracing::*;
|
||||
|
||||
/// Maximum UDP packet size we'll accept
|
||||
/// Sphinx packets are typically ~2KB, LP overhead is ~50 bytes, so 4KB is plenty
|
||||
const MAX_UDP_PACKET_SIZE: usize = 4096;
|
||||
|
||||
/// LP Data Handler for UDP data plane
|
||||
pub struct LpDataHandler {
|
||||
/// UDP socket for receiving LP-wrapped Sphinx packets
|
||||
socket: Arc<UdpSocket>,
|
||||
|
||||
/// Shared state with TCP control plane
|
||||
#[allow(dead_code)]
|
||||
state: LpHandlerState,
|
||||
|
||||
/// Shutdown token
|
||||
shutdown: nym_task::ShutdownToken,
|
||||
}
|
||||
|
||||
impl LpDataHandler {
|
||||
/// Create a new LP data handler
|
||||
pub async fn new(
|
||||
bind_addr: SocketAddr,
|
||||
state: LpHandlerState,
|
||||
shutdown: nym_task::ShutdownToken,
|
||||
) -> Result<Self, NymNodeError> {
|
||||
let socket = UdpSocket::bind(bind_addr).await.map_err(|source| {
|
||||
error!("Failed to bind LP data socket to {bind_addr}: {source}");
|
||||
NymNodeError::LpBindFailure {
|
||||
address: bind_addr,
|
||||
source,
|
||||
}
|
||||
})?;
|
||||
|
||||
info!("LP data handler listening on UDP {bind_addr}");
|
||||
|
||||
Ok(Self {
|
||||
socket: Arc::new(socket),
|
||||
state,
|
||||
shutdown,
|
||||
})
|
||||
}
|
||||
|
||||
/// Run the UDP packet receive loop
|
||||
pub async fn run(self) -> Result<(), LpHandlerError> {
|
||||
let mut buf = vec![0u8; MAX_UDP_PACKET_SIZE];
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
|
||||
_ = self.shutdown.cancelled() => {
|
||||
info!("LP data handler: received shutdown signal");
|
||||
break;
|
||||
}
|
||||
|
||||
result = self.socket.recv_from(&mut buf) => {
|
||||
match result {
|
||||
Ok((len, src_addr)) => {
|
||||
// Process packet in place (no spawn - UDP is fast)
|
||||
if let Err(e) = self.handle_packet(&buf[..len], src_addr).await {
|
||||
debug!("LP data packet error from {src_addr}: {e}");
|
||||
inc!("lp_data_packet_errors");
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("LP data socket recv error: {e}");
|
||||
inc!("lp_data_recv_errors");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
info!("LP data handler shutdown complete");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Handle a single UDP packet
|
||||
///
|
||||
/// # Packet Processing Steps
|
||||
/// 1. Parse LP header to get receiver_idx (for routing)
|
||||
/// 2. Look up session state machine by receiver_idx
|
||||
/// 3. Process packet through state machine (handles decryption + replay protection)
|
||||
/// 4. Forward decrypted Sphinx packet to mixnet
|
||||
///
|
||||
/// # Security
|
||||
/// The state machine's `process_input()` method handles replay protection by:
|
||||
/// - Checking packet counter against receiving window
|
||||
/// - Marking counter as used after successful decryption
|
||||
///
|
||||
/// This prevents replay attacks where captured packets are re-sent.
|
||||
async fn handle_packet(
|
||||
&self,
|
||||
packet: &[u8],
|
||||
src_addr: SocketAddr,
|
||||
) -> Result<(), LpHandlerError> {
|
||||
inc!("lp_data_packets_received");
|
||||
|
||||
let _ = OuterHeader::parse(packet)?;
|
||||
trace!(
|
||||
"received {} bytes from {src_addr} on the unimplemented LP Data endpoint",
|
||||
packet.len()
|
||||
);
|
||||
|
||||
Err(LpHandlerError::UnimplementedDataChannel)
|
||||
// leave old code for future reference
|
||||
|
||||
//
|
||||
// // Step 1: Parse LP header (always cleartext for routing)
|
||||
// let header = nym_lp::codec::parse_lp_header_only(packet).map_err(|e| {
|
||||
// LpHandlerError::LpProtocolError(format!("Failed to parse LP header: {}", e))
|
||||
// })?;
|
||||
//
|
||||
// let receiver_idx = header.receiver_idx;
|
||||
// let counter = header.counter;
|
||||
// let len = 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
|
||||
// .state
|
||||
// .session_states
|
||||
// .get_mut(&receiver_idx)
|
||||
// .ok_or_else(|| {
|
||||
// inc!("lp_data_unknown_session");
|
||||
// LpHandlerError::LpProtocolError(format!(
|
||||
// "Unknown session for receiver_idx {receiver_idx}"
|
||||
// ))
|
||||
// })?;
|
||||
//
|
||||
// // Update last activity timestamp
|
||||
// state_entry.value().touch();
|
||||
//
|
||||
// // Step 3: Get outer AEAD key for packet parsing
|
||||
// let outer_key = state_entry
|
||||
// .value()
|
||||
// .state
|
||||
// .session()
|
||||
// .map_err(|e| LpHandlerError::LpProtocolError(format!("Session error: {e}")))?
|
||||
// .outer_aead_key();
|
||||
//
|
||||
// // Parse full packet with outer AEAD decryption
|
||||
// let lp_packet = nym_lp::codec::parse_lp_packet(packet, Some(outer_key)).map_err(|e| {
|
||||
// inc!("lp_data_decrypt_errors");
|
||||
// LpHandlerError::LpProtocolError(format!("Failed to decrypt LP packet: {}", e))
|
||||
// })?;
|
||||
//
|
||||
// // Step 4: Process packet through state machine
|
||||
// // This handles:
|
||||
// // - Replay protection (counter check + mark)
|
||||
// // - Inner Noise decryption
|
||||
// // - Subsession handling if applicable
|
||||
// let state_machine = &mut state_entry.value_mut().state;
|
||||
//
|
||||
// let action = state_machine
|
||||
// .process_input(LpInput::ReceivePacket(lp_packet))
|
||||
// .ok_or_else(|| {
|
||||
// LpHandlerError::LpProtocolError("State machine returned no action".to_string())
|
||||
// })?
|
||||
// .map_err(|e| {
|
||||
// inc!("lp_data_state_machine_errors");
|
||||
// LpHandlerError::LpProtocolError(format!("State machine error: {}", e))
|
||||
// })?;
|
||||
//
|
||||
// // Release session lock before forwarding
|
||||
// drop(state_entry);
|
||||
//
|
||||
// // Step 5: Handle the action from state machine
|
||||
// match action {
|
||||
// LpAction::DeliverData(data) => {
|
||||
// // Decrypted application data - forward as Sphinx packet
|
||||
// self.forward_sphinx_packet(&data.content).await?;
|
||||
// inc!("lp_data_packets_forwarded");
|
||||
// Ok(())
|
||||
// }
|
||||
// LpAction::SendPacket(_response_packet) => {
|
||||
// // 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={receiver_idx}) - use TCP for rekeying",
|
||||
// );
|
||||
// inc!("lp_data_ignored_send_actions");
|
||||
// Ok(())
|
||||
// }
|
||||
// other => {
|
||||
// warn!(
|
||||
// "Unexpected action on UDP data plane from {}: {:?}",
|
||||
// src_addr, other
|
||||
// );
|
||||
// inc!("lp_data_unexpected_actions");
|
||||
// Err(LpHandlerError::LpProtocolError(format!(
|
||||
// "Unexpected state machine action on UDP: {:?}",
|
||||
// other
|
||||
// )))
|
||||
// }
|
||||
// }
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
// Sphinx packets are typically around 2KB
|
||||
// LP overhead is small (~50 bytes header + AEAD tag)
|
||||
// 4KB should be plenty with room to spare
|
||||
const _: () = {
|
||||
assert!(MAX_UDP_PACKET_SIZE >= 2048 + 100);
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use crate::node::lp::LpReceiverIndex;
|
||||
use nym_lp::state_machine::{LpAction, LpDataKind};
|
||||
use nym_lp::transport::LpTransportError;
|
||||
use nym_lp::{LpError, packet::MalformedLpPacketError};
|
||||
use std::net::SocketAddr;
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum LpHandlerError {
|
||||
#[error("failed to establish egress connection to {egress}: {reason}")]
|
||||
ConnectionFailure { egress: SocketAddr, reason: String },
|
||||
|
||||
#[error(transparent)]
|
||||
LpTransportError(#[from] LpTransportError),
|
||||
|
||||
#[error("missing session state for {receiver_index} - has it been removed due to inactivity?")]
|
||||
MissingLpSession { receiver_index: LpReceiverIndex },
|
||||
|
||||
#[error(transparent)]
|
||||
LpProtocolError(#[from] LpError),
|
||||
|
||||
#[error("the initial KKT/PSQ handshake has not been completed")]
|
||||
IncompleteHandshake,
|
||||
|
||||
#[error("receiver_idx mismatch: connection bound to {established}, packet has {received}")]
|
||||
MismatchedReceiverIndex {
|
||||
established: LpReceiverIndex,
|
||||
received: LpReceiverIndex,
|
||||
},
|
||||
|
||||
#[error("no action has been emitted from the LP State Machine")]
|
||||
UnexpectedStateMachineHalt,
|
||||
|
||||
#[error("the state machine instructed an unexpected action: {action:?}")]
|
||||
UnexpectedStateMachineAction { action: LpAction },
|
||||
|
||||
#[error("received registration request was malformed: {source}")]
|
||||
MalformedRegistrationRequest { source: bincode::Error },
|
||||
|
||||
#[error("received a malformed packet: {0}")]
|
||||
MalformedLpPacket(#[from] MalformedLpPacketError),
|
||||
|
||||
#[error("received payload type of an unexpected type: {typ:?}")]
|
||||
UnexpectedLpPayload { typ: LpDataKind },
|
||||
|
||||
#[error("timed out while attempting to send to/receive from the connection")]
|
||||
ConnectionTimeout,
|
||||
|
||||
#[error("data channel is not yet implemented")]
|
||||
UnimplementedDataChannel,
|
||||
|
||||
#[error("{0}")]
|
||||
Other(String),
|
||||
}
|
||||
|
||||
impl LpHandlerError {
|
||||
pub fn is_connection_closed(&self) -> bool {
|
||||
match self {
|
||||
LpHandlerError::LpTransportError(transport_err) => {
|
||||
matches!(transport_err, LpTransportError::ConnectionClosed)
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn other(msg: impl Into<String>) -> Self {
|
||||
LpHandlerError::Other(msg.into())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,821 @@
|
||||
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: GPL-3.0-only
|
||||
|
||||
use super::{LpHandlerState, LpReceiverIndex, TimestampedState};
|
||||
use crate::node::lp::error::LpHandlerError;
|
||||
use dashmap::mapref::one::RefMut;
|
||||
use nym_lp::packet::{EncryptedLpPacket, ForwardPacketData};
|
||||
use nym_lp::state_machine::{LpAction, LpData, LpDataKind, LpInput};
|
||||
use nym_lp::transport::LpHandshakeChannel;
|
||||
use nym_lp::transport::traits::LpTransportChannel;
|
||||
use nym_lp::{LpSession, LpStateMachine, packet::message::ExpectedResponseSize};
|
||||
use nym_metrics::{add_histogram_obs, inc};
|
||||
use nym_registration_common::{LpRegistrationRequest, RegistrationStatus};
|
||||
use std::net::SocketAddr;
|
||||
use std::time::Duration;
|
||||
use tokio::net::TcpStream;
|
||||
use tokio::time::timeout;
|
||||
use tracing::*;
|
||||
|
||||
// Histogram buckets for LP operation duration (legacy - used by unused forwarding methods)
|
||||
const LP_DURATION_BUCKETS: &[f64] = &[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0];
|
||||
|
||||
// Timeout for forward I/O operations (send + receive on exit stream)
|
||||
// Must be long enough to cover exit gateway processing time
|
||||
const FORWARD_IO_TIMEOUT_SECS: u64 = 30;
|
||||
|
||||
// Histogram buckets for LP connection lifecycle duration
|
||||
// LP connections can be very short (registration only: ~1s) or very long (dVPN sessions: hours/days)
|
||||
// Covers full range from seconds to 24 hours
|
||||
const LP_CONNECTION_DURATION_BUCKETS: &[f64] = &[
|
||||
1.0, // 1 second
|
||||
5.0, // 5 seconds
|
||||
10.0, // 10 seconds
|
||||
30.0, // 30 seconds
|
||||
60.0, // 1 minute
|
||||
300.0, // 5 minutes
|
||||
600.0, // 10 minutes
|
||||
1800.0, // 30 minutes
|
||||
3600.0, // 1 hour
|
||||
7200.0, // 2 hours
|
||||
14400.0, // 4 hours
|
||||
28800.0, // 8 hours
|
||||
43200.0, // 12 hours
|
||||
86400.0, // 24 hours
|
||||
];
|
||||
|
||||
/// Connection lifecycle statistics tracking
|
||||
struct ConnectionStats {
|
||||
/// When the connection started
|
||||
start_time: std::time::Instant,
|
||||
/// Total bytes received (including protocol framing)
|
||||
bytes_received: u64,
|
||||
/// Total bytes sent (including protocol framing)
|
||||
bytes_sent: u64,
|
||||
}
|
||||
|
||||
impl ConnectionStats {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
start_time: std::time::Instant::now(),
|
||||
bytes_received: 0,
|
||||
bytes_sent: 0,
|
||||
}
|
||||
}
|
||||
|
||||
fn record_bytes_received(&mut self, bytes: usize) {
|
||||
self.bytes_received += bytes as u64;
|
||||
}
|
||||
|
||||
fn record_bytes_sent(&mut self, bytes: usize) {
|
||||
self.bytes_sent += bytes as u64;
|
||||
}
|
||||
}
|
||||
|
||||
pub struct LpConnectionHandler<S = TcpStream> {
|
||||
stream: S,
|
||||
remote_addr: SocketAddr,
|
||||
state: LpHandlerState,
|
||||
stats: ConnectionStats,
|
||||
|
||||
// /// Flag indicating whether this is a connection from an entry gateway serving as a proxy
|
||||
// forwarded_connection: bool,
|
||||
/// Bound receiver_idx for this connection (set after first packet).
|
||||
/// All subsequent packets on this connection must use this receiver_idx.
|
||||
/// Set from ClientHello's proposed receiver_index, or from header for non-bootstrap packets.
|
||||
bound_receiver_idx: Option<LpReceiverIndex>,
|
||||
|
||||
/// Persistent connection to exit gateway for forwarding.
|
||||
/// Opened on first forward, reused for subsequent forwards, closed when client disconnects.
|
||||
/// Tuple contains (stream, target_address) to verify subsequent forwards go to same exit.
|
||||
exit_stream: Option<(S, SocketAddr)>,
|
||||
}
|
||||
|
||||
impl<S> LpConnectionHandler<S>
|
||||
where
|
||||
S: LpTransportChannel + LpHandshakeChannel + Unpin,
|
||||
{
|
||||
pub fn new(
|
||||
stream: S,
|
||||
// forwarded_connection: bool,
|
||||
remote_addr: SocketAddr,
|
||||
state: LpHandlerState,
|
||||
) -> Self {
|
||||
Self {
|
||||
stream,
|
||||
remote_addr,
|
||||
// forwarded_connection,
|
||||
state,
|
||||
stats: ConnectionStats::new(),
|
||||
bound_receiver_idx: None,
|
||||
exit_stream: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Get the mutable reference to the state machine associated with this client.
|
||||
/// It is vital it's never held across await points or this might lead to a deadlock.
|
||||
fn state_entry_mut(
|
||||
&self,
|
||||
) -> Result<RefMut<'_, LpReceiverIndex, TimestampedState<LpStateMachine>>, LpHandlerError> {
|
||||
let receiver_index = self.bound_receiver_index()?;
|
||||
self.state
|
||||
.session_states
|
||||
.get_mut(&receiver_index)
|
||||
.ok_or_else(|| LpHandlerError::MissingLpSession { receiver_index })
|
||||
}
|
||||
|
||||
/// AIDEV-NOTE: Stream-oriented packet loop
|
||||
/// This handler processes multiple packets on a single TCP connection.
|
||||
/// Connection lifecycle: handshake + registration, then client closes.
|
||||
/// First packet binds the connection to a receiver_idx (session-affine).
|
||||
/// Binding is set by handle_client_hello() from payload's receiver_index,
|
||||
/// or by validate_or_set_binding() for non-bootstrap first packets.
|
||||
pub async fn handle(mut self) -> Result<(), LpHandlerError> {
|
||||
debug!("Handling LP connection from {}", self.remote_addr);
|
||||
|
||||
// Track total LP connections handled
|
||||
inc!("lp_connections_total");
|
||||
|
||||
// ============================================================
|
||||
// STREAM-ORIENTED PROCESSING: Loop until connection closes
|
||||
// State persists in LpHandlerState maps across packets
|
||||
// ============================================================
|
||||
|
||||
// 1. complete KKT/PSQ handshake before doing anything else.
|
||||
// bail if it takes too long
|
||||
let timeout = self.state.lp_config.debug.handshake_ttl;
|
||||
let local_peer = self.state.local_lp_peer.clone();
|
||||
let stream = &mut self.stream;
|
||||
|
||||
let session = match tokio::time::timeout(timeout, async move {
|
||||
LpSession::psq_handshake_responder(stream, local_peer)
|
||||
.complete_handshake()
|
||||
.await
|
||||
})
|
||||
.await
|
||||
{
|
||||
Err(_timeout) => {
|
||||
debug!(
|
||||
"timed out attempting to complete KTT/PSQ handshake with {}",
|
||||
self.remote_addr
|
||||
);
|
||||
self.emit_lifecycle_metrics(false);
|
||||
return Ok(());
|
||||
}
|
||||
Ok(Err(handshake_failure)) => {
|
||||
debug!(
|
||||
"failed to complete KKT/PSQ handshake with {}: {handshake_failure}",
|
||||
self.remote_addr
|
||||
);
|
||||
self.emit_lifecycle_metrics(false);
|
||||
return Ok(());
|
||||
}
|
||||
Ok(Ok(session)) => session,
|
||||
};
|
||||
let receiver_idx = session.receiver_index();
|
||||
|
||||
// 2. insert the state machine into the shared state
|
||||
let state_machine = LpStateMachine::new(session);
|
||||
self.state
|
||||
.session_states
|
||||
.insert(receiver_idx, TimestampedState::new(state_machine));
|
||||
self.bound_receiver_idx = Some(receiver_idx);
|
||||
|
||||
// 3. handle any new incoming packet
|
||||
loop {
|
||||
// Step 1: Receive raw packet bytes and parse header only (for routing)
|
||||
let encrypted_packet = match self.receive_raw_packet().await {
|
||||
Ok(result) => result,
|
||||
Err(err) => {
|
||||
if err.is_connection_closed() {
|
||||
// Graceful EOF - client closed connection
|
||||
trace!("Connection closed by {} (EOF)", self.remote_addr);
|
||||
break;
|
||||
} else {
|
||||
inc!("lp_errors_receive_packet");
|
||||
self.emit_lifecycle_metrics(false);
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let receiver_idx = encrypted_packet.outer_header().receiver_idx;
|
||||
|
||||
// Step 2: Validate the binding
|
||||
if let Err(e) = self.validate_binding(receiver_idx) {
|
||||
self.emit_lifecycle_metrics(false);
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
// Step 3: Process the packet
|
||||
if let Err(e) = self.process_packet(encrypted_packet).await {
|
||||
self.emit_lifecycle_metrics(false);
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
|
||||
self.emit_lifecycle_metrics(true);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn bound_receiver_index(&self) -> Result<LpReceiverIndex, LpHandlerError> {
|
||||
self.bound_receiver_idx
|
||||
.ok_or_else(|| LpHandlerError::IncompleteHandshake)
|
||||
}
|
||||
|
||||
/// Validate that the receiver_idx matches the bound session.
|
||||
fn validate_binding(&self, receiver_idx: LpReceiverIndex) -> Result<(), LpHandlerError> {
|
||||
let bound_receiver_idx = self.bound_receiver_index()?;
|
||||
|
||||
if bound_receiver_idx != receiver_idx {
|
||||
warn!(
|
||||
"Receiver_idx mismatch from {}: expected {bound_receiver_idx}, got {receiver_idx}",
|
||||
self.remote_addr
|
||||
);
|
||||
inc!("lp_errors_receiver_idx_mismatch");
|
||||
return Err(LpHandlerError::MismatchedReceiverIndex {
|
||||
established: bound_receiver_idx,
|
||||
received: receiver_idx,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Process a single packet: lookup session, parse, route to handler.
|
||||
/// Individual handlers do NOT emit lifecycle metrics - the main loop handles that.
|
||||
///
|
||||
/// This handles packets on established sessions, which can be either:
|
||||
/// EncryptedData containing LpRegistrationRequest or ForwardPacketData
|
||||
///
|
||||
/// We process all transport packets through the state machine.
|
||||
/// The state machine returns appropriate actions:
|
||||
/// - DeliverData: decrypted application data to process
|
||||
/// - SendPacket: response to send
|
||||
async fn process_packet(
|
||||
&mut self,
|
||||
encrypted_packet: EncryptedLpPacket,
|
||||
) -> Result<(), LpHandlerError> {
|
||||
let receiver_index = encrypted_packet.outer_header().receiver_idx;
|
||||
|
||||
let mut state_entry = self.state_entry_mut()?;
|
||||
|
||||
// Update last activity timestamp
|
||||
state_entry.value().touch();
|
||||
|
||||
let state_machine = &mut state_entry.value_mut().state;
|
||||
|
||||
trace!(
|
||||
"Received packet from {} (receiver_idx={receiver_index}, counter={})",
|
||||
self.remote_addr,
|
||||
encrypted_packet.outer_header().counter,
|
||||
);
|
||||
|
||||
// Process packet through state machine
|
||||
let action = state_machine
|
||||
.process_input(LpInput::ReceivePacket(encrypted_packet))
|
||||
.ok_or(LpHandlerError::UnexpectedStateMachineHalt)??;
|
||||
|
||||
drop(state_entry);
|
||||
|
||||
match action {
|
||||
LpAction::SendPacket(response_packet) => {
|
||||
self.send_serialised_packet(&response_packet).await?;
|
||||
Ok(())
|
||||
}
|
||||
LpAction::DeliverData(data) => {
|
||||
// Decrypted application data - process as registration/forwarding
|
||||
self.handle_decrypted_payload(receiver_index, data).await
|
||||
}
|
||||
other @ LpAction::ConnectionClosed => {
|
||||
warn!(
|
||||
"Unexpected action in transport from {}: {other:?}",
|
||||
self.remote_addr
|
||||
);
|
||||
Err(LpHandlerError::UnexpectedStateMachineAction { action: other })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle decrypted transport payload (registration or forwarding request)
|
||||
async fn handle_decrypted_payload(
|
||||
&mut self,
|
||||
receiver_idx: LpReceiverIndex,
|
||||
decrypted_data: LpData,
|
||||
) -> Result<(), LpHandlerError> {
|
||||
let remote = self.remote_addr;
|
||||
|
||||
let bytes = decrypted_data.content;
|
||||
match decrypted_data.kind {
|
||||
LpDataKind::Registration => {
|
||||
let request = LpRegistrationRequest::try_deserialise(&bytes)
|
||||
.map_err(|source| LpHandlerError::MalformedRegistrationRequest { source })?;
|
||||
|
||||
debug!(
|
||||
"LP registration request from {remote} (receiver_idx={receiver_idx}): mode={:?}",
|
||||
request.mode()
|
||||
);
|
||||
|
||||
self.handle_registration_request(receiver_idx, request)
|
||||
.await
|
||||
}
|
||||
LpDataKind::Forward => {
|
||||
let forward_data = ForwardPacketData::decode(&bytes)?;
|
||||
|
||||
self.handle_forwarding_request(receiver_idx, forward_data)
|
||||
.await
|
||||
}
|
||||
typ @ 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(LpHandlerError::UnexpectedLpPayload { typ })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Attempt to wrap and send specified response back to the client
|
||||
async fn send_response_packet(
|
||||
&mut self,
|
||||
serialised_response: Vec<u8>,
|
||||
response_kind: LpDataKind,
|
||||
) -> Result<(), LpHandlerError> {
|
||||
let mut state_entry = self.state_entry_mut()?;
|
||||
|
||||
// Access session via state machine for subsession support
|
||||
let state_machine = &mut state_entry.value_mut().state;
|
||||
|
||||
let wrapped_lp_data = LpData::new(response_kind, serialised_response);
|
||||
|
||||
// Process packet through state machine
|
||||
let action = state_machine
|
||||
.process_input(LpInput::SendData(wrapped_lp_data))
|
||||
.ok_or(LpHandlerError::UnexpectedStateMachineHalt)??;
|
||||
|
||||
let packet = match action {
|
||||
LpAction::SendPacket(packet) => packet,
|
||||
action => return Err(LpHandlerError::UnexpectedStateMachineAction { action }),
|
||||
};
|
||||
|
||||
drop(state_entry);
|
||||
|
||||
self.send_serialised_packet(&packet).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Handle registration request on an established session
|
||||
async fn handle_registration_request(
|
||||
&mut self,
|
||||
receiver_idx: LpReceiverIndex,
|
||||
request: LpRegistrationRequest,
|
||||
) -> Result<(), LpHandlerError> {
|
||||
// Process registration (might modify state)
|
||||
let response = self.state.process_registration(receiver_idx, request).await;
|
||||
let response_bytes = response
|
||||
.serialise()
|
||||
.map_err(|source| LpHandlerError::MalformedRegistrationRequest { source })?;
|
||||
|
||||
self.send_response_packet(response_bytes, LpDataKind::Registration)
|
||||
.await?;
|
||||
|
||||
match response.status {
|
||||
RegistrationStatus::Completed => {
|
||||
info!("LP registration successful for {}", self.remote_addr);
|
||||
}
|
||||
RegistrationStatus::Failed => {
|
||||
warn!(
|
||||
"LP registration failed for {}: {:?}",
|
||||
self.remote_addr,
|
||||
response.error_message()
|
||||
);
|
||||
}
|
||||
RegistrationStatus::PendingMoreData => {
|
||||
info!(
|
||||
"we required more deta from {} to complete registration",
|
||||
self.remote_addr
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Handle forwarding request on an established session
|
||||
///
|
||||
/// Entry gateway receives ForwardPacketData from client, forwards inner packet
|
||||
/// to exit gateway, receives response, encrypts it, and sends back to client.
|
||||
async fn handle_forwarding_request(
|
||||
&mut self,
|
||||
receiver_idx: LpReceiverIndex,
|
||||
forward_data: ForwardPacketData,
|
||||
) -> Result<(), LpHandlerError> {
|
||||
// Forward the packet to the target gateway and retrieve its response
|
||||
let response_bytes = self.handle_forward_packet(forward_data).await?;
|
||||
|
||||
self.send_response_packet(response_bytes, LpDataKind::Forward)
|
||||
.await?;
|
||||
|
||||
debug!(
|
||||
"LP forwarding completed for {} (receiver_idx={receiver_idx})",
|
||||
self.remote_addr
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Returns reference to the established forwarding channel to the exit.
|
||||
#[allow(dead_code)]
|
||||
pub fn forwarding_channel(&self) -> &Option<(S, SocketAddr)> {
|
||||
&self.exit_stream
|
||||
}
|
||||
|
||||
/// This method establishes connection to the target gateway in order to
|
||||
/// forward received packets and retrieve any responses
|
||||
//
|
||||
// In the future it will also perform identity validation to make sure
|
||||
// the target node is a valid gateway present in the network
|
||||
//
|
||||
// Do not manually call this function. It is only exposed for the purposes of integration tests
|
||||
#[doc(hidden)]
|
||||
pub async fn establish_exit_stream(
|
||||
&mut self,
|
||||
target_addr: SocketAddr,
|
||||
) -> Result<(), LpHandlerError> {
|
||||
// Acquire semaphore permit to limit concurrent connection opens (FD exhaustion protection)
|
||||
// Permit is scoped to this block - only protects the connect() call, not stream reuse
|
||||
let _permit = match self.state.forward_semaphore.try_acquire() {
|
||||
Ok(permit) => permit,
|
||||
Err(_) => {
|
||||
inc!("lp_forward_rejected");
|
||||
return Err(LpHandlerError::other("Gateway at forward capacity"));
|
||||
}
|
||||
};
|
||||
|
||||
// Connect to target gateway with timeout
|
||||
let stream = match timeout(Duration::from_secs(5), S::connect(target_addr)).await {
|
||||
Ok(Ok(stream)) => stream,
|
||||
Ok(Err(e)) => {
|
||||
inc!("lp_forward_failed");
|
||||
return Err(LpHandlerError::ConnectionFailure {
|
||||
egress: target_addr,
|
||||
reason: e.to_string(),
|
||||
});
|
||||
}
|
||||
Err(_) => {
|
||||
inc!("lp_forward_failed");
|
||||
return Err(LpHandlerError::ConnectionFailure {
|
||||
egress: target_addr,
|
||||
reason: "target gateway connection timeout".to_string(),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
debug!("Opened persistent exit connection to {target_addr} for forwarding");
|
||||
self.exit_stream = Some((stream, target_addr));
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Forward an LP packet to another gateway
|
||||
///
|
||||
/// This method connects to the target gateway, forwards the inner packet bytes,
|
||||
/// receives the response, and returns it. Used for telescoping (hiding client IP).
|
||||
///
|
||||
/// # Arguments
|
||||
/// * `forward_data` - ForwardPacketData containing target gateway info and inner packet
|
||||
///
|
||||
/// # Returns
|
||||
/// * `Ok(Vec<u8>)` - Raw response bytes from target gateway
|
||||
/// * `Err(LpHandlerError)` - If forwarding fails
|
||||
///
|
||||
/// AIDEV-NOTE: Persistent exit stream forwarding
|
||||
/// Uses self.exit_stream to maintain a persistent connection to the exit gateway.
|
||||
/// First forward opens the connection, subsequent forwards reuse it.
|
||||
/// Connection errors clear exit_stream, causing reconnection on next forward.
|
||||
///
|
||||
/// Semaphore rationale: The forward_semaphore limits concurrent connection OPENS
|
||||
/// (FD exhaustion protection), not concurrent operations. Since:
|
||||
/// 1. Each LpConnectionHandler owns its exit_stream exclusively
|
||||
/// 2. The handler loop processes packets sequentially (no concurrent access)
|
||||
/// 3. Only connection opens consume new FDs
|
||||
///
|
||||
/// The semaphore is only acquired when opening a new connection, not for reuse.
|
||||
async fn handle_forward_packet(
|
||||
&mut self,
|
||||
forward_data: ForwardPacketData,
|
||||
) -> Result<Vec<u8>, LpHandlerError> {
|
||||
inc!("lp_forward_total");
|
||||
let start = std::time::Instant::now();
|
||||
|
||||
// Parse target gateway address
|
||||
let target_addr = forward_data.target_lp_address;
|
||||
|
||||
// Check if we need to open a new connection
|
||||
let need_new_connection = match &self.exit_stream {
|
||||
Some((_, existing_addr)) if *existing_addr == target_addr => false,
|
||||
Some((_, existing_addr)) => {
|
||||
// Target mismatch - this shouldn't happen in normal operation
|
||||
// (client should only forward to one exit gateway)
|
||||
// Return error to prevent silent behavior changes that could mask bugs
|
||||
inc!("lp_forward_failed");
|
||||
return Err(LpHandlerError::other(format!(
|
||||
"Forward target mismatch: session bound to {existing_addr}, got request for {target_addr}"
|
||||
)));
|
||||
}
|
||||
None => true,
|
||||
};
|
||||
|
||||
if need_new_connection {
|
||||
self.establish_exit_stream(target_addr).await?;
|
||||
}
|
||||
|
||||
// Get mutable reference to the exit stream
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let (target_stream, _) = self.exit_stream.as_mut().unwrap();
|
||||
|
||||
debug!(
|
||||
"Forwarding packet to {} ({} bytes)",
|
||||
target_addr,
|
||||
forward_data.inner_packet_bytes.len()
|
||||
);
|
||||
|
||||
// Wrap all I/O in timeout to prevent hanging on unresponsive exit gateway
|
||||
let io_timeout = Duration::from_secs(FORWARD_IO_TIMEOUT_SECS);
|
||||
let inner_bytes = &forward_data.inner_packet_bytes;
|
||||
|
||||
let io_result: Result<Vec<u8>, LpHandlerError> = timeout(io_timeout, async {
|
||||
// Forward inner packet bytes.
|
||||
// it's up to the client to ensure correct formatting,
|
||||
// i.e. relevant headers or length-prefixes
|
||||
target_stream.write_all_and_flush(inner_bytes).await?;
|
||||
|
||||
// attempt to read response based on the provided information
|
||||
|
||||
let response = match forward_data.expected_response_size {
|
||||
ExpectedResponseSize::Handshake(size) => {
|
||||
// client told us exactly how many bytes to expect
|
||||
target_stream.read_n_bytes(size as usize).await?
|
||||
}
|
||||
ExpectedResponseSize::Transport => {
|
||||
// transport packets are length-prefixed
|
||||
target_stream
|
||||
.receive_length_prefixed_transport_bytes()
|
||||
.await?
|
||||
}
|
||||
};
|
||||
Ok(response)
|
||||
})
|
||||
.await
|
||||
.map_err(|_| LpHandlerError::ConnectionTimeout)?;
|
||||
|
||||
// Handle result - clear exit_stream on any error
|
||||
let response_buf = match io_result {
|
||||
Ok(buf) => buf,
|
||||
Err(e) => {
|
||||
inc!("lp_forward_failed");
|
||||
self.exit_stream = None;
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
// Record metrics
|
||||
let duration = start.elapsed().as_secs_f64();
|
||||
add_histogram_obs!("lp_forward_duration_seconds", duration, LP_DURATION_BUCKETS);
|
||||
|
||||
inc!("lp_forward_success");
|
||||
debug!(
|
||||
"Forwarding successful to {} ({} bytes response, {:.3}s)",
|
||||
target_addr,
|
||||
response_buf.len(),
|
||||
duration
|
||||
);
|
||||
|
||||
Ok(response_buf)
|
||||
}
|
||||
|
||||
/// Receive raw packet bytes and parse outer header only (for routing before session lookup).
|
||||
///
|
||||
/// Returns the raw packet bytes and parsed outer header (receiver_idx + counter).
|
||||
/// The caller should look up the session to get outer_aead_key, then call
|
||||
/// `parse_lp_packet()` with the key.
|
||||
async fn receive_raw_packet(&mut self) -> Result<EncryptedLpPacket, LpHandlerError> {
|
||||
let packet = self
|
||||
.stream
|
||||
.receive_length_prefixed_transport_packet()
|
||||
.await?;
|
||||
|
||||
// Track bytes sent (4 byte header + packet data)
|
||||
self.stats
|
||||
.record_bytes_received(4 + packet.encoded_length());
|
||||
|
||||
Ok(packet)
|
||||
}
|
||||
|
||||
/// Send a serialised LP packet over the stream with proper length-prefixed framing.
|
||||
async fn send_serialised_packet(
|
||||
&mut self,
|
||||
packet: &EncryptedLpPacket,
|
||||
) -> Result<(), LpHandlerError> {
|
||||
self.stream
|
||||
.send_length_prefixed_transport_packet(packet)
|
||||
.await?;
|
||||
|
||||
// Track bytes sent (4 byte header + packet data)
|
||||
self.stats.record_bytes_sent(4 + packet.encoded_length());
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Emit connection lifecycle metrics
|
||||
fn emit_lifecycle_metrics(&self, graceful: bool) {
|
||||
use nym_metrics::inc_by;
|
||||
|
||||
// Track connection duration
|
||||
let duration = self.stats.start_time.elapsed().as_secs_f64();
|
||||
add_histogram_obs!(
|
||||
"lp_connection_duration_seconds",
|
||||
duration,
|
||||
LP_CONNECTION_DURATION_BUCKETS
|
||||
);
|
||||
|
||||
// Track bytes transferred
|
||||
inc_by!(
|
||||
"lp_connection_bytes_received_total",
|
||||
self.stats.bytes_received as i64
|
||||
);
|
||||
inc_by!(
|
||||
"lp_connection_bytes_sent_total",
|
||||
self.stats.bytes_sent as i64
|
||||
);
|
||||
|
||||
// Track completion type
|
||||
if graceful {
|
||||
inc!("lp_connections_completed_gracefully");
|
||||
} else {
|
||||
inc!("lp_connections_completed_with_error");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::config::lp::LpDebug;
|
||||
use crate::node::lp::LpConfig;
|
||||
use nym_lp::peer::{KEMKeys, LpLocalPeer, generate_keypair_mceliece, generate_keypair_mlkem};
|
||||
use nym_lp::{Ciphersuite, SessionManager, sessions_for_tests};
|
||||
use nym_test_utils::helpers::{deterministic_rng, deterministic_rng_09};
|
||||
use std::sync::Arc;
|
||||
// ==================== Test Helpers ====================
|
||||
|
||||
/// Create a minimal test state for handler tests
|
||||
async fn create_minimal_test_state() -> LpHandlerState {
|
||||
use nym_crypto::asymmetric::ed25519;
|
||||
|
||||
let mut rng = deterministic_rng();
|
||||
let mut rng09 = deterministic_rng_09();
|
||||
|
||||
let lp_config = LpConfig {
|
||||
debug: LpDebug {
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
};
|
||||
let forward_semaphore = Arc::new(tokio::sync::Semaphore::new(
|
||||
lp_config.debug.max_concurrent_forwards,
|
||||
));
|
||||
|
||||
// Create mix forwarding channel (unused in tests but required by struct)
|
||||
let (mix_sender, _mix_receiver) = nym_mixnet_client::forwarder::mix_forwarding_channels();
|
||||
|
||||
let id_keys = Arc::new(ed25519::KeyPair::new(&mut rng));
|
||||
let x_keys = Arc::new(id_keys.to_x25519().try_into().unwrap());
|
||||
|
||||
let kem_keys = KEMKeys::new(
|
||||
generate_keypair_mceliece(&mut rng09),
|
||||
generate_keypair_mlkem(&mut rng09),
|
||||
);
|
||||
let lp_peer = LpLocalPeer::new(Ciphersuite::default(), x_keys).with_kem_keys(kem_keys);
|
||||
|
||||
LpHandlerState {
|
||||
lp_config,
|
||||
local_lp_peer: lp_peer,
|
||||
metrics: nym_node_metrics::NymNodeMetrics::default(),
|
||||
outbound_mix_sender: mix_sender,
|
||||
session_states: Arc::new(dashmap::DashMap::new()),
|
||||
peer_registrator: None,
|
||||
forward_semaphore,
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Existing Tests ====================
|
||||
|
||||
// ==================== Packet I/O Tests ====================
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_receive_raw_packet_valid() {
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
|
||||
let (init, resp) = sessions_for_tests();
|
||||
let mut init_sm = SessionManager::new();
|
||||
let mut resp_sm = SessionManager::new();
|
||||
resp_sm.create_session_state_machine(resp).unwrap();
|
||||
let id = init_sm.create_session_state_machine(init).unwrap();
|
||||
|
||||
// Bind to localhost
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
|
||||
// Spawn server task
|
||||
let server_task = tokio::spawn(async move {
|
||||
let (stream, remote_addr) = listener.accept().await.unwrap();
|
||||
let state = create_minimal_test_state().await;
|
||||
let mut handler = LpConnectionHandler::new(stream, remote_addr, state);
|
||||
// Two-phase: receive raw bytes + header, then parse full packet
|
||||
let packet = handler.receive_raw_packet().await?;
|
||||
let header = packet.outer_header();
|
||||
assert_eq!(packet.outer_header().receiver_idx, id);
|
||||
let Some(LpAction::DeliverData(data)) = resp_sm.receive_packet(id, packet).unwrap()
|
||||
else {
|
||||
panic!("illegal state")
|
||||
};
|
||||
Ok::<_, LpHandlerError>((header, data))
|
||||
});
|
||||
|
||||
// Connect as client
|
||||
let mut client_stream = TcpStream::connect(addr).await.unwrap();
|
||||
|
||||
// Send a valid packet from client side
|
||||
let LpAction::SendPacket(packet) = init_sm
|
||||
.send_data(id, LpData::new_opaque(b"foomp".to_vec()))
|
||||
.unwrap()
|
||||
else {
|
||||
panic!("illegal state")
|
||||
};
|
||||
|
||||
client_stream
|
||||
.send_length_prefixed_transport_packet(&packet)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Handler should receive and parse it correctly
|
||||
// Note: header is OuterHeader (receiver_idx + counter only), not LpHeader
|
||||
let (header, received) = server_task.await.unwrap().unwrap();
|
||||
assert_eq!(header.receiver_idx, id);
|
||||
assert_eq!(header.counter, 0);
|
||||
assert_eq!(received.content.as_ref(), b"foomp");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_send_lp_packet_valid() {
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
|
||||
let (init, resp) = sessions_for_tests();
|
||||
let mut init_sm = SessionManager::new();
|
||||
let mut resp_sm = SessionManager::new();
|
||||
resp_sm.create_session_state_machine(resp).unwrap();
|
||||
let id = init_sm.create_session_state_machine(init).unwrap();
|
||||
|
||||
let server_task = tokio::spawn(async move {
|
||||
let (mut stream, _) = listener.accept().await.unwrap();
|
||||
|
||||
let LpAction::SendPacket(packet) = resp_sm
|
||||
.send_data(id, LpData::new_opaque(b"foomp".to_vec()))
|
||||
.unwrap()
|
||||
else {
|
||||
panic!("illegal state")
|
||||
};
|
||||
|
||||
stream
|
||||
.send_length_prefixed_transport_packet(&packet)
|
||||
.await
|
||||
.unwrap();
|
||||
});
|
||||
|
||||
let mut client_stream = TcpStream::connect(addr).await.unwrap();
|
||||
|
||||
// Wait for server to send
|
||||
server_task.await.unwrap();
|
||||
|
||||
// Client should receive it correctly
|
||||
let received = client_stream
|
||||
.receive_length_prefixed_transport_packet()
|
||||
.await
|
||||
.unwrap();
|
||||
let header = received.outer_header();
|
||||
let Some(LpAction::DeliverData(data)) = init_sm.receive_packet(id, received).unwrap()
|
||||
else {
|
||||
panic!("illegal state")
|
||||
};
|
||||
|
||||
assert_eq!(header.receiver_idx, id);
|
||||
assert_eq!(header.counter, 0);
|
||||
assert_eq!(data.content.as_ref(), b"foomp");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,470 @@
|
||||
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: GPL-3.0-only
|
||||
|
||||
// LP (Lewes Protocol) Metrics Documentation
|
||||
//
|
||||
// This module implements comprehensive metrics collection for LP operations using nym-metrics macros.
|
||||
// All metrics are automatically prefixed with the package name (nym_gateway) when registered.
|
||||
//
|
||||
// ## Connection Metrics (via NetworkStats in nym-node-metrics)
|
||||
// - active_lp_connections: Gauge tracking current active LP connections (incremented on accept, decremented on close)
|
||||
//
|
||||
// ## Handler Metrics (in handler.rs)
|
||||
// - lp_connections_total: Counter for total LP connections handled
|
||||
// - lp_client_hello_failed: Counter for ClientHello failures (timestamp validation, protocol errors)
|
||||
// - lp_handshakes_success: Counter for successful handshake completions
|
||||
// - lp_handshakes_failed: Counter for failed handshakes
|
||||
// - lp_handshake_duration_seconds: Histogram of handshake durations (buckets: 10ms to 10s)
|
||||
// - lp_timestamp_validation_accepted: Counter for timestamp validations that passed
|
||||
// - lp_timestamp_validation_rejected: Counter for timestamp validations that failed
|
||||
// - lp_errors_handshake: Counter for handshake errors
|
||||
// - lp_errors_send_response: Counter for errors sending registration responses
|
||||
// - lp_errors_timestamp_too_old: Counter for ClientHello timestamps that are too old
|
||||
// - lp_errors_timestamp_too_far_future: Counter for ClientHello timestamps that are too far in the future
|
||||
//
|
||||
// ## Registration Metrics (in registration.rs)
|
||||
// - lp_registration_attempts_total: Counter for all registration attempts
|
||||
// - lp_registration_success_total: Counter for successful registrations (any mode)
|
||||
// - lp_registration_failed_total: Counter for failed registrations (any mode)
|
||||
// - lp_registration_failed_timestamp: Counter for registrations rejected due to invalid timestamp
|
||||
// - lp_registration_duration_seconds: Histogram of registration durations (buckets: 100ms to 30s)
|
||||
//
|
||||
// ## Mode-Specific Registration Metrics (in registration.rs)
|
||||
// - lp_registration_dvpn_attempts: Counter for dVPN mode registration attempts
|
||||
// - lp_registration_dvpn_success: Counter for successful dVPN registrations
|
||||
// - lp_registration_dvpn_failed: Counter for failed dVPN registrations
|
||||
// - lp_registration_mixnet_attempts: Counter for Mixnet mode registration attempts
|
||||
// - lp_registration_mixnet_success: Counter for successful Mixnet registrations
|
||||
// - lp_registration_mixnet_failed: Counter for failed Mixnet registrations
|
||||
//
|
||||
// ## Credential Verification Metrics (in registration.rs)
|
||||
// - lp_credential_verification_attempts: Counter for credential verification attempts
|
||||
// - lp_credential_verification_success: Counter for successful credential verifications
|
||||
// - lp_credential_verification_failed: Counter for failed credential verifications
|
||||
// - lp_bandwidth_allocated_bytes_total: Counter for total bandwidth allocated (in bytes)
|
||||
//
|
||||
// ## Error Categorization Metrics
|
||||
// - lp_errors_wg_peer_registration: Counter for WireGuard peer registration failures
|
||||
//
|
||||
// ## Connection Lifecycle Metrics (in handler.rs)
|
||||
// - lp_connection_duration_seconds: Histogram of connection duration from start to end (buckets: 1s to 24h)
|
||||
// - lp_connection_bytes_received_total: Counter for total bytes received including protocol framing
|
||||
// - lp_connection_bytes_sent_total: Counter for total bytes sent including protocol framing
|
||||
// - lp_connections_completed_gracefully: Counter for connections that completed successfully
|
||||
// - lp_connections_completed_with_error: Counter for connections that terminated with an error
|
||||
//
|
||||
// ## State Cleanup Metrics (in cleanup task)
|
||||
// - lp_states_cleanup_handshake_removed: Counter for stale handshakes removed by cleanup task
|
||||
// - lp_states_cleanup_session_removed: Counter for stale sessions removed by cleanup task
|
||||
// - lp_states_cleanup_demoted_removed: Counter for demoted (read-only) sessions removed by cleanup task
|
||||
//
|
||||
// ## Subsession/Rekeying Metrics (in handler.rs)
|
||||
// - lp_subsession_kk2_sent: Counter for SubsessionKK2 responses sent (indicates client initiated rekeying)
|
||||
// - lp_subsession_complete: Counter for successful subsession promotions
|
||||
// - lp_subsession_receiver_index_collision: Counter for subsession receiver_index collisions
|
||||
//
|
||||
// ## 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::config::lp::LpConfig;
|
||||
use crate::error::NymNodeError;
|
||||
use dashmap::DashMap;
|
||||
use nym_gateway::node::wireguard::PeerRegistrator;
|
||||
use nym_lp::peer::LpLocalPeer;
|
||||
use nym_lp::peer_config::LpReceiverIndex;
|
||||
use nym_lp::state_machine::LpStateMachine;
|
||||
use nym_mixnet_client::forwarder::MixForwardingSender;
|
||||
use nym_node_metrics::NymNodeMetrics;
|
||||
use nym_task::ShutdownTracker;
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::Semaphore;
|
||||
use tracing::*;
|
||||
|
||||
pub use nym_mixnet_client::forwarder::{MixForwardingReceiver, mix_forwarding_channels};
|
||||
|
||||
mod data_handler;
|
||||
pub mod error;
|
||||
pub mod handler;
|
||||
mod registration;
|
||||
|
||||
/// Wrapper for state entries with timestamp tracking for cleanup
|
||||
///
|
||||
/// This wrapper adds `created_at` and `last_activity` timestamps to state entries,
|
||||
/// enabling TTL-based cleanup of stale handshakes and sessions.
|
||||
pub struct TimestampedState<T> {
|
||||
/// The actual state (LpStateMachine or LpSession)
|
||||
pub state: T,
|
||||
|
||||
/// When this state was created (never changes)
|
||||
created_at: std::time::Instant,
|
||||
|
||||
/// Last activity timestamp (unix seconds, atomically updated)
|
||||
///
|
||||
/// For handshakes: never updated (use created_at for TTL)
|
||||
/// For sessions: updated on every packet received
|
||||
last_activity: std::sync::atomic::AtomicU64,
|
||||
}
|
||||
|
||||
impl<T> TimestampedState<T> {
|
||||
/// Create a new timestamped state
|
||||
pub fn new(state: T) -> Self {
|
||||
let now_instant = std::time::Instant::now();
|
||||
let now_unix = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs();
|
||||
|
||||
Self {
|
||||
state,
|
||||
created_at: now_instant,
|
||||
last_activity: std::sync::atomic::AtomicU64::new(now_unix),
|
||||
}
|
||||
}
|
||||
|
||||
/// Update last_activity timestamp (cheap, lock-free operation)
|
||||
pub fn touch(&self) {
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs();
|
||||
self.last_activity
|
||||
.store(now, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Get age since creation
|
||||
#[allow(dead_code)]
|
||||
pub fn age(&self) -> Duration {
|
||||
self.created_at.elapsed()
|
||||
}
|
||||
|
||||
/// Get time since last activity
|
||||
pub fn since_activity(&self) -> Duration {
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs();
|
||||
let last = self
|
||||
.last_activity
|
||||
.load(std::sync::atomic::Ordering::Relaxed);
|
||||
Duration::from_secs(now.saturating_sub(last))
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared state for LP connection handlers
|
||||
#[derive(Clone)]
|
||||
pub struct LpHandlerState {
|
||||
/// Encapsulates all required key information of a local Lewes Protocol Peer.
|
||||
pub local_lp_peer: LpLocalPeer,
|
||||
|
||||
/// Metrics collection
|
||||
pub metrics: NymNodeMetrics,
|
||||
|
||||
/// Handle registering new wireguard peers
|
||||
pub peer_registrator: Option<PeerRegistrator>,
|
||||
|
||||
/// LP configuration (for timestamp validation, etc.)
|
||||
pub lp_config: LpConfig,
|
||||
|
||||
/// Channel for forwarding Sphinx packets into the mixnet
|
||||
///
|
||||
/// Used by the LP data handler (UDP:51264) to forward decrypted Sphinx packets
|
||||
/// from LP clients into the mixnet for routing.
|
||||
#[allow(dead_code)]
|
||||
pub outbound_mix_sender: MixForwardingSender,
|
||||
|
||||
/// Established sessions keyed by session_id
|
||||
///
|
||||
/// Used after handshake completes (session_id is deterministically computed from
|
||||
/// both parties' X25519 keys). Enables stateless transport - each packet lookup
|
||||
/// by session_id, decrypt/process, respond.
|
||||
///
|
||||
/// Wrapped in TimestampedState for TTL-based cleanup of inactive sessions.
|
||||
///
|
||||
/// Sessions are stored as LpStateMachine (not LpSession) to enable
|
||||
/// subsession/rekeying support. The state machine handles subsession initiation
|
||||
/// (SubsessionKK1/KK2/Ready) during transport phase, allowing long-lived connections
|
||||
/// to rekey without re-authentication.
|
||||
pub session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
|
||||
|
||||
/// Semaphore limiting concurrent forward connections
|
||||
///
|
||||
/// Prevents file descriptor exhaustion when forwarding LP packets during
|
||||
/// telescope setup. When at capacity, forward requests return an error
|
||||
/// so clients can choose a different gateway.
|
||||
// Connection limiting (not pooling) chosen for forward requests.
|
||||
//
|
||||
// Why not connection pooling?
|
||||
// 1. Forwarding is one-time per telescope setup (handshake only), not ongoing traffic.
|
||||
// Once telescope is established, data flows directly through the tunnel.
|
||||
// 2. Telescope targets are distributed across many different gateways - each client
|
||||
// typically connects to a different exit gateway, so pooled connections would
|
||||
// rarely be reused.
|
||||
// 3. Connections already go out of scope after each request-response. FD exhaustion
|
||||
// only happens from concurrent spikes, not accumulation.
|
||||
// 4. A pool would accumulate one idle connection per unique destination, most of
|
||||
// which would never be reused before TTL expiration.
|
||||
//
|
||||
// Why semaphore limiting is better:
|
||||
// 1. Directly caps concurrent forward connections regardless of destination.
|
||||
// 2. When at capacity, returns "busy" error - client can choose another gateway.
|
||||
// This is better than silently queuing requests behind a pool.
|
||||
// 3. Simple implementation: no TTL management, stale connection handling, or cleanup.
|
||||
pub forward_semaphore: Arc<Semaphore>,
|
||||
}
|
||||
|
||||
/// LP listener that accepts TCP connections on port 41264
|
||||
pub struct LpListener {
|
||||
/// Shared state for connection handlers
|
||||
handler_state: LpHandlerState,
|
||||
|
||||
/// Shutdown coordination
|
||||
shutdown: ShutdownTracker,
|
||||
}
|
||||
|
||||
impl LpListener {
|
||||
pub fn new(handler_state: LpHandlerState, shutdown: ShutdownTracker) -> Self {
|
||||
Self {
|
||||
handler_state,
|
||||
shutdown,
|
||||
}
|
||||
}
|
||||
|
||||
fn lp_config(&self) -> LpConfig {
|
||||
self.handler_state.lp_config
|
||||
}
|
||||
|
||||
pub async fn run(&mut self) -> Result<(), NymNodeError> {
|
||||
let control_bind_address = self.lp_config().control_bind_address;
|
||||
let data_bind_address = self.lp_config().data_bind_address;
|
||||
let listener = TcpListener::bind(control_bind_address)
|
||||
.await
|
||||
.map_err(|source| {
|
||||
error!("Failed to bind LP listener to {control_bind_address}: {source}",);
|
||||
NymNodeError::LpBindFailure {
|
||||
address: control_bind_address,
|
||||
source,
|
||||
}
|
||||
})?;
|
||||
|
||||
let shutdown_token = self.shutdown.clone_shutdown_token();
|
||||
|
||||
// Spawn background task for state cleanup
|
||||
let _cleanup_handle = self.spawn_state_cleanup_task();
|
||||
|
||||
// Spawn UDP data handler for LP data plane (port 51264)
|
||||
let _data_handler_handle = self.spawn_data_handler().await?;
|
||||
|
||||
info!(
|
||||
"LP listener started on {control_bind_address} (data handler on: {data_bind_address})",
|
||||
);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = shutdown_token.cancelled() => {
|
||||
trace!("LP listener: received shutdown signal");
|
||||
break;
|
||||
}
|
||||
|
||||
result = listener.accept() => {
|
||||
match result {
|
||||
Ok((stream, addr)) => self.handle_connection(stream, addr),
|
||||
Err(e) => warn!("Failed to accept LP connection: {e}")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
info!("LP listener shutdown complete");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn handle_connection(&self, stream: tokio::net::TcpStream, remote_addr: SocketAddr) {
|
||||
// Check connection limit
|
||||
let active_connections = self.active_lp_connections();
|
||||
let max_connections = self.lp_config().debug.max_connections;
|
||||
if active_connections >= max_connections {
|
||||
warn!(
|
||||
"LP connection limit exceeded ({active_connections}/{max_connections}), rejecting connection from {remote_addr}"
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
debug!(
|
||||
"Accepting LP connection from {remote_addr} ({active_connections} 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 metrics = self.handler_state.metrics.clone();
|
||||
self.shutdown.try_spawn_named_with_shutdown(
|
||||
async move {
|
||||
let result = handler.handle().await;
|
||||
|
||||
// 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}");
|
||||
// Note: metrics are emitted in handle() for graceful path
|
||||
// On error path, handle() returns early without emitting
|
||||
// So we track errors here
|
||||
}
|
||||
|
||||
// Decrement connection counter on exit
|
||||
metrics.network.lp_connection_closed();
|
||||
},
|
||||
&format!("LP::{remote_addr}"),
|
||||
);
|
||||
}
|
||||
|
||||
/// Spawn the UDP data handler for LP data plane
|
||||
///
|
||||
/// The data handler listens on UDP port 51264 and processes LP-wrapped Sphinx packets
|
||||
/// from registered clients. It decrypts the LP layer and forwards the Sphinx packets
|
||||
/// into the mixnet.
|
||||
async fn spawn_data_handler(&self) -> Result<tokio::task::JoinHandle<()>, NymNodeError> {
|
||||
// Create data handler
|
||||
let data_handler = data_handler::LpDataHandler::new(
|
||||
self.lp_config().data_bind_address,
|
||||
self.handler_state.clone(),
|
||||
self.shutdown.clone_shutdown_token(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
// Spawn data handler task
|
||||
let handle = self.shutdown.try_spawn_named(
|
||||
async move {
|
||||
if let Err(e) = data_handler.run().await {
|
||||
error!("LP data handler error: {e}");
|
||||
}
|
||||
},
|
||||
"LP::DataHandler",
|
||||
);
|
||||
|
||||
Ok(handle)
|
||||
}
|
||||
|
||||
/// Spawn background task for cleaning up stale state entries
|
||||
///
|
||||
/// This task runs periodically (every `state_cleanup_interval_secs`) to remove:
|
||||
/// - Handshake states older than `handshake_ttl_secs`
|
||||
/// - Session states with no activity for `session_ttl_secs`
|
||||
///
|
||||
/// The task automatically stops when the shutdown signal is received.
|
||||
fn spawn_state_cleanup_task(&self) -> tokio::task::JoinHandle<()> {
|
||||
let session_states = Arc::clone(&self.handler_state.session_states);
|
||||
let dbg_cfg = self.handler_state.lp_config.debug;
|
||||
|
||||
let handshake_ttl = dbg_cfg.handshake_ttl;
|
||||
let session_ttl = dbg_cfg.session_ttl;
|
||||
let interval = dbg_cfg.state_cleanup_interval;
|
||||
let shutdown = self.shutdown.clone_shutdown_token();
|
||||
let metrics = self.handler_state.metrics.clone();
|
||||
|
||||
info!(
|
||||
"Starting LP state cleanup task (handshake_ttl={}s, session_ttl={}s, interval={}s)",
|
||||
handshake_ttl.as_secs(),
|
||||
session_ttl.as_secs(),
|
||||
interval.as_secs()
|
||||
);
|
||||
|
||||
self.shutdown.try_spawn_named(
|
||||
cleanup_task::cleanup_loop(session_states, dbg_cfg, shutdown, metrics),
|
||||
"LP::StateCleanup",
|
||||
)
|
||||
}
|
||||
|
||||
fn active_lp_connections(&self) -> usize {
|
||||
self.handler_state
|
||||
.metrics
|
||||
.network
|
||||
.active_lp_connections_count()
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) mod cleanup_task {
|
||||
use crate::config::lp::LpDebug;
|
||||
use crate::node::lp::{LpReceiverIndex, TimestampedState};
|
||||
use dashmap::DashMap;
|
||||
use nym_lp::LpStateMachine;
|
||||
use nym_metrics::inc_by;
|
||||
use nym_node_metrics::NymNodeMetrics;
|
||||
use std::sync::Arc;
|
||||
use tracing::{debug, info};
|
||||
|
||||
async fn perform_cleanup(
|
||||
session_states: &Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
|
||||
cfg: LpDebug,
|
||||
) {
|
||||
let session_ttl = cfg.session_ttl;
|
||||
|
||||
let start = std::time::Instant::now();
|
||||
let mut ss_removed = 0u64;
|
||||
|
||||
// Remove stale sessions (based on time since last activity)
|
||||
// Use shorter TTL for demoted (ReadOnlyTransport) sessions
|
||||
session_states.retain(|_, timestamped| {
|
||||
if timestamped.since_activity() > session_ttl {
|
||||
ss_removed += 1;
|
||||
false
|
||||
} else {
|
||||
true
|
||||
}
|
||||
});
|
||||
|
||||
if ss_removed > 0 {
|
||||
let duration = start.elapsed();
|
||||
info!(
|
||||
"LP state cleanup: {ss_removed} sessions (took {:.3}s)",
|
||||
duration.as_secs_f64()
|
||||
);
|
||||
|
||||
// Track metrics
|
||||
if ss_removed > 0 {
|
||||
inc_by!("lp_states_cleanup_session_removed", ss_removed as i64);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Background loop for cleaning up stale state entries
|
||||
///
|
||||
/// Runs periodically to scan handshake_states and session_states maps,
|
||||
/// removing entries that have exceeded their TTL.
|
||||
///
|
||||
/// Demoted sessions (ReadOnlyTransport) use shorter TTL since they
|
||||
/// only need to drain in-flight packets after subsession promotion.
|
||||
pub(crate) async fn cleanup_loop(
|
||||
session_states: Arc<DashMap<LpReceiverIndex, TimestampedState<LpStateMachine>>>,
|
||||
cfg: LpDebug,
|
||||
shutdown: nym_task::ShutdownToken,
|
||||
_metrics: NymNodeMetrics,
|
||||
) {
|
||||
let interval = cfg.state_cleanup_interval;
|
||||
|
||||
let mut cleanup_interval = tokio::time::interval(interval);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = shutdown.cancelled() => {
|
||||
debug!("LP state cleanup task: received shutdown signal");
|
||||
break;
|
||||
}
|
||||
_ = cleanup_interval.tick() => {
|
||||
perform_cleanup(&session_states, cfg).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
info!("LP state cleanup task shutdown complete");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: GPL-3.0-only
|
||||
|
||||
use crate::node::lp::{LpHandlerState, LpReceiverIndex};
|
||||
use nym_metrics::{add_histogram_obs, inc};
|
||||
use nym_registration_common::dvpn::{
|
||||
LpDvpnRegistrationFinalisation, LpDvpnRegistrationInitialRequest,
|
||||
LpDvpnRegistrationRequestMessage, LpDvpnRegistrationRequestMessageContent,
|
||||
};
|
||||
use nym_registration_common::mixnet::LpMixnetRegistrationRequestMessage;
|
||||
use nym_registration_common::{
|
||||
LpRegistrationRequest, LpRegistrationRequestData, LpRegistrationResponse, RegistrationMode,
|
||||
RegistrationStatus,
|
||||
};
|
||||
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
|
||||
];
|
||||
|
||||
impl LpHandlerState {
|
||||
async fn process_dvpn_initial_registration(
|
||||
&self,
|
||||
sender: LpReceiverIndex,
|
||||
request: LpDvpnRegistrationInitialRequest,
|
||||
) -> LpRegistrationResponse {
|
||||
let Some(registrator) = self.peer_registrator.as_ref() else {
|
||||
return LpRegistrationResponse::error(
|
||||
"dVPN via LP is not enabled on this node",
|
||||
RegistrationMode::Dvpn,
|
||||
);
|
||||
};
|
||||
|
||||
registrator
|
||||
.on_initial_lp_request(request, sender)
|
||||
.await
|
||||
.unwrap_or_else(|err| {
|
||||
LpRegistrationResponse::error(
|
||||
format!("LP registration has failed: {err}"),
|
||||
RegistrationMode::Dvpn,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
async fn process_dvpn_registration_finalisation(
|
||||
&self,
|
||||
sender: LpReceiverIndex,
|
||||
request: LpDvpnRegistrationFinalisation,
|
||||
) -> LpRegistrationResponse {
|
||||
let Some(registrator) = self.peer_registrator.as_ref() else {
|
||||
return LpRegistrationResponse::error(
|
||||
"dVPN via LP is not enabled on this node",
|
||||
RegistrationMode::Dvpn,
|
||||
);
|
||||
};
|
||||
|
||||
registrator
|
||||
.on_final_lp_request(request, sender)
|
||||
.await
|
||||
.unwrap_or_else(|err| {
|
||||
LpRegistrationResponse::error(
|
||||
format!("LP registration has failed: {err}"),
|
||||
RegistrationMode::Dvpn,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
async fn process_dvpn_registration(
|
||||
&self,
|
||||
sender: LpReceiverIndex,
|
||||
request: Box<LpDvpnRegistrationRequestMessage>,
|
||||
) -> LpRegistrationResponse {
|
||||
// Track dVPN registration attempts
|
||||
inc!("lp_registration_dvpn_attempts");
|
||||
|
||||
match request.content {
|
||||
LpDvpnRegistrationRequestMessageContent::InitialRequest(req) => {
|
||||
self.process_dvpn_initial_registration(sender, req).await
|
||||
}
|
||||
LpDvpnRegistrationRequestMessageContent::Finalisation(req) => {
|
||||
self.process_dvpn_registration_finalisation(sender, req)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn process_mixnet_registration(
|
||||
&self,
|
||||
request: LpMixnetRegistrationRequestMessage,
|
||||
) -> LpRegistrationResponse {
|
||||
let _ = request;
|
||||
LpRegistrationResponse::error(
|
||||
"mixnet registration is not yet supported",
|
||||
RegistrationMode::Mixnet,
|
||||
)
|
||||
}
|
||||
|
||||
/// Process an LP registration request
|
||||
pub async fn process_registration(
|
||||
&self,
|
||||
sender: LpReceiverIndex,
|
||||
request: LpRegistrationRequest,
|
||||
) -> LpRegistrationResponse {
|
||||
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");
|
||||
inc!("lp_registration_failed_timestamp");
|
||||
return LpRegistrationResponse::error("invalid timestamp", request.mode());
|
||||
}
|
||||
|
||||
// 2. Process based on mode
|
||||
let result = match request.registration_data {
|
||||
LpRegistrationRequestData::Dvpn { data } => {
|
||||
self.process_dvpn_registration(sender, data).await
|
||||
}
|
||||
LpRegistrationRequestData::Mixnet { data } => {
|
||||
self.process_mixnet_registration(data).await
|
||||
}
|
||||
};
|
||||
|
||||
// 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
|
||||
match result.status {
|
||||
RegistrationStatus::Completed => {
|
||||
inc!("lp_registration_success_total");
|
||||
}
|
||||
RegistrationStatus::Failed => {
|
||||
inc!("lp_registration_failed_total");
|
||||
}
|
||||
RegistrationStatus::PendingMoreData => {
|
||||
inc!("lp_registration_pending_more_data");
|
||||
}
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
}
|
||||
+120
-63
@@ -10,7 +10,9 @@ use crate::error::{EntryGatewayError, NymNodeError, ServiceProvidersError};
|
||||
use crate::node::description::{load_node_description, save_node_description};
|
||||
use crate::node::helpers::{
|
||||
DisplayDetails, get_current_rotation_id, load_ed25519_identity_keypair, load_key,
|
||||
load_mceliece_keypair, load_mlkem768_keypair, load_x25519_lp_keypair,
|
||||
load_x25519_noise_keypair, store_ed25519_identity_keypair, store_key, store_keypair,
|
||||
store_mceliece_keypair, store_mlkem768_keypair, store_x25519_lp_keypair,
|
||||
store_x25519_noise_keypair,
|
||||
};
|
||||
use crate::node::http::api::api_requests;
|
||||
@@ -20,6 +22,7 @@ use crate::node::http::{HttpServerConfig, NymNodeHttpServer, NymNodeRouter};
|
||||
use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
|
||||
use crate::node::key_rotation::controller::KeyRotationController;
|
||||
use crate::node::key_rotation::manager::SphinxKeyManager;
|
||||
use crate::node::lp::{LpHandlerState, LpListener};
|
||||
use crate::node::metrics::aggregator::MetricsAggregator;
|
||||
use crate::node::metrics::console_logger::ConsoleLogger;
|
||||
use crate::node::metrics::handler::client_sessions::GatewaySessionStatsHandler;
|
||||
@@ -41,7 +44,14 @@ use crate::node::shared_network::{
|
||||
use nym_bin_common::bin_info;
|
||||
use nym_credential_verification::UpgradeModeState;
|
||||
use nym_crypto::asymmetric::{ed25519, x25519};
|
||||
use nym_gateway::node::{ActiveClientsStore, GatewayTasksBuilder, UpgradeModeCheckRequestSender};
|
||||
use nym_gateway::node::wireguard::PeerRegistrator;
|
||||
use nym_gateway::node::{GatewayTasksBuilder, UpgradeModeCheckRequestSender};
|
||||
use nym_kkt::key_utils::{
|
||||
generate_keypair_mceliece, generate_keypair_mlkem, generate_lp_keypair_x25519,
|
||||
};
|
||||
use nym_kkt::keys::{DHKeyPair, KEMKeys};
|
||||
use nym_lp::Ciphersuite;
|
||||
use nym_lp::peer::LpLocalPeer;
|
||||
use nym_mixnet_client::client::ActiveConnections;
|
||||
use nym_mixnet_client::forwarder::MixForwardingSender;
|
||||
use nym_network_requester::{
|
||||
@@ -50,6 +60,8 @@ use nym_network_requester::{
|
||||
};
|
||||
use nym_node_metrics::NymNodeMetrics;
|
||||
use nym_node_metrics::events::MetricEventsSender;
|
||||
use nym_node_requests::api::SignedData;
|
||||
use nym_node_requests::api::v1::lewes_protocol::models::{LPHashFunction, LPKEM, LewesProtocol};
|
||||
use nym_node_requests::api::v1::node::models::{AnnouncePorts, NodeDescription};
|
||||
use nym_noise::config::{NoiseConfig, NoiseNetworkView};
|
||||
use nym_noise_keys::VersionedNoiseKeyV1;
|
||||
@@ -62,19 +74,25 @@ use nym_verloc::{self, measurements::VerlocMeasurer};
|
||||
use nym_wireguard::{WireguardGatewayData, peer_controller::PeerControlRequest};
|
||||
use rand::rngs::OsRng;
|
||||
use rand::{CryptoRng, RngCore};
|
||||
use rand09::SeedableRng;
|
||||
use std::collections::BTreeMap;
|
||||
use std::net::SocketAddr;
|
||||
use std::ops::Deref;
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::sync::{Semaphore, mpsc};
|
||||
use tracing::{debug, info, trace};
|
||||
use zeroize::Zeroizing;
|
||||
|
||||
pub use nym_gateway::node::ActiveClientsStore;
|
||||
pub use nym_gateway::node::GatewayStorage;
|
||||
|
||||
pub mod bonding_information;
|
||||
pub mod description;
|
||||
pub mod helpers;
|
||||
pub(crate) mod http;
|
||||
pub(crate) mod key_rotation;
|
||||
pub mod lp;
|
||||
pub(crate) mod metrics;
|
||||
pub(crate) mod mixnet;
|
||||
mod nym_apis_client;
|
||||
@@ -84,8 +102,7 @@ mod shared_network;
|
||||
|
||||
pub struct GatewayTasksData {
|
||||
mnemonic: Arc<Zeroizing<bip39::Mnemonic>>,
|
||||
psq_kem_key: Arc<x25519::KeyPair>,
|
||||
client_storage: nym_gateway::node::GatewayStorage,
|
||||
client_storage: GatewayStorage,
|
||||
stats_storage: nym_gateway::node::PersistentStatsStorage,
|
||||
}
|
||||
|
||||
@@ -106,12 +123,8 @@ impl GatewayTasksData {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn new(
|
||||
config: &GatewayTasksConfig,
|
||||
// this argument is temporary while we still derive KEM x25519 out of identity ed25519
|
||||
ed25519_identity: &ed25519::KeyPair,
|
||||
) -> Result<GatewayTasksData, EntryGatewayError> {
|
||||
let client_storage = nym_gateway::node::GatewayStorage::init(
|
||||
async fn new(config: &GatewayTasksConfig) -> Result<GatewayTasksData, EntryGatewayError> {
|
||||
let client_storage = GatewayStorage::init(
|
||||
&config.storage_paths.clients_storage,
|
||||
config.debug.message_retrieval_limit,
|
||||
)
|
||||
@@ -125,7 +138,6 @@ impl GatewayTasksData {
|
||||
|
||||
Ok(GatewayTasksData {
|
||||
mnemonic: Arc::new(config.storage_paths.load_mnemonic_from_file()?),
|
||||
psq_kem_key: Arc::new(ed25519_identity.to_x25519()),
|
||||
client_storage,
|
||||
stats_storage,
|
||||
})
|
||||
@@ -373,7 +385,7 @@ impl From<WireguardData> for nym_wireguard::WireguardData {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct NymNode {
|
||||
pub struct NymNode {
|
||||
config: Config,
|
||||
accepted_operator_terms_and_conditions: bool,
|
||||
shutdown_manager: ShutdownManager,
|
||||
@@ -396,8 +408,10 @@ pub(crate) struct NymNode {
|
||||
ed25519_identity_keys: Arc<ed25519::KeyPair>,
|
||||
sphinx_key_manager: Option<SphinxKeyManager>,
|
||||
|
||||
// to be used when noise is integrated
|
||||
x25519_noise_keys: Arc<x25519::KeyPair>,
|
||||
|
||||
psq_kem_keys: KEMKeys,
|
||||
x25519_lp_keys: Arc<DHKeyPair>,
|
||||
}
|
||||
|
||||
impl NymNode {
|
||||
@@ -405,12 +419,19 @@ impl NymNode {
|
||||
config: &Config,
|
||||
custom_mnemonic: Option<Zeroizing<bip39::Mnemonic>>,
|
||||
) -> Result<(), NymNodeError> {
|
||||
debug!("initialising nym-node with id: {}", config.id);
|
||||
info!("initialising nym-node with id: {}", config.id);
|
||||
let mut rng = OsRng;
|
||||
let mut rng09 = rand09::rngs::StdRng::from_os_rng();
|
||||
|
||||
// global initialisation
|
||||
info!("generating new node keys (this might take a while)");
|
||||
let ed25519_identity_keys = ed25519::KeyPair::new(&mut rng);
|
||||
let x25519_noise_keys = x25519::KeyPair::new(&mut rng);
|
||||
|
||||
let x25519_lp_keys = generate_lp_keypair_x25519(&mut rng09);
|
||||
let mlkem = generate_keypair_mlkem(&mut rng09);
|
||||
let mceliece = generate_keypair_mceliece(&mut rng09);
|
||||
|
||||
let current_rotation_id =
|
||||
get_current_rotation_id(&config.mixnet.nym_api_urls, &config.mixnet.nyxd_urls).await?;
|
||||
let _ = SphinxKeyManager::initialise_new(
|
||||
@@ -432,6 +453,18 @@ impl NymNode {
|
||||
&config.storage_paths.keys.x25519_noise_storage_paths(),
|
||||
)?;
|
||||
|
||||
trace!("attempting to x25519 lp keypair");
|
||||
store_x25519_lp_keypair(
|
||||
&x25519_lp_keys,
|
||||
&config.storage_paths.keys.x25519_lp_key_paths(),
|
||||
)?;
|
||||
|
||||
trace!("attempting to mlkem768 keypair");
|
||||
store_mlkem768_keypair(&mlkem, &config.storage_paths.keys.mlkem768_key_paths())?;
|
||||
|
||||
trace!("attempting to mceliece keypair");
|
||||
store_mceliece_keypair(&mceliece, &config.storage_paths.keys.mceliece_key_paths())?;
|
||||
|
||||
trace!("creating description file");
|
||||
save_node_description(
|
||||
&config.storage_paths.description,
|
||||
@@ -454,6 +487,30 @@ impl NymNode {
|
||||
config.save()
|
||||
}
|
||||
|
||||
pub async fn build_lp_listener(
|
||||
&mut self,
|
||||
peer_registrator: Option<PeerRegistrator>,
|
||||
mix_packet_sender: MixForwardingSender,
|
||||
) -> Result<LpListener, NymNodeError> {
|
||||
let handler_state = LpHandlerState {
|
||||
local_lp_peer: LpLocalPeer::new(Ciphersuite::default(), self.x25519_lp_keys.clone())
|
||||
.with_kem_keys(self.psq_kem_keys.clone()),
|
||||
metrics: self.metrics.clone(),
|
||||
peer_registrator,
|
||||
lp_config: self.config.lp,
|
||||
outbound_mix_sender: mix_packet_sender,
|
||||
session_states: Arc::new(dashmap::DashMap::new()),
|
||||
forward_semaphore: Arc::new(Semaphore::new(
|
||||
self.config.lp.debug.max_concurrent_forwards,
|
||||
)),
|
||||
};
|
||||
|
||||
Ok(LpListener::new(
|
||||
handler_state,
|
||||
self.shutdown_manager.shutdown_tracker().clone(),
|
||||
))
|
||||
}
|
||||
|
||||
pub(crate) async fn new(config: Config) -> Result<Self, NymNodeError> {
|
||||
let wireguard_data = WireguardData::new(&config.wireguard)?;
|
||||
let current_rotation_id =
|
||||
@@ -462,8 +519,12 @@ impl NymNode {
|
||||
let ed25519_identity_keys = load_ed25519_identity_keypair(
|
||||
&config.storage_paths.keys.ed25519_identity_storage_paths(),
|
||||
)?;
|
||||
let entry_gateway =
|
||||
GatewayTasksData::new(&config.gateway_tasks, &ed25519_identity_keys).await?;
|
||||
let entry_gateway = GatewayTasksData::new(&config.gateway_tasks).await?;
|
||||
let x25519_lp_keys =
|
||||
load_x25519_lp_keypair(&config.storage_paths.keys.x25519_lp_key_paths())?;
|
||||
let mlkem = load_mlkem768_keypair(&config.storage_paths.keys.mlkem768_key_paths())?;
|
||||
let mceliece = load_mceliece_keypair(&config.storage_paths.keys.mceliece_key_paths())?;
|
||||
let psq_kem_keys = KEMKeys::new(mceliece, mlkem);
|
||||
|
||||
Ok(NymNode {
|
||||
ed25519_identity_keys: Arc::new(ed25519_identity_keys),
|
||||
@@ -475,6 +536,7 @@ impl NymNode {
|
||||
x25519_noise_keys: Arc::new(load_x25519_noise_keypair(
|
||||
&config.storage_paths.keys.x25519_noise_storage_paths(),
|
||||
)?),
|
||||
psq_kem_keys,
|
||||
description: load_node_description(&config.storage_paths.description)?,
|
||||
metrics: NymNodeMetrics::new(),
|
||||
verloc_stats: Default::default(),
|
||||
@@ -488,6 +550,7 @@ impl NymNode {
|
||||
accepted_operator_terms_and_conditions: false,
|
||||
shutdown_manager: ShutdownManager::build_new_default()
|
||||
.map_err(|source| NymNodeError::ShutdownSignalFailure { source })?,
|
||||
x25519_lp_keys: Arc::new(x25519_lp_keys),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -640,15 +703,14 @@ impl NymNode {
|
||||
let mut gateway_tasks_builder = GatewayTasksBuilder::new(
|
||||
config.gateway,
|
||||
self.ed25519_identity_keys.clone(),
|
||||
self.x25519_noise_keys.clone(),
|
||||
self.entry_gateway.psq_kem_key.clone(),
|
||||
self.entry_gateway.client_storage.clone(),
|
||||
mix_packet_sender,
|
||||
mix_packet_sender.clone(),
|
||||
metrics_sender,
|
||||
self.metrics.clone(),
|
||||
self.entry_gateway.mnemonic.clone(),
|
||||
Self::user_agent(),
|
||||
self.upgrade_mode_state.clone(),
|
||||
self.config.lp.debug.use_mock_ecash,
|
||||
self.shutdown_tracker().clone(),
|
||||
);
|
||||
|
||||
@@ -706,16 +768,13 @@ impl NymNode {
|
||||
// Start LP listener if enabled
|
||||
info!(
|
||||
"starting the LP listener on {} (data handler on: {})",
|
||||
self.config.gateway_tasks.lp.control_bind_address,
|
||||
self.config.gateway_tasks.lp.data_bind_address,
|
||||
self.config.lp.control_bind_address, self.config.lp.data_bind_address,
|
||||
);
|
||||
let lp_listener = gateway_tasks_builder
|
||||
.build_lp_listener(wg_peer_registrator.clone(), active_clients_store.clone())
|
||||
let mut lp_listener = self
|
||||
.build_lp_listener(wg_peer_registrator.clone(), mix_packet_sender)
|
||||
.await?;
|
||||
// make sure lp listener can be built, but do not start it
|
||||
let _ = lp_listener;
|
||||
// self.shutdown_tracker()
|
||||
// .try_spawn_named(async move { lp_listener.run().await }, "LpListener");
|
||||
self.shutdown_tracker()
|
||||
.try_spawn_named(async move { lp_listener.run().await }, "LpListener");
|
||||
} else {
|
||||
info!("node not running in entry mode: the websocket and LP will remain closed");
|
||||
}
|
||||
@@ -791,40 +850,24 @@ impl NymNode {
|
||||
|
||||
Ok(())
|
||||
}
|
||||
//
|
||||
// fn compute_kem_key_hashes(&self) -> HashMap<LPKEM, HashMap<LPHashFunction, String>> {
|
||||
// let kem = LPKEM::X25519;
|
||||
//
|
||||
// let kem_key_bytes = self.entry_gateway.psq_kem_key.public_key().as_bytes();
|
||||
//
|
||||
// // convert from `nym_kkt_ciphersuite` types into `nym_nodes_requests`
|
||||
// let digests = produce_key_digests(kem_key_bytes.as_ref())
|
||||
// .into_iter()
|
||||
// .map(|(f, digest)| (f.into(), hex::encode(&digest)))
|
||||
// .collect();
|
||||
//
|
||||
// let mut hashes = HashMap::new();
|
||||
// hashes.insert(kem, digests);
|
||||
// hashes
|
||||
// }
|
||||
//
|
||||
// fn compute_signing_key_hashes(
|
||||
// &self,
|
||||
// ) -> HashMap<LPSignatureScheme, HashMap<LPHashFunction, String>> {
|
||||
// let scheme = LPSignatureScheme::Ed25519;
|
||||
//
|
||||
// let kem_key_bytes = self.ed25519_identity_keys.public_key().as_bytes();
|
||||
//
|
||||
// // convert from `nym_kkt_ciphersuite` types into `nym_nodes_requests`
|
||||
// let digests = produce_key_digests(kem_key_bytes.as_ref())
|
||||
// .into_iter()
|
||||
// .map(|(f, digest)| (f.into(), hex::encode(&digest)))
|
||||
// .collect();
|
||||
//
|
||||
// let mut hashes = HashMap::new();
|
||||
// hashes.insert(scheme, digests);
|
||||
// hashes
|
||||
// }
|
||||
|
||||
fn compute_kem_key_hashes(&self) -> BTreeMap<LPKEM, BTreeMap<LPHashFunction, String>> {
|
||||
let digests = self.psq_kem_keys.encapsulation_keys_digests();
|
||||
|
||||
// convert from `nym_kkt_ciphersuite` types into `nym_nodes_requests`
|
||||
digests
|
||||
.into_iter()
|
||||
.map(|(kem, kem_digests)| {
|
||||
(
|
||||
kem.into(),
|
||||
kem_digests
|
||||
.into_iter()
|
||||
.map(|(f, digest)| (f.into(), hex::encode(&digest)))
|
||||
.collect(),
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub(crate) async fn build_http_server(&self) -> Result<NymNodeHttpServer, NymNodeError> {
|
||||
let auxiliary_details = api_requests::v1::node::models::AuxiliaryDetails {
|
||||
@@ -899,7 +942,21 @@ impl NymNode {
|
||||
policy: None,
|
||||
};
|
||||
|
||||
let mut config = HttpServerConfig::new()
|
||||
let lewes_protocol = LewesProtocol {
|
||||
enabled: self.modes().entry,
|
||||
control_port: self.config.lp.announced_control_port(),
|
||||
data_port: self.config.lp.announced_data_port(),
|
||||
x25519: self.x25519_lp_keys.pk,
|
||||
kem_keys: self.compute_kem_key_hashes(),
|
||||
};
|
||||
|
||||
// SAFETY: the only way for this call to fail is if serialisation of LewesProtocol fails.
|
||||
// however, that conversion is stable and infallible
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let signed_lewes_protocol =
|
||||
SignedData::new(lewes_protocol, self.ed25519_identity_keys.private_key()).unwrap();
|
||||
|
||||
let mut config = HttpServerConfig::new(signed_lewes_protocol)
|
||||
.with_landing_page_assets(self.config.http.landing_page_assets_path.as_ref())
|
||||
.with_mixnode_details(mixnode_details)
|
||||
.with_gateway_details(gateway_details)
|
||||
@@ -1345,7 +1402,7 @@ impl NymNode {
|
||||
Ok(self.shutdown_manager)
|
||||
}
|
||||
|
||||
pub(crate) async fn run(mut self) -> Result<(), NymNodeError> {
|
||||
pub async fn run(mut self) -> Result<(), NymNodeError> {
|
||||
let mut shutdown_signals = self.shutdown_manager.detach_shutdown_signals();
|
||||
|
||||
// listen for shutdown signal in case we received it when attempting to spawn all the tasks
|
||||
|
||||
Reference in New Issue
Block a user