From 05ed77568655ebca41937bc5ca6615e9b11ffdd9 Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Fri, 22 May 2026 11:49:37 +0200 Subject: [PATCH] new addressing + pipeline unification and routing stubs --- Cargo.lock | 3 + .../client-libs/mixnet-client/src/client.rs | 9 +- common/nym-lp-data/src/lib.rs | 9 + common/nymsphinx/addressing/Cargo.toml | 3 + common/nymsphinx/addressing/src/clients.rs | 87 +++++- common/nymsphinx/addressing/src/lib.rs | 2 +- common/nymsphinx/addressing/src/nodes.rs | 256 +++++++++++------- common/nymsphinx/forwarding/src/packet.rs | 5 - nym-node/nym-node-metrics/src/mixnet.rs | 26 ++ .../src/prometheus_wrapper.rs | 30 +- nym-node/src/config/mod.rs | 6 + nym-node/src/node/lp/data/handler/error.rs | 7 + .../messages/{mixnode.rs => mixing.rs} | 2 +- .../src/node/lp/data/handler/messages/mod.rs | 8 +- .../messages/{gateway.rs => nymnode.rs} | 38 ++- nym-node/src/node/lp/data/handler/mod.rs | 4 +- .../node/lp/data/handler/pipeline/gateway.rs | 161 ----------- .../src/node/lp/data/handler/pipeline/mod.rs | 8 +- .../pipeline/{mixnode.rs => nymnode.rs} | 163 +++++++---- .../src/node/lp/data/handler/pipeline/wire.rs | 2 +- .../node/lp/data/handler/processing/outfox.rs | 42 ++- .../node/lp/data/handler/processing/sphinx.rs | 90 +++++- nym-node/src/node/lp/data/shared.rs | 25 +- .../at_last_update.rs | 11 + .../handler/global_prometheus_updater/mod.rs | 16 ++ .../src/node/mixnet/packet_forwarding/mod.rs | 23 +- 26 files changed, 665 insertions(+), 371 deletions(-) rename nym-node/src/node/lp/data/handler/messages/{mixnode.rs => mixing.rs} (97%) rename nym-node/src/node/lp/data/handler/messages/{gateway.rs => nymnode.rs} (73%) delete mode 100644 nym-node/src/node/lp/data/handler/pipeline/gateway.rs rename nym-node/src/node/lp/data/handler/pipeline/{mixnode.rs => nymnode.rs} (84%) diff --git a/Cargo.lock b/Cargo.lock index 8a651faef5..d36e502eca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8427,11 +8427,14 @@ name = "nym-sphinx-addressing" version = "1.21.0" dependencies = [ "bincode", + "blake3", + "bs58", "nym-crypto", "nym-sphinx-types", "rand 0.8.6", "serde", "serde_json", + "strum 0.28.0", "thiserror 2.0.18", ] diff --git a/common/client-libs/mixnet-client/src/client.rs b/common/client-libs/mixnet-client/src/client.rs index 0bdb06a198..50fc5fc0b1 100644 --- a/common/client-libs/mixnet-client/src/client.rs +++ b/common/client-libs/mixnet-client/src/client.rs @@ -5,6 +5,7 @@ use dashmap::DashMap; use futures::StreamExt; use nym_noise::config::NoiseConfig; use nym_noise::upgrade_noise_initiator; +use nym_sphinx::addressing::nodes::NymNodeRoutingAddress; use nym_sphinx::forwarding::packet::MixPacket; use nym_sphinx::framing::codec::NymCodec; use nym_sphinx::framing::packet::FramedNymPacket; @@ -309,7 +310,13 @@ impl Client { impl SendWithoutResponse for Client { fn send_without_response(&self, packet: MixPacket) -> io::Result<()> { - let address = packet.next_hop_address(); + let address = match packet.next_hop() { + NymNodeRoutingAddress::Client(_) => { + warn!("mix packet addressed to a client in the legacy send_without_response path. This should never happen!"); + return Ok(()); + } + NymNodeRoutingAddress::Node(address) => address, + }; trace!("Sending packet to {address}"); // TODO: optimisation for the future: rather than constantly using legacy encoding, diff --git a/common/nym-lp-data/src/lib.rs b/common/nym-lp-data/src/lib.rs index 9f411ba224..a1b30e4da3 100644 --- a/common/nym-lp-data/src/lib.rs +++ b/common/nym-lp-data/src/lib.rs @@ -152,6 +152,15 @@ impl PipelineData { } } + /// Set a new destination + pub fn with_dst(self, new_dst: NewNdId) -> PipelineData { + PipelineData { + data: self.data, + options: self.options, + dst: new_dst, + } + } + /// Drop the pipeline options, producing a plain addressed payload. pub fn into_addressed(self) -> AddressedTimedData { AddressedTimedData { diff --git a/common/nymsphinx/addressing/Cargo.toml b/common/nymsphinx/addressing/Cargo.toml index a5b2ed291e..e1d0263f19 100644 --- a/common/nymsphinx/addressing/Cargo.toml +++ b/common/nymsphinx/addressing/Cargo.toml @@ -13,9 +13,12 @@ readme.workspace = true publish = true [dependencies] +blake3 = { workspace = true } # short-fingerprint hash for ClientAddress +bs58 = { workspace = true } # base58 string form for ClientAddress nym-crypto = { workspace = true, features = ["asymmetric", "sphinx"] } # all addresses are expressed in terms on their crypto keys nym-sphinx-types = { workspace = true, features = ["sphinx"] } # we need to be able to refer to some types defined inside sphinx crate serde = { workspace = true } # implementing serialization/deserialization for some types, like `Recipient` +strum = { workspace = true } thiserror = { workspace = true } [dev-dependencies] diff --git a/common/nymsphinx/addressing/src/clients.rs b/common/nymsphinx/addressing/src/clients.rs index d7aba4a13e..8655c93356 100644 --- a/common/nymsphinx/addressing/src/clients.rs +++ b/common/nymsphinx/addressing/src/clients.rs @@ -4,7 +4,7 @@ // of a helper/utils structure, because before it reaches the gateway // it's already destructed). -use crate::nodes::{NODE_IDENTITY_SIZE, NodeIdentity}; +use crate::nodes::{NODE_IDENTITY_SIZE, NodeIdentity, NymNodeRoutingAddress}; use nym_crypto::asymmetric::{ed25519, x25519}; use nym_sphinx_types::Destination; use serde::de::{Error as SerdeError, SeqAccess, Unexpected, Visitor}; @@ -21,7 +21,53 @@ const CLIENT_ENCRYPTION_KEY_SIZE: usize = x25519::PUBLIC_KEY_SIZE; pub type ClientIdentity = ed25519::PublicKey; const CLIENT_IDENTITY_SIZE: usize = ed25519::PUBLIC_KEY_LENGTH; -pub type RecipientBytes = [u8; Recipient::LEN]; +/// 20-byte fingerprint of a client's identity, used as the routing handle +/// that a gateway dispatches over UDP to the corresponding connected client. +#[derive(Copy, Clone, PartialEq, Eq, Hash)] +pub struct ClientAddress([u8; ClientAddress::LEN]); + +impl ClientAddress { + pub const LEN: usize = 20; + + pub fn from_bytes(bytes: [u8; Self::LEN]) -> Self { + Self(bytes) + } + + pub fn to_bytes(&self) -> Vec { + self.0.to_vec() + } + + pub fn into_bytes(self) -> [u8; Self::LEN] { + self.0 + } + + /// Derive the address from a client's ed25519 identity. Hash is BLAKE3 of the + /// identity's 32-byte encoding, truncated to 20 bytes. + // Eth addressing uses 20 bytes, 2^80 birthday paradox collision is fine for our use case + pub fn from_identity(identity: &ClientIdentity) -> Self { + let digest = blake3::hash(&identity.to_bytes()); + let mut out = [0u8; Self::LEN]; + out.copy_from_slice(&digest.as_bytes()[..Self::LEN]); + Self(out) + } + + pub fn to_base58_string(&self) -> String { + bs58::encode(&self.0).into_string() + } +} + +impl fmt::Display for ClientAddress { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + f.write_str(&self.to_base58_string()) + } +} + +impl fmt::Debug for ClientAddress { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + // use the Display implementation + ::fmt(self, f) + } +} #[derive(Debug, Error)] pub enum RecipientFormattingError { @@ -38,12 +84,20 @@ pub enum RecipientFormattingError { MalformedGatewayError(ed25519::Ed25519RecoveryError), } +pub type RecipientBytes = [u8; Recipient::LEN]; + // TODO: this should a different home... somewhere, but where? #[derive(Clone, Copy, PartialEq, Eq)] pub struct Recipient { client_identity: ClientIdentity, + + // x25519 key. Either legacy used for e2e encryption of payload, or for the last sphinx layer client_encryption_key: ClientEncryptionKey, + gateway: NodeIdentity, + // Cached blake3 fingerprint of `client_identity`. Not part of the wire format; + // recomputed on every constructor so deserialization stays consistent. + client_address: ClientAddress, } // Serialize + Deserialize is not really used anymore (it was for a CBOR experiment) @@ -143,10 +197,12 @@ impl Recipient { client_encryption_key: ClientEncryptionKey, gateway: NodeIdentity, ) -> Self { + let client_address = ClientAddress::from_identity(&client_identity); Recipient { client_identity, client_encryption_key, gateway, + client_address, } } @@ -162,6 +218,10 @@ impl Recipient { ) } + pub fn as_sphinx_hop(&self) -> NymNodeRoutingAddress { + NymNodeRoutingAddress::Client(self.client_address) + } + pub fn identity(&self) -> &ClientIdentity { &self.client_identity } @@ -174,6 +234,10 @@ impl Recipient { self.gateway } + pub fn client_address(&self) -> &ClientAddress { + &self.client_address + } + pub fn to_bytes(self) -> RecipientBytes { let mut out = [0u8; Self::LEN]; out[..CLIENT_IDENTITY_SIZE].copy_from_slice(&self.client_identity.to_bytes()); @@ -203,11 +267,11 @@ impl Recipient { Err(err) => return Err(RecipientFormattingError::MalformedGatewayError(err)), }; - Ok(Recipient { + Ok(Recipient::new( client_identity, client_encryption_key, gateway, - }) + )) } pub fn try_from_base58_string>( @@ -244,11 +308,11 @@ impl Recipient { Err(err) => return Err(RecipientFormattingError::MalformedGatewayError(err)), }; - Ok(Recipient { + Ok(Recipient::new( client_identity, client_encryption_key, gateway, - }) + )) } } @@ -395,4 +459,15 @@ mod tests { assert_eq!(a, b); } + + #[test] + fn client_address_is_deterministic_blake3_truncated() { + let recipient = mock_recipient(); + let recomputed = ClientAddress::from_identity(&recipient.client_identity); + assert_eq!(recipient.client_address(), &recomputed); + + // also verify cached field survives the bytes round-trip + let roundtripped = Recipient::try_from_bytes(recipient.to_bytes()).unwrap(); + assert_eq!(roundtripped.client_address(), recipient.client_address()); + } } diff --git a/common/nymsphinx/addressing/src/lib.rs b/common/nymsphinx/addressing/src/lib.rs index 0d5e26b92a..545e6bd464 100644 --- a/common/nymsphinx/addressing/src/lib.rs +++ b/common/nymsphinx/addressing/src/lib.rs @@ -4,5 +4,5 @@ pub mod clients; pub mod nodes; -pub use clients::Recipient; +pub use clients::{ClientAddress, Recipient}; pub use nodes::NodeIdentity; diff --git a/common/nymsphinx/addressing/src/nodes.rs b/common/nymsphinx/addressing/src/nodes.rs index 49fba4129e..290520cc45 100644 --- a/common/nymsphinx/addressing/src/nodes.rs +++ b/common/nymsphinx/addressing/src/nodes.rs @@ -5,9 +5,10 @@ //! //! This module is responsible for encoding and decoding node routing information, so that //! they could be later put into an appropriate field in a sphinx header. -//! Currently, that routing information is an IP address, but in principle it can be anything -//! for as long as it's going to fit in the field. +//! A routing address is either a `SocketAddr` (mix node / gateway socket) or a `ClientAddress` +//! (a 20-byte fingerprint or a client's identity key. +use crate::clients::ClientAddress; use nym_crypto::asymmetric::ed25519; use nym_sphinx_types::{NODE_ADDRESS_LENGTH, NodeAddressBytes}; @@ -20,7 +21,9 @@ pub type NodeIdentity = ed25519::PublicKey; pub const NODE_IDENTITY_SIZE: usize = ed25519::PUBLIC_KEY_LENGTH; /// MAX_UNPADDED_LEN represents maximum length an unpadded address could have. -/// In this case it's an ipv6 socket address (with version prefix) +/// Reserved for ACK-SURB hop overhead calculations, which only ever target mix nodes, +/// so the cap is the IPv6 socket variant (1 + 2 + 16 = 19 bytes). +// SW Double check that pub const MAX_NODE_ADDRESS_UNPADDED_LEN: usize = 19; #[derive(Debug, Error)] @@ -29,18 +32,16 @@ pub enum NymNodeRoutingAddressError { NoBytesProvided, #[error( - "Provided insufficient amount of few bytes to deserialize a valid NymNodeRoutingAddress for IPv{protocol_version} variant. Received {received} and required {required}" + "Provided insufficient amount of few bytes to deserialize a valid NymNodeRoutingAddress for type {address_type}. Received {received} and required {required}" )] TooFewBytesProvided { - protocol_version: u8, + address_type: AddressType, received: usize, required: usize, }, - #[error( - "{received} is not a valid version of the Internet Protocol (IP). Expected either '4' or '6'" - )] - InvalidIpVersion { received: u8 }, + #[error("{received:#x} is not a valid NymNodeRoutingAddress address type")] + InvalidAddressType { received: u8 }, #[error( "Could not serialize NymNodeRoutingAddress into NodeAddressBytes as that requires using at least {required} bytes and only {NODE_ADDRESS_LENGTH} are available" @@ -48,14 +49,60 @@ pub enum NymNodeRoutingAddressError { TooSmallBytesRepresentation { required: usize }, } -/// Current representation of Node routing information used in Nym system. -/// At this point of time it is a simple `SocketAddr`. +/// On-wire variant tag of a [`NymNodeRoutingAddress`]. Always the first byte +/// of the encoded routing field. +#[repr(u8)] +#[derive(Copy, Clone, Debug, PartialEq, Eq, strum::Display)] +pub enum AddressType { + Ipv4 = 4, + Ipv6 = 6, + Client = 12, +} + +impl AddressType { + pub fn as_u8(self) -> u8 { + self as u8 + } + + /// Number of bytes needed to represent the address type. + pub fn bytes_len(&self) -> usize { + match self { + AddressType::Ipv4 => 1 + 4 + 2, // marker, address, port + AddressType::Ipv6 => 1 + 16 + 2, // marker, address, port + AddressType::Client => 1 + ClientAddress::LEN, // marker, fingerprint + } + } +} + +impl TryFrom for AddressType { + type Error = NymNodeRoutingAddressError; + + fn try_from(b: u8) -> Result { + match b { + x if x == AddressType::Ipv4 as u8 => Ok(AddressType::Ipv4), + x if x == AddressType::Ipv6 as u8 => Ok(AddressType::Ipv6), + x if x == AddressType::Client as u8 => Ok(AddressType::Client), + v => Err(NymNodeRoutingAddressError::InvalidAddressType { received: v }), + } + } +} + +/// Routing information that can appear in a sphinx hop's address field. +/// +/// `Node` carries an inter-node socket address (mix node or gateway egress). +/// `Client` carries a 20-byte fingerprint of the destination client's identity #[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)] -pub struct NymNodeRoutingAddress(SocketAddr); +pub enum NymNodeRoutingAddress { + Node(SocketAddr), + Client(ClientAddress), +} impl std::fmt::Display for NymNodeRoutingAddress { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - self.0.fmt(f) + match self { + NymNodeRoutingAddress::Node(addr) => addr.fmt(f), + NymNodeRoutingAddress::Client(ca) => ca.fmt(f), + } } } @@ -64,26 +111,29 @@ impl NymNodeRoutingAddress { /// The value has no upper bound as when converted into bytes, it's always /// padded with zeroes to be exactly NODE_ADDRESS_LENGTH long. pub fn bytes_min_len(&self) -> usize { - match self.0 { - SocketAddr::V4(_) => 7, - SocketAddr::V6(_) => 19, - } + self.addr_type().bytes_len() } /// Converts self into a vector of bytes. /// Note, this represents a generic bytes vector, not necessarily a NodeAddressBytes /// and hence is not zero-padded. pub fn as_bytes(&self) -> Vec { - let port_bytes = self.0.port().to_be_bytes(); - let ip_octets_vec = match self.0.ip() { - IpAddr::V4(ip) => ip.octets().to_vec(), - IpAddr::V6(ip) => ip.octets().to_vec(), - }; - - std::iter::once(self.addr_type_as_u8()) - .chain(port_bytes.iter().cloned()) - .chain(ip_octets_vec.iter().cloned()) - .collect() + match self { + NymNodeRoutingAddress::Node(socket) => { + let port_bytes = socket.port().to_be_bytes(); + let ip_octets_vec = match socket.ip() { + IpAddr::V4(ip) => ip.octets().to_vec(), + IpAddr::V6(ip) => ip.octets().to_vec(), + }; + std::iter::once(self.addr_type().as_u8()) + .chain(port_bytes.iter().cloned()) + .chain(ip_octets_vec.iter().cloned()) + .collect() + } + NymNodeRoutingAddress::Client(address) => std::iter::once(self.addr_type().as_u8()) + .chain(address.to_bytes()) + .collect(), + } } /// Converts self into a vector of bytes optionally padded with zeroes to the `expected_len`. @@ -110,80 +160,68 @@ impl NymNodeRoutingAddress { return Err(NymNodeRoutingAddressError::NoBytesProvided); } - let ip_version = b[0]; - let ip = match ip_version { - 4 => { - if b.len() < 7 { - return Err(NymNodeRoutingAddressError::TooFewBytesProvided { - protocol_version: 4, - received: b.len(), - required: 7, - }); - } - IpAddr::V4(Ipv4Addr::new(b[3], b[4], b[5], b[6])) + let address_type = AddressType::try_from(b[0])?; + if b.len() < address_type.bytes_len() { + return Err(NymNodeRoutingAddressError::TooFewBytesProvided { + address_type, + received: b.len(), + required: address_type.bytes_len(), + }); + } + + match address_type { + AddressType::Ipv4 => { + let port = u16::from_be_bytes([b[1], b[2]]); + let ip = IpAddr::V4(Ipv4Addr::new(b[3], b[4], b[5], b[6])); + Ok(NymNodeRoutingAddress::Node(SocketAddr::new(ip, port))) } - 6 => { - if b.len() < 19 { - return Err(NymNodeRoutingAddressError::TooFewBytesProvided { - protocol_version: 6, - received: b.len(), - required: 19, - }); - } + AddressType::Ipv6 => { + let port = u16::from_be_bytes([b[1], b[2]]); let mut address_octets = [0u8; 16]; address_octets.copy_from_slice(&b[3..19]); - IpAddr::V6(Ipv6Addr::from(address_octets)) + let ip = IpAddr::V6(Ipv6Addr::from(address_octets)); + Ok(NymNodeRoutingAddress::Node(SocketAddr::new(ip, port))) } - v => return Err(NymNodeRoutingAddressError::InvalidIpVersion { received: v }), - }; - - let port: u16 = u16::from_be_bytes([b[1], b[2]]); - - Ok(Self(SocketAddr::new(ip, port))) + AddressType::Client => { + let mut address_bytes = [0u8; ClientAddress::LEN]; + address_bytes.copy_from_slice(&b[1..ClientAddress::LEN]); + Ok(NymNodeRoutingAddress::Client(ClientAddress::from_bytes( + address_bytes, + ))) + } + } } - /// Single byte representation of self ip version. - pub fn addr_type_as_u8(&self) -> u8 { - match self.0 { - SocketAddr::V4(_) => 4, - SocketAddr::V6(_) => 6, + /// Variant tag of this routing address. + pub fn addr_type(&self) -> AddressType { + match self { + NymNodeRoutingAddress::Node(SocketAddr::V4(_)) => AddressType::Ipv4, + NymNodeRoutingAddress::Node(SocketAddr::V6(_)) => AddressType::Ipv6, + NymNodeRoutingAddress::Client(_) => AddressType::Client, } } } -/// Considering `NymNodeRoutingAddress` is equivalent to a `SocketAddr` at this point, -/// it makes perfect sense to allow the bilateral transformation. impl From for NymNodeRoutingAddress { fn from(addr: SocketAddr) -> Self { - Self(addr) + NymNodeRoutingAddress::Node(addr) } } -/// Considering `NymNodeRoutingAddress` is equivalent to a `SocketAddr` at this point, -/// it makes perfect sense to allow the bilateral transformation. -impl From for SocketAddr { - fn from(addr: NymNodeRoutingAddress) -> Self { - addr.0 - } -} - -impl AsRef for NymNodeRoutingAddress { - fn as_ref(&self) -> &SocketAddr { - &self.0 +impl From for NymNodeRoutingAddress { + fn from(addr: ClientAddress) -> Self { + NymNodeRoutingAddress::Client(addr) } } impl TryInto for NymNodeRoutingAddress { type Error = NymNodeRoutingAddressError; - /// `NymNodeRoutingAddress` (as a `SocketAddr`) is represented the following way: - /// VersionFlag || port || octets || zeropad - /// VersionFlag is one byte representing whether self is ipv4 or ipv6 address, - /// port is 16bit big endian representation of port value - /// octets is bytes representation of octets making up the ip address of the socket address - /// (either 4 bytes for ipv4 or 16 bytes for ipv6) - /// zeropad is padding of 0 for the `NymNodeRoutingAddress` to be - /// exactly `NODE_ADDRESS_LENGTH` long. + /// On-wire encoding of a `NymNodeRoutingAddress` as a fixed-size sphinx routing field. + /// VARIANT_TAG || payload || zeropad + /// - 0x04: IPv4 socket — payload is `port (2) || octets (4)` + /// - 0x06: IPv6 socket — payload is `port (2) || octets (16)` + /// - 0x0C: client fingerprint — payload is `client_address (20)` fn try_into(self) -> Result { // first check if we have enough bytes to represent `self`: if self.bytes_min_len() > NODE_ADDRESS_LENGTH { @@ -213,9 +251,24 @@ impl TryFrom for NymNodeRoutingAddress { mod tests { use super::*; + fn v4() -> NymNodeRoutingAddress { + NymNodeRoutingAddress::Node(SocketAddr::new(IpAddr::from([1, 2, 3, 4]), 42)) + } + + fn v6() -> NymNodeRoutingAddress { + NymNodeRoutingAddress::Node(SocketAddr::new( + IpAddr::from([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]), + 42, + )) + } + + fn client() -> NymNodeRoutingAddress { + NymNodeRoutingAddress::Client(ClientAddress::from_bytes([7u8; ClientAddress::LEN])) + } + #[test] fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_v4_address() { - let address = NymNodeRoutingAddress(SocketAddr::new(IpAddr::from([1, 2, 3, 4]), 42)); + let address = v4(); let address_bytes = address.as_bytes(); assert_eq!( address, @@ -225,10 +278,17 @@ mod tests { #[test] fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_v6_address() { - let address = NymNodeRoutingAddress(SocketAddr::new( - IpAddr::from([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]), - 42, - )); + let address = v6(); + let address_bytes = address.as_bytes(); + assert_eq!( + address, + NymNodeRoutingAddress::try_from_bytes(&address_bytes).unwrap() + ) + } + + #[test] + fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_client_address() { + let address = client(); let address_bytes = address.as_bytes(); assert_eq!( address, @@ -238,7 +298,7 @@ mod tests { #[test] fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_empty_v4_address() { - let address = NymNodeRoutingAddress(SocketAddr::new(IpAddr::from([0, 0, 0, 0]), 42)); + let address = NymNodeRoutingAddress::Node(SocketAddr::new(IpAddr::from([0, 0, 0, 0]), 42)); let address_bytes = address.as_bytes(); assert_eq!( address, @@ -248,7 +308,7 @@ mod tests { #[test] fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_empty_v6_address() { - let address = NymNodeRoutingAddress(SocketAddr::new( + let address = NymNodeRoutingAddress::Node(SocketAddr::new( IpAddr::from([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]), 42, )); @@ -262,16 +322,18 @@ mod tests { #[test] fn nym_node_routing_address_can_be_converted_to_and_from_node_address_bytes_with_no_data_loss() { - let address_v4 = NymNodeRoutingAddress(SocketAddr::new(IpAddr::from([1, 2, 3, 4]), 42)); - let address_v6 = NymNodeRoutingAddress(SocketAddr::new( - IpAddr::from([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]), - 42, + for address in [v4(), v6(), client()] { + let node_address: NodeAddressBytes = address.try_into().unwrap(); + assert_eq!(address, node_address.try_into().unwrap()); + } + } + + #[test] + fn try_from_bytes_rejects_unknown_variant_tag() { + let bytes = [0xFFu8, 0, 0, 0, 0, 0, 0]; + assert!(matches!( + NymNodeRoutingAddress::try_from_bytes(&bytes), + Err(NymNodeRoutingAddressError::InvalidAddressType { received: 0xFF }) )); - - let node_address1: NodeAddressBytes = address_v4.try_into().unwrap(); - let node_address2: NodeAddressBytes = address_v6.try_into().unwrap(); - - assert_eq!(address_v4, node_address1.try_into().unwrap()); - assert_eq!(address_v6, node_address2.try_into().unwrap()); } } diff --git a/common/nymsphinx/forwarding/src/packet.rs b/common/nymsphinx/forwarding/src/packet.rs index 9fb9f3a915..ac8190649a 100644 --- a/common/nymsphinx/forwarding/src/packet.rs +++ b/common/nymsphinx/forwarding/src/packet.rs @@ -9,7 +9,6 @@ use nym_sphinx_anonymous_replies::reply_surb::AppliedReplySurb; use nym_sphinx_params::key_rotation::InvalidSphinxKeyRotation; use nym_sphinx_params::packet_sizes::InvalidPacketSize; use nym_sphinx_params::packet_types::InvalidPacketType; -use std::net::SocketAddr; use thiserror::Error; #[derive(Debug, Error)] @@ -81,10 +80,6 @@ impl MixPacket { self.next_hop } - pub fn next_hop_address(&self) -> SocketAddr { - self.next_hop.into() - } - pub fn packet(&self) -> &NymPacket { &self.packet } diff --git a/nym-node/nym-node-metrics/src/mixnet.rs b/nym-node/nym-node-metrics/src/mixnet.rs index b771b10154..fd6e1f80f3 100644 --- a/nym-node/nym-node-metrics/src/mixnet.rs +++ b/nym-node/nym-node-metrics/src/mixnet.rs @@ -221,6 +221,16 @@ impl MixingStats { .egress_overloaded_dropped_packets .fetch_add(1, Ordering::Relaxed); } + + pub fn lp_client_forwarding_disabled_dropped(&self) { + self.lp + .client_forwarding_disabled_dropped + .fetch_add(1, Ordering::Relaxed); + } + + pub fn lp_internal_sp_routed(&self) { + self.lp.internal_sp_routed.fetch_add(1, Ordering::Relaxed); + } } #[derive(Clone, Copy, Default, PartialEq)] @@ -472,6 +482,13 @@ pub struct LpMixingStats { worker_pool_overloaded_dropped_packets: AtomicUsize, /// Packets dropped because the handler->listener egress channel was full. egress_overloaded_dropped_packets: AtomicUsize, + + /// Client-destined packets dropped because this node has client forwarding disabled + /// (i.e. not a gateway). Receiving these indicates upstream misrouting or abuse. + client_forwarding_disabled_dropped: AtomicUsize, + /// Packets routed to an internal service provider channel (delivered off-wire, + /// so they don't show up in `packets_forwarded`). + internal_sp_routed: AtomicUsize, } impl LpMixingStats { @@ -563,4 +580,13 @@ impl LpMixingStats { self.egress_overloaded_dropped_packets .load(Ordering::Relaxed) } + + pub fn client_forwarding_disabled_dropped(&self) -> usize { + self.client_forwarding_disabled_dropped + .load(Ordering::Relaxed) + } + + pub fn internal_sp_routed(&self) -> usize { + self.internal_sp_routed.load(Ordering::Relaxed) + } } diff --git a/nym-node/nym-node-metrics/src/prometheus_wrapper.rs b/nym-node/nym-node-metrics/src/prometheus_wrapper.rs index f4253fe040..7efa2ab4b2 100644 --- a/nym-node/nym-node-metrics/src/prometheus_wrapper.rs +++ b/nym-node/nym-node-metrics/src/prometheus_wrapper.rs @@ -170,6 +170,16 @@ pub enum PrometheusMetric { ))] MixnetLpEgressOverloadedDropped, + #[strum(props( + help = "The number of LP client-destined packets dropped because client forwarding is disabled on this node" + ))] + MixnetLpClientForwardingDisabledDropped, + + #[strum(props( + help = "The number of LP packets routed to an internal service provider (delivered off-wire)" + ))] + MixnetLpInternalSpRouted, + #[strum(props(help = "The current rate of receiving LP UDP datagrams"))] MixnetLpPacketsReceivedRate, @@ -217,6 +227,14 @@ pub enum PrometheusMetric { ))] MixnetLpEgressOverloadedDroppedRate, + #[strum(props( + help = "The current rate of LP client-destined packets dropped because client forwarding is disabled" + ))] + MixnetLpClientForwardingDisabledDroppedRate, + + #[strum(props(help = "The current rate of LP packets routed to an internal service provider"))] + MixnetLpInternalSpRoutedRate, + // # ENTRY #[strum(props(help = "The number of unique users"))] EntryClientUniqueUsers, @@ -423,6 +441,10 @@ impl PrometheusMetric { PrometheusMetric::MixnetLpEgressOverloadedDropped => { Metric::new_int_gauge(&name, help) } + PrometheusMetric::MixnetLpClientForwardingDisabledDropped => { + Metric::new_int_gauge(&name, help) + } + PrometheusMetric::MixnetLpInternalSpRouted => Metric::new_int_gauge(&name, help), PrometheusMetric::MixnetLpPacketsReceivedRate => { Metric::new_float_gauge(&name, help) } @@ -462,6 +484,12 @@ impl PrometheusMetric { PrometheusMetric::MixnetLpEgressOverloadedDroppedRate => { Metric::new_float_gauge(&name, help) } + PrometheusMetric::MixnetLpClientForwardingDisabledDroppedRate => { + Metric::new_float_gauge(&name, help) + } + PrometheusMetric::MixnetLpInternalSpRoutedRate => { + Metric::new_float_gauge(&name, help) + } PrometheusMetric::EntryClientUniqueUsers => Metric::new_int_gauge(&name, help), PrometheusMetric::EntryClientSessionsStarted => Metric::new_int_gauge(&name, help), PrometheusMetric::EntryClientSessionsFinished => Metric::new_int_gauge(&name, help), @@ -609,7 +637,7 @@ mod tests { // a sanity check for anyone adding new metrics. if this test fails, // make sure any methods on `PrometheusMetric` enum don't need updating // or require custom Display impl - assert_eq!(71, PrometheusMetric::COUNT) + assert_eq!(75, PrometheusMetric::COUNT) } #[test] diff --git a/nym-node/src/config/mod.rs b/nym-node/src/config/mod.rs index 14b00b7f5e..fe7a9b6d2c 100644 --- a/nym-node/src/config/mod.rs +++ b/nym-node/src/config/mod.rs @@ -145,6 +145,12 @@ impl NodeModes { self.entry || self.exit } + // Duplicate of `expects_final_hop_traffic` for now, but with more explicit naming for LP context + // Comment can be removed along the aformentioned fn + pub fn expects_client_traffic(&self) -> bool { + self.entry || self.exit + } + pub fn with_mixnode(&mut self) -> &mut Self { self.mixnode = true; self diff --git a/nym-node/src/node/lp/data/handler/error.rs b/nym-node/src/node/lp/data/handler/error.rs index f3e85f637b..308acecba5 100644 --- a/nym-node/src/node/lp/data/handler/error.rs +++ b/nym-node/src/node/lp/data/handler/error.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use nym_lp_data::packet::frame::LpFrameKind; +use nym_sphinx_addressing::clients::ClientAddress; use nym_sphinx_addressing::nodes::NymNodeRoutingAddressError; use nym_sphinx_forwarding::packet::MixPacketFormattingError; use nym_sphinx_framing::processing::PacketProcessingError; @@ -34,6 +35,12 @@ pub enum LpDataHandlerError { #[error("unwrapped a packet into a final hop packet. This is no longer supported")] FinalHop, + #[error("a mix node received a forward hop addressed to a client; only gateways may forward to clients")] + ClientForwardHopAtMixNode, + + #[error("no UDP endpoint registered for client {client_addr}")] + UnknownClientAddress { client_addr: ClientAddress }, + #[error("{0}")] Internal(String), diff --git a/nym-node/src/node/lp/data/handler/messages/mixnode.rs b/nym-node/src/node/lp/data/handler/messages/mixing.rs similarity index 97% rename from nym-node/src/node/lp/data/handler/messages/mixnode.rs rename to nym-node/src/node/lp/data/handler/messages/mixing.rs index 79755b834f..bd95fc445b 100644 --- a/nym-node/src/node/lp/data/handler/messages/mixnode.rs +++ b/nym-node/src/node/lp/data/handler/messages/mixing.rs @@ -8,7 +8,7 @@ use nym_sphinx_params::SphinxKeyRotation; use crate::node::lp::data::handler::error::LpDataHandlerError; -/// Message types supported by mixnodes. +/// Message types supported by nym-nodes with only mixnode role #[derive(Debug, Clone, Copy)] pub enum MixMessage { Sphinx(SphinxMixMessage), diff --git a/nym-node/src/node/lp/data/handler/messages/mod.rs b/nym-node/src/node/lp/data/handler/messages/mod.rs index 138e871762..336e824cd7 100644 --- a/nym-node/src/node/lp/data/handler/messages/mod.rs +++ b/nym-node/src/node/lp/data/handler/messages/mod.rs @@ -1,8 +1,8 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -mod gateway; -mod mixnode; +mod mixing; +mod nymnode; -pub use gateway::{ForwardOutfoxMixMessage, ForwardSphinxMixMessage, GatewayMessage}; -pub use mixnode::{MixMessage, OutfoxMixMessage, SphinxMixMessage}; +pub use mixing::{MixMessage, OutfoxMixMessage, SphinxMixMessage}; +pub use nymnode::{ForwardOutfoxMixMessage, ForwardSphinxMixMessage, NymNodeMessage}; diff --git a/nym-node/src/node/lp/data/handler/messages/gateway.rs b/nym-node/src/node/lp/data/handler/messages/nymnode.rs similarity index 73% rename from nym-node/src/node/lp/data/handler/messages/gateway.rs rename to nym-node/src/node/lp/data/handler/messages/nymnode.rs index e503102e43..3feabade45 100644 --- a/nym-node/src/node/lp/data/handler/messages/gateway.rs +++ b/nym-node/src/node/lp/data/handler/messages/nymnode.rs @@ -12,26 +12,24 @@ use crate::node::lp::data::handler::{ messages::{MixMessage, SphinxMixMessage}, }; -/// Message types supported by gateways. Gateways accept everything mixnodes -/// do (carried as [`GatewayMessage::Mix`]) plus the forward-only variants -/// that bypass sphinx/outfox processing. +/// Message types supported by nym-nodes with a gateway role. #[derive(Debug, Clone, Copy)] -pub enum GatewayMessage { +pub enum NymNodeMessage { Mix(MixMessage), ForwardSphinx(ForwardSphinxMixMessage), ForwardOutfox(ForwardOutfoxMixMessage), } -impl GatewayMessage { +impl NymNodeMessage { pub fn from_frame_header(header: LpFrameHeader) -> Result { match header.kind { LpFrameKind::SphinxPacket | LpFrameKind::OutfoxPacket => { - Ok(GatewayMessage::Mix(MixMessage::from_frame_header(header)?)) + Ok(NymNodeMessage::Mix(MixMessage::from_frame_header(header)?)) } - LpFrameKind::ForwardSphinxPacket => Ok(GatewayMessage::ForwardSphinx( + LpFrameKind::ForwardSphinxPacket => Ok(NymNodeMessage::ForwardSphinx( header.frame_attributes.try_into()?, )), - LpFrameKind::ForwardOutfoxPacket => Ok(GatewayMessage::ForwardOutfox( + LpFrameKind::ForwardOutfoxPacket => Ok(NymNodeMessage::ForwardOutfox( header.frame_attributes.try_into()?, )), _ => Err(LpDataHandlerError::UnsupportedLpFrameKind { typ: header.kind }), @@ -39,30 +37,30 @@ impl GatewayMessage { } } -impl From for PacketKind { - fn from(value: GatewayMessage) -> Self { +impl From for PacketKind { + fn from(value: NymNodeMessage) -> Self { match value { - GatewayMessage::Mix(msg) => msg.into(), - GatewayMessage::ForwardSphinx(_) => PacketKind::LpSphinx, - GatewayMessage::ForwardOutfox(_) => PacketKind::LpOutfox, + NymNodeMessage::Mix(msg) => msg.into(), + NymNodeMessage::ForwardSphinx(_) => PacketKind::LpSphinx, + NymNodeMessage::ForwardOutfox(_) => PacketKind::LpOutfox, } } } -impl From for GatewayMessage { +impl From for NymNodeMessage { fn from(value: MixMessage) -> Self { - GatewayMessage::Mix(value) + NymNodeMessage::Mix(value) } } -impl From for LpFrameHeader { - fn from(value: GatewayMessage) -> Self { +impl From for LpFrameHeader { + fn from(value: NymNodeMessage) -> Self { match value { - GatewayMessage::Mix(msg) => msg.into(), - GatewayMessage::ForwardSphinx(msg) => { + NymNodeMessage::Mix(msg) => msg.into(), + NymNodeMessage::ForwardSphinx(msg) => { LpFrameHeader::new(LpFrameKind::ForwardSphinxPacket, msg) } - GatewayMessage::ForwardOutfox(msg) => { + NymNodeMessage::ForwardOutfox(msg) => { LpFrameHeader::new(LpFrameKind::ForwardOutfoxPacket, msg) } } diff --git a/nym-node/src/node/lp/data/handler/mod.rs b/nym-node/src/node/lp/data/handler/mod.rs index 04bc407d56..fcedb68f35 100644 --- a/nym-node/src/node/lp/data/handler/mod.rs +++ b/nym-node/src/node/lp/data/handler/mod.rs @@ -16,7 +16,7 @@ //! use crate::node::lp::data::PACKET_BUFFER_SIZE; -use crate::node::lp::data::handler::pipeline::MixnodeDataPipeline; +use crate::node::lp::data::handler::pipeline::NymNodeDataPipeline; use crate::node::lp::data::shared::SharedLpDataState; use nym_lp_data::AddressedTimedData; use nym_lp_data::nymnodes::traits::NymNodeProcessingPipeline; @@ -215,7 +215,7 @@ impl LpDataHandler { input_rx: mpsc::Receiver, output_rx: mpsc::SyncSender, ) { - let mut pipeline = MixnodeDataPipeline::new(state.clone(), OsRng); + let mut pipeline = NymNodeDataPipeline::new(state.clone(), OsRng); while let Ok(input) = input_rx.recv() { // Blocking is fine, we don't want to unclog ourself and process a new packet that will be dropped anyway if let Err(e) = output_rx.send(pipeline.process(input.packet, input.timestamp)) { diff --git a/nym-node/src/node/lp/data/handler/pipeline/gateway.rs b/nym-node/src/node/lp/data/handler/pipeline/gateway.rs deleted file mode 100644 index 904994b935..0000000000 --- a/nym-node/src/node/lp/data/handler/pipeline/gateway.rs +++ /dev/null @@ -1,161 +0,0 @@ -// Copyright 2026 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use std::{net::SocketAddr, sync::Arc, time::Instant}; - -use nym_lp_data::{ - AddressedTimedData, PipelinePayload, TimedData, TimedPayload, - common::traits::{ - Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline, - WireWrappingPipeline, - }, - nymnodes::traits::NymNodeProcessingPipeline, - packet::{EncryptedLpPacket, LpFrame, LpHeader, MalformedLpPacketError, frame::LpFrameHeader}, -}; -use rand::Rng; -use tracing::warn; - -use crate::node::{ - lp::data::{ - handler::{ - messages::GatewayMessage, - pipeline::{MixnodeDataPipeline, wire::WirePipeline}, - processing, - }, - shared::SharedLpDataState, - }, - routing_filter::RoutingFilter, -}; - -pub(crate) struct GatewayDataPipeline { - state: Arc, - wire: WirePipeline, -} - -impl GatewayDataPipeline { - pub(crate) fn new(state: Arc, rng: R) -> Self { - Self { - state: state.clone(), - wire: WirePipeline::new(state, rng), - } - } -} - -// Processing logic -impl - NymNodeProcessingPipeline< - Instant, - EncryptedLpPacket, - GatewayMessage, - GatewayMessage, - SocketAddr, - > for GatewayDataPipeline -{ - fn mix( - &mut self, - message_kind: GatewayMessage, - payload: TimedPayload, - timestamp: Instant, - ) -> Vec> { - let processing_result = match message_kind { - GatewayMessage::Mix(msg) => { - return MixnodeDataPipeline::::process_mix_packet( - &self.state, - msg, - payload, - timestamp, - ) - .into_iter() - .map(|payload| payload.options_transform(Into::into)) - .collect(); - } - - // SW Need to check for bandwidth and routing filter as well - GatewayMessage::ForwardSphinx(metadata) => { - processing::sphinx::process_forward(&self.state, payload, metadata) - } - GatewayMessage::ForwardOutfox(metadata) => { - processing::outfox::process_forward(&self.state, payload, metadata) - } - }; - - Vec::new() - } -} - -// ============== Wire wrap/unwrap: pure delegation to WirePipeline ============== - -impl Framing for GatewayDataPipeline { - type Frame = LpFrame; - const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE; - - fn to_frame( - &mut self, - payload: PipelinePayload, - frame_size: usize, - ) -> Vec> { - let frame = LpFrame { - header: payload.options.into(), - content: payload.data.data.into(), - }; - self.wire - .message_to_frame(payload.data.timestamp, frame, payload.dst, frame_size) - } -} - -impl Transport for GatewayDataPipeline { - type Frame = LpFrame; - const OVERHEAD_SIZE: usize = LpHeader::SIZE; - - fn to_transport_packet( - &mut self, - frame: AddressedTimedData, - ) -> AddressedTimedData { - self.wire.frame_to_packet(frame) - } -} - -impl WireWrappingPipeline - for GatewayDataPipeline -{ - fn packet_size(&self) -> usize { - nym_lp_data::packet::MTU - } -} - -impl TransportUnwrap for GatewayDataPipeline { - type Frame = LpFrame; - type Error = MalformedLpPacketError; - - fn packet_to_frame( - &mut self, - packet: EncryptedLpPacket, - timestamp: Instant, - ) -> Result, Self::Error> { - self.wire.packet_to_frame(packet, timestamp) - } -} - -impl FramingUnwrap for GatewayDataPipeline { - type Frame = LpFrame; - - fn frame_to_message( - &mut self, - frame: TimedData, - ) -> Option<(TimedPayload, GatewayMessage)> { - let reassembled = self.wire.frame_to_maybe_message(frame)?; - let message_kind = GatewayMessage::from_frame_header(reassembled.data.header) - .inspect_err(|e| warn!("{e}")) - .ok()?; - self.state.message_received(message_kind); - Some(( - TimedPayload::new(reassembled.timestamp, reassembled.data.content.to_vec()), - message_kind, - )) - } -} - -impl WireUnwrappingPipeline - for GatewayDataPipeline -{ -} diff --git a/nym-node/src/node/lp/data/handler/pipeline/mod.rs b/nym-node/src/node/lp/data/handler/pipeline/mod.rs index bf0e5f45b8..28ae3361bf 100644 --- a/nym-node/src/node/lp/data/handler/pipeline/mod.rs +++ b/nym-node/src/node/lp/data/handler/pipeline/mod.rs @@ -1,12 +1,10 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -//! Per-role data pipelines (mixnode, gateway). The wire wrapping/unwrapping +//! Data pipelines. The wire wrapping/unwrapping //! shared between them lives in [`wire`]. -mod gateway; -mod mixnode; +mod nymnode; mod wire; -pub(crate) use gateway::GatewayDataPipeline; -pub(crate) use mixnode::MixnodeDataPipeline; +pub(crate) use nymnode::NymNodeDataPipeline; diff --git a/nym-node/src/node/lp/data/handler/pipeline/mixnode.rs b/nym-node/src/node/lp/data/handler/pipeline/nymnode.rs similarity index 84% rename from nym-node/src/node/lp/data/handler/pipeline/mixnode.rs rename to nym-node/src/node/lp/data/handler/pipeline/nymnode.rs index a1012fb344..bd171ab884 100644 --- a/nym-node/src/node/lp/data/handler/pipeline/mixnode.rs +++ b/nym-node/src/node/lp/data/handler/pipeline/nymnode.rs @@ -12,23 +12,29 @@ use nym_lp_data::{ nymnodes::traits::NymNodeProcessingPipeline, packet::{EncryptedLpPacket, LpFrame, LpHeader, MalformedLpPacketError, frame::LpFrameHeader}, }; +use nym_sphinx_addressing::nodes::NymNodeRoutingAddress; use rand::Rng; use tracing::warn; use crate::node::{ lp::data::{ - handler::{messages::MixMessage, pipeline::wire::WirePipeline, processing}, + handler::{ + error::LpDataHandlerError, + messages::{MixMessage, NymNodeMessage}, + pipeline::wire::WirePipeline, + processing, + }, shared::SharedLpDataState, }, routing_filter::RoutingFilter, }; -pub(crate) struct MixnodeDataPipeline { +pub(crate) struct NymNodeDataPipeline { state: Arc, wire: WirePipeline, } -impl MixnodeDataPipeline { +impl NymNodeDataPipeline { pub(crate) fn new(state: Arc, rng: R) -> Self { Self { state: state.clone(), @@ -36,23 +42,55 @@ impl MixnodeDataPipeline { } } - // Processing logic so gateways can reuse them + // Processing logic for packets supported by mixnode enabled node pub fn process_mix_packet( - shared_state: &SharedLpDataState, + &mut self, message_kind: MixMessage, payload: TimedPayload, _: Instant, - ) -> Vec> { - let processing_result = match message_kind { + ) -> Result, LpDataHandlerError> + { + match message_kind { MixMessage::Sphinx(metadata) => { - processing::sphinx::process(shared_state, payload, metadata) + processing::sphinx::process(&self.state, payload, metadata) } MixMessage::Outfox(metadata) => { - processing::outfox::process(shared_state, payload, metadata) + processing::outfox::process(&self.state, payload, metadata) + } + } + } +} + +// Processing logic +impl + NymNodeProcessingPipeline< + Instant, + EncryptedLpPacket, + NymNodeMessage, + NymNodeMessage, + SocketAddr, + > for NymNodeDataPipeline +{ + fn mix( + &mut self, + message_kind: NymNodeMessage, + payload: TimedPayload, + timestamp: Instant, + ) -> Vec> { + // Everything specific to a given packet type should happen here + let processing_result = match message_kind { + NymNodeMessage::Mix(msg) => self + .process_mix_packet(msg, payload, timestamp) + .map(|payload| payload.options_transform(Into::into)), + NymNodeMessage::ForwardSphinx(metadata) => { + processing::sphinx::process_forward(&self.state, payload, metadata) + } + NymNodeMessage::ForwardOutfox(metadata) => { + processing::outfox::process_forward(&self.state, payload, metadata) } }; - shared_state.update_processing_metrics(&processing_result); + self.state.update_processing_metrics(&processing_result); let packet_to_forward = match processing_result { Ok(packet) => packet, @@ -62,45 +100,54 @@ impl MixnodeDataPipeline { } }; - let next_hop = packet_to_forward.dst; - if !shared_state.routing_filter.should_route(next_hop.ip()) { - warn!( - event = "packet.dropped.routing_filter", - next_hop = %next_hop, - "dropping packet: egress address does not belong to any known node" - ); - shared_state.routing_filter_dropped(next_hop); - Vec::new() - } else { - vec![packet_to_forward] - } - } -} + // Now we are deciding if we are routing the packet and where -// Mixing logic -impl - NymNodeProcessingPipeline - for MixnodeDataPipeline -{ - fn mix( - &mut self, - message_kind: MixMessage, - payload: TimedPayload, - timestamp: Instant, - ) -> Vec> { - MixnodeDataPipeline::::process_mix_packet(&self.state, message_kind, payload, timestamp) + match packet_to_forward.dst { + NymNodeRoutingAddress::Node(next_hop) => { + if !self.state.routing_filter.should_route(next_hop.ip()) { + warn!( + event = "packet.dropped.routing_filter", + next_hop = %next_hop, + "dropping packet: egress address does not belong to any known node" + ); + self.state.routing_filter_dropped(next_hop); + Vec::new() + } else { + vec![packet_to_forward.with_dst(next_hop)] + } + } + NymNodeRoutingAddress::Client(client_address) => { + if !self.state.processing_config.client_forwarding_enabled { + warn!( + event = "packet.dropped.client_forwarding_disabled", + "dropping packet destined to a client_address on a client_forwarding_disabled node" + ); + self.state.client_forwarding_disabled_dropped(); + Vec::new() + } else if self.state.is_internal_service_provider(client_address) { + // SW TODO route packet to SP channel, no output (no need for framing and encrytpion, which is why we exit here) + self.state.internal_sp_routed(); + Vec::new() + } else { + // SW TODO client address lookup + #[allow(clippy::unwrap_used)] // This is tmp anyway + let client_socket_address = "0.0.0.0:0".parse().unwrap(); + vec![packet_to_forward.with_dst(client_socket_address)] + } + } + } } } // ============== Wire wrap/unwrap: pure delegation to WirePipeline ============== -impl Framing for MixnodeDataPipeline { +impl Framing for NymNodeDataPipeline { type Frame = LpFrame; const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE; fn to_frame( &mut self, - payload: PipelinePayload, + payload: PipelinePayload, frame_size: usize, ) -> Vec> { let frame = LpFrame { @@ -112,7 +159,7 @@ impl Framing for MixnodeDataPipeline } } -impl Transport for MixnodeDataPipeline { +impl Transport for NymNodeDataPipeline { type Frame = LpFrame; const OVERHEAD_SIZE: usize = LpHeader::SIZE; @@ -124,15 +171,15 @@ impl Transport for MixnodeDataPi } } -impl WireWrappingPipeline - for MixnodeDataPipeline +impl WireWrappingPipeline + for NymNodeDataPipeline { fn packet_size(&self) -> usize { nym_lp_data::packet::MTU } } -impl TransportUnwrap for MixnodeDataPipeline { +impl TransportUnwrap for NymNodeDataPipeline { type Frame = LpFrame; type Error = MalformedLpPacketError; @@ -145,17 +192,32 @@ impl TransportUnwrap for MixnodeDataPipeline } } -impl FramingUnwrap for MixnodeDataPipeline { +impl FramingUnwrap for NymNodeDataPipeline { type Frame = LpFrame; fn frame_to_message( &mut self, frame: TimedData, - ) -> Option<(TimedPayload, MixMessage)> { + ) -> Option<(TimedPayload, NymNodeMessage)> { let reassembled = self.wire.frame_to_maybe_message(frame)?; - let message_kind = MixMessage::from_frame_header(reassembled.data.header) + let message_kind = NymNodeMessage::from_frame_header(reassembled.data.header) .inspect_err(|e| warn!("{e}")) .ok()?; + + // Mix-only nodes must not accept gateway-role frames. Reject before processing + // so a malformed/malicious peer can't drive us through `process_forward`. + if !self.state.processing_config.client_forwarding_enabled + && !matches!(message_kind, NymNodeMessage::Mix(_)) + { + warn!( + event = "packet.dropped.unexpected_forward_frame", + ?message_kind, + "dropping unsupported frame on a node with client forwarding disabled" + ); + self.state.processing_misc_error(); + return None; + } + self.state.message_received(message_kind); Some(( TimedPayload::new(reassembled.timestamp, reassembled.data.content.to_vec()), @@ -164,8 +226,8 @@ impl FramingUnwrap for MixnodeDataPipeline { } } -impl WireUnwrappingPipeline - for MixnodeDataPipeline +impl WireUnwrappingPipeline + for NymNodeDataPipeline { } @@ -200,7 +262,7 @@ mod tests { use crate::node::key_rotation::active_keys::ActiveSphinxKeys; use crate::node::key_rotation::key::SphinxPrivateKey; use crate::node::lp::data::handler::messages::{MixMessage, SphinxMixMessage}; - use crate::node::lp::data::handler::pipeline::MixnodeDataPipeline; + use crate::node::lp::data::handler::pipeline::NymNodeDataPipeline; use crate::node::lp::data::shared::{ProcessingConfig, SharedLpDataState}; use crate::node::replay_protection::bloomfilter::{ ReplayProtectionBloomfilters, RotationFilter, @@ -238,6 +300,7 @@ mod tests { lp_config: LpConfig::default(), processing_config: ProcessingConfig { maximum_packet_delay: TEST_MAX_FORWARD_DELAY, + client_forwarding_enabled: true, }, sphinx_keys: ActiveSphinxKeys::new_loaded(primary, None), replay_protection_filter: ReplayProtectionBloomfilters::new(primary_bloom_filter, None), @@ -253,12 +316,12 @@ mod tests { /// Returns the pipeline together with the shared state (so tests can /// inspect metrics or trigger replays directly) fn mock_pipeline() -> ( - MixnodeDataPipeline, + NymNodeDataPipeline, Arc, ) { let mut rng = deterministic_rng(); let state = Arc::new(mock_shared_state(&mut rng)); - let pipeline = MixnodeDataPipeline::new(state.clone(), rng); + let pipeline = NymNodeDataPipeline::new(state.clone(), rng); (pipeline, state) } diff --git a/nym-node/src/node/lp/data/handler/pipeline/wire.rs b/nym-node/src/node/lp/data/handler/pipeline/wire.rs index c0960584db..9328f571a7 100644 --- a/nym-node/src/node/lp/data/handler/pipeline/wire.rs +++ b/nym-node/src/node/lp/data/handler/pipeline/wire.rs @@ -1,7 +1,7 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -//! Wire wrapping / unwrapping shared by the mixnode and gateway data pipelines. +//! Wire wrapping / unwrapping //! //! Everything below the application-message layer (LP packet encode/decode and //! fragment reassembly) is identical between roles, so it lives here. The diff --git a/nym-node/src/node/lp/data/handler/processing/outfox.rs b/nym-node/src/node/lp/data/handler/processing/outfox.rs index df29b239f7..556918db5b 100644 --- a/nym-node/src/node/lp/data/handler/processing/outfox.rs +++ b/nym-node/src/node/lp/data/handler/processing/outfox.rs @@ -10,7 +10,7 @@ use tracing::warn; use crate::node::lp::data::{ handler::{ error::LpDataHandlerError, - messages::{ForwardOutfoxMixMessage, GatewayMessage, MixMessage, OutfoxMixMessage}, + messages::{ForwardOutfoxMixMessage, MixMessage, NymNodeMessage, OutfoxMixMessage}, }, shared::SharedLpDataState, }; @@ -19,7 +19,7 @@ pub(crate) fn process( shared_state: &SharedLpDataState, outfox_packet: TimedPayload, metadata: OutfoxMixMessage, -) -> Result, LpDataHandlerError> { +) -> Result, LpDataHandlerError> { let TimedPayload { data: outfox_bytes, timestamp: arrival_timestamp, @@ -38,15 +38,49 @@ pub(crate) fn process( arrival_timestamp, // Outfox doesn't have mixing delays !!!!! outfox_packet.to_bytes()?, // OutfoxPacket::to_bytes is actually infallible MixMessage::Outfox(metadata), - NymNodeRoutingAddress::try_from_bytes(&next_address)?.into(), + NymNodeRoutingAddress::try_from_bytes(&next_address)?, )) } } +/// Gateway-side outfox processing. See [`super::sphinx::process_forward`] for the +/// equivalent sphinx semantics — `Client(_)` routing addresses are resolved via the +/// gateway's `ClientAddress → SocketAddr` map; `Node(_)` next-hops are forwarded as-is. +// SW Need to check for bandwidth and routing filter as well pub(crate) fn process_forward( shared_state: &SharedLpDataState, outfox_packet: TimedPayload, metadata: ForwardOutfoxMixMessage, -) -> Result, LpDataHandlerError> { +) -> Result, LpDataHandlerError> { + // let TimedPayload { + // data: outfox_bytes, + // timestamp: arrival_timestamp, + // } = outfox_packet; + + // let mut outfox_packet = OutfoxPacket::try_from(outfox_bytes.as_slice())?; + + // let key = shared_state.resolve_rotation_key(metadata.key_rotation)?; + // let next_address = outfox_packet.decode_next_layer(key.inner().as_ref())?; + + // if outfox_packet.is_final_hop() { + // warn!("Gateway received a final-hop outfox packet; clients are the final hop"); + // return Err(LpDataHandlerError::FinalHop); + // } + + // let routing_address = NymNodeRoutingAddress::try_from_bytes(&next_address)?; + // let destination_socket = match routing_address { + // NymNodeRoutingAddress::Node(socket) => socket, + // NymNodeRoutingAddress::Client(client_addr) => shared_state + // .lookup_client_udp(&client_addr) + // .ok_or(LpDataHandlerError::UnknownClientAddress { client_addr })?, + // }; + + // Ok(PipelinePayload::new( + // arrival_timestamp, + // outfox_packet.to_bytes()?, + // GatewayMessage::ForwardOutfox(metadata), + // destination_socket, + // )) + todo!() } diff --git a/nym-node/src/node/lp/data/handler/processing/sphinx.rs b/nym-node/src/node/lp/data/handler/processing/sphinx.rs index 7537372ea7..7546eef453 100644 --- a/nym-node/src/node/lp/data/handler/processing/sphinx.rs +++ b/nym-node/src/node/lp/data/handler/processing/sphinx.rs @@ -12,7 +12,7 @@ use tracing::{error, warn}; use crate::node::lp::data::{ handler::{ error::LpDataHandlerError, - messages::{ForwardSphinxMixMessage, GatewayMessage, MixMessage, SphinxMixMessage}, + messages::{ForwardSphinxMixMessage, MixMessage, NymNodeMessage, SphinxMixMessage}, }, shared::SharedLpDataState, }; @@ -21,7 +21,7 @@ pub(crate) fn process( shared_state: &SharedLpDataState, sphinx_packet: TimedPayload, metadata: SphinxMixMessage, -) -> Result, LpDataHandlerError> { +) -> Result, LpDataHandlerError> { let TimedPayload { data: sphinx_bytes, timestamp: arrival_timestamp, @@ -74,11 +74,12 @@ pub(crate) fn process( shared_state.excessive_delay_packet(); delay = shared_state.processing_config.maximum_packet_delay; } + Ok(PipelinePayload::new( arrival_timestamp + delay, next_hop_packet.to_bytes(), MixMessage::Sphinx(metadata), - NymNodeRoutingAddress::try_from(next_hop_address)?.into(), + NymNodeRoutingAddress::try_from(next_hop_address)?, )) } nym_sphinx_types::ProcessedPacketData::FinalHop { .. } => { @@ -88,10 +89,91 @@ pub(crate) fn process( } } +/// Gateway-side sphinx processing. +/// +/// Differs from [`process`] in two ways: +/// 1. Accepts a `ForwardHop` whose next address is a `Client(ClientAddress)` variant — +/// looks up the connected client's UDP endpoint in `shared_state.client_udp_map` and +/// addresses the outgoing payload to it. +/// 2. Still forwards `Node(SocketAddr)` next-hops normally (e.g., gateway-to-mixnode +/// traffic, gateway-as-relay paths). +/// +/// `FinalHop` should never reach here in the new flow (the client is the final sphinx +/// hop on the UDP path) — treat it as a misrouted packet. +/// +// SW Need to check for bandwidth and routing filter as well pub(crate) fn process_forward( shared_state: &SharedLpDataState, sphinx_packet: TimedPayload, metadata: ForwardSphinxMixMessage, -) -> Result, LpDataHandlerError> { +) -> Result, LpDataHandlerError> { + // let TimedPayload { + // data: sphinx_bytes, + // timestamp: arrival_timestamp, + // } = sphinx_packet; + + // let sphinx_packet = SphinxPacket::from_bytes(&sphinx_bytes)?; + + // let key = shared_state.resolve_rotation_key(metadata.key_rotation)?; + // let rotation_id = key.rotation_id(); + // let expanded_shared_secret = sphinx_packet + // .header + // .compute_expanded_shared_secret(key.inner().as_ref()); + + // if !shared_state.replay_protection_filter.disabled() { + // let replay_tag = expanded_shared_secret.replay_tag(); + // let Ok(replayed_packet) = shared_state + // .replay_protection_filter + // .check_and_set(rotation_id, replay_tag) + // else { + // error!("CRITICAL FAILURE: replay bloomfilter mutex poisoning!"); + // shared_state.shutdown_token.cancel(); + // Err(LpDataHandlerError::internal( + // "replay bloomfilter mutex poisoning!", + // ))? + // }; + // if replayed_packet { + // warn!( + // event = "packet.dropped.replay", + // rotation_id, "dropping replayed packet" + // ); + // Err(PacketProcessingError::PacketReplay)? + // } + // } + + // let processed_packet = sphinx_packet.process_with_expanded_secret(&expanded_shared_secret)?; + + // match processed_packet.data { + // nym_sphinx_types::ProcessedPacketData::ForwardHop { + // next_hop_packet, + // next_hop_address, + // delay, + // } => { + // let mut delay = delay.to_duration(); + // if delay > shared_state.processing_config.maximum_packet_delay { + // shared_state.excessive_delay_packet(); + // delay = shared_state.processing_config.maximum_packet_delay; + // } + + // let routing_address = NymNodeRoutingAddress::try_from(next_hop_address)?; + // let destination_socket = match routing_address { + // NymNodeRoutingAddress::Node(socket) => socket, + // NymNodeRoutingAddress::Client(client_addr) => shared_state + // .lookup_client_udp(&client_addr) + // .ok_or(LpDataHandlerError::UnknownClientAddress { client_addr })?, + // }; + + // Ok(PipelinePayload::new( + // arrival_timestamp + delay, + // next_hop_packet.to_bytes(), + // GatewayMessage::ForwardSphinx(metadata), + // destination_socket, + // )) + // } + // nym_sphinx_types::ProcessedPacketData::FinalHop { .. } => { + // warn!("Gateway received a final-hop sphinx packet; clients are the final hop"); + // Err(LpDataHandlerError::FinalHop) + // } + // } todo!() } diff --git a/nym-node/src/node/lp/data/shared.rs b/nym-node/src/node/lp/data/shared.rs index 7be9e319bf..34f388590b 100644 --- a/nym-node/src/node/lp/data/shared.rs +++ b/nym-node/src/node/lp/data/shared.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::{ - config::{Config, LpConfig, NodeModes}, + config::{Config, LpConfig}, node::{ key_rotation::active_keys::{ActiveSphinxKeys, SphinxKeyGuard}, lp::data::handler::error::LpDataHandlerError, @@ -12,6 +12,7 @@ use crate::{ }; use nym_lp_data::{PipelinePayload, fragmentation::reconstruction::MessageReconstructor}; use nym_node_metrics::{NymNodeMetrics, mixnet::PacketKind}; +use nym_sphinx_addressing::{ClientAddress, nodes::NymNodeRoutingAddress}; use nym_sphinx_framing::processing::PacketProcessingError; use nym_sphinx_params::SphinxKeyRotation; use nym_task::ShutdownToken; @@ -24,12 +25,14 @@ use tracing::{Span, warn}; #[derive(Clone, Copy)] pub(crate) struct ProcessingConfig { pub(crate) maximum_packet_delay: Duration, + pub(crate) client_forwarding_enabled: bool, } impl ProcessingConfig { pub(crate) fn new(config: &Config) -> Self { ProcessingConfig { maximum_packet_delay: config.mixnet.debug.maximum_forward_packet_delay, + client_forwarding_enabled: config.modes.expects_client_traffic(), } } } @@ -53,6 +56,7 @@ pub(crate) struct SharedLpDataState { pub metrics: NymNodeMetrics, pub shutdown_token: ShutdownToken, + // SW At some point we will need an `Option` somewhere, that will be required for gateways } impl SharedLpDataState { @@ -150,10 +154,22 @@ impl SharedLpDataState { self.metrics.mixnet.lp_routing_filter_dropped(dst) } + pub(super) fn client_forwarding_disabled_dropped(&self) { + self.metrics.mixnet.lp_client_forwarding_disabled_dropped() + } + + pub(super) fn internal_sp_routed(&self) { + self.metrics.mixnet.lp_internal_sp_routed() + } + + pub(super) fn processing_misc_error(&self) { + self.metrics.mixnet.lp_processing_misc_error() + } + pub(super) fn update_processing_metrics( &self, processing_result: &Result< - PipelinePayload, SocketAddr>, + PipelinePayload, NymNodeRoutingAddress>, LpDataHandlerError, >, ) { @@ -174,4 +190,9 @@ impl SharedLpDataState { } } } + + // SW Placeholder for SP routing while we don't have gateway state + pub(super) fn is_internal_service_provider(&self, client_address: ClientAddress) -> bool { + false + } } diff --git a/nym-node/src/node/metrics/handler/global_prometheus_updater/at_last_update.rs b/nym-node/src/node/metrics/handler/global_prometheus_updater/at_last_update.rs index d1bc866811..48e9784bff 100644 --- a/nym-node/src/node/metrics/handler/global_prometheus_updater/at_last_update.rs +++ b/nym-node/src/node/metrics/handler/global_prometheus_updater/at_last_update.rs @@ -179,6 +179,8 @@ struct LastLpMixnet { pipeline_overloaded_dropped: usize, worker_pool_overloaded_dropped: usize, egress_overloaded_dropped: usize, + client_forwarding_disabled_dropped: usize, + internal_sp_routed: usize, } impl LastLpMixnet { @@ -222,6 +224,11 @@ impl LastLpMixnet { self.egress_overloaded_dropped, previous.egress_overloaded_dropped, ), + client_forwarding_disabled_dropped_sec: per_sec( + self.client_forwarding_disabled_dropped, + previous.client_forwarding_disabled_dropped, + ), + internal_sp_routed_sec: per_sec(self.internal_sp_routed, previous.internal_sp_routed), } } } @@ -242,6 +249,8 @@ impl From<&LpMixingStats> for LastLpMixnet { pipeline_overloaded_dropped: value.pipeline_overloaded_dropped_packets(), worker_pool_overloaded_dropped: value.worker_pool_overloaded_dropped_packets(), egress_overloaded_dropped: value.egress_overloaded_dropped_packets(), + client_forwarding_disabled_dropped: value.client_forwarding_disabled_dropped(), + internal_sp_routed: value.internal_sp_routed(), } } } @@ -313,6 +322,8 @@ pub(crate) struct LpMixnetRateSinceUpdate { pub(crate) pipeline_overloaded_dropped_sec: f64, pub(crate) worker_pool_overloaded_dropped_sec: f64, pub(crate) egress_overloaded_dropped_sec: f64, + pub(crate) client_forwarding_disabled_dropped_sec: f64, + pub(crate) internal_sp_routed_sec: f64, } pub(crate) struct WireguardRateSinceUpdate { diff --git a/nym-node/src/node/metrics/handler/global_prometheus_updater/mod.rs b/nym-node/src/node/metrics/handler/global_prometheus_updater/mod.rs index 632f6411e5..0a8ff67cfb 100644 --- a/nym-node/src/node/metrics/handler/global_prometheus_updater/mod.rs +++ b/nym-node/src/node/metrics/handler/global_prometheus_updater/mod.rs @@ -159,6 +159,14 @@ impl OnUpdateMetricsHandler for PrometheusGlobalNodeMetricsRegistryUpdater { MixnetLpEgressOverloadedDropped, self.metrics.mixnet.lp.egress_overloaded_dropped_packets() as i64, ); + self.prometheus_wrapper.set( + MixnetLpClientForwardingDisabledDropped, + self.metrics.mixnet.lp.client_forwarding_disabled_dropped() as i64, + ); + self.prometheus_wrapper.set( + MixnetLpInternalSpRouted, + self.metrics.mixnet.lp.internal_sp_routed() as i64, + ); // # ENTRY self.prometheus_wrapper.set( @@ -330,6 +338,14 @@ impl OnUpdateMetricsHandler for PrometheusGlobalNodeMetricsRegistryUpdater { MixnetLpEgressOverloadedDroppedRate, diff.mixnet.lp.egress_overloaded_dropped_sec, ); + self.prometheus_wrapper.set_float( + MixnetLpClientForwardingDisabledDroppedRate, + diff.mixnet.lp.client_forwarding_disabled_dropped_sec, + ); + self.prometheus_wrapper.set_float( + MixnetLpInternalSpRoutedRate, + diff.mixnet.lp.internal_sp_routed_sec, + ); // # WIREGUARD self.prometheus_wrapper diff --git a/nym-node/src/node/mixnet/packet_forwarding/mod.rs b/nym-node/src/node/mixnet/packet_forwarding/mod.rs index dc8d4af7d4..a833607fb0 100644 --- a/nym-node/src/node/mixnet/packet_forwarding/mod.rs +++ b/nym-node/src/node/mixnet/packet_forwarding/mod.rs @@ -9,6 +9,7 @@ use nym_mixnet_client::forwarder::{ }; use nym_node_metrics::NymNodeMetrics; use nym_nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue}; +use nym_sphinx_addressing::nodes::NymNodeRoutingAddress; use nym_sphinx_forwarding::packet::MixPacket; use nym_task::ShutdownToken; use std::io; @@ -49,7 +50,11 @@ impl PacketForwarder { C: SendWithoutResponse, F: RoutingFilter, { - let next_hop = packet.next_hop_address(); + // Legacy non-LP path should never get packets adressed to clients + let NymNodeRoutingAddress::Node(next_hop) = packet.next_hop() else { + warn!("client routing address reached legacy forward_packet, this should never happen"); + return; + }; if let Err(err) = self.mixnet_client.send_without_response(packet) { if err.kind() == io::ErrorKind::WouldBlock { @@ -87,17 +92,23 @@ impl PacketForwarder { C: SendWithoutResponse, F: RoutingFilter, { - let next_hop = new_packet.packet.next_hop(); + // Legacy non-LP path should never get packets adressed to clients + let NymNodeRoutingAddress::Node(next_hop) = new_packet.packet.next_hop() else { + warn!( + event = "packet.dropped.client_variant_at_mix_forward", + next_hop = % new_packet.packet.next_hop(), + "dropping packet: client routing address reached legacy mix forwarder handle_new_packet path, this should never happen" + ); + return; + }; - if !self.routing_filter.should_route(next_hop.as_ref().ip()) { + if !self.routing_filter.should_route(next_hop.ip()) { warn!( event = "packet.dropped.routing_filter", next_hop = %next_hop, "dropping packet: egress address does not belong to any known node" ); - self.metrics - .mixnet - .egress_dropped_forward_packet(next_hop.into()); + self.metrics.mixnet.egress_dropped_forward_packet(next_hop); return; }