From c8f9959d7a90724c4aeb612d71c0768e95cf9b2a Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Wed, 20 May 2026 14:05:51 +0200 Subject: [PATCH] stub gateway pipeline --- common/nym-lp-data/src/lib.rs | 15 ++ common/nym-lp-data/src/packet/frame.rs | 6 +- .../node/lp/data/handler/messages/gateway.rs | 113 +++++++++++ .../{messages.rs => messages/mixnode.rs} | 37 ++-- .../src/node/lp/data/handler/messages/mod.rs | 8 + nym-node/src/node/lp/data/handler/mod.rs | 2 +- .../node/lp/data/handler/pipeline/gateway.rs | 161 +++++++++++++++ .../{pipeline.rs => pipeline/mixnode.rs} | 185 ++++++------------ .../src/node/lp/data/handler/pipeline/mod.rs | 12 ++ .../src/node/lp/data/handler/pipeline/wire.rs | 130 ++++++++++++ .../node/lp/data/handler/processing/outfox.rs | 10 +- .../node/lp/data/handler/processing/sphinx.rs | 10 +- nym-node/src/node/lp/data/shared.rs | 50 ++--- 13 files changed, 564 insertions(+), 175 deletions(-) create mode 100644 nym-node/src/node/lp/data/handler/messages/gateway.rs rename nym-node/src/node/lp/data/handler/{messages.rs => messages/mixnode.rs} (66%) create mode 100644 nym-node/src/node/lp/data/handler/messages/mod.rs create mode 100644 nym-node/src/node/lp/data/handler/pipeline/gateway.rs rename nym-node/src/node/lp/data/handler/{pipeline.rs => pipeline/mixnode.rs} (86%) create mode 100644 nym-node/src/node/lp/data/handler/pipeline/mod.rs create mode 100644 nym-node/src/node/lp/data/handler/pipeline/wire.rs diff --git a/common/nym-lp-data/src/lib.rs b/common/nym-lp-data/src/lib.rs index 3ec81edd09..9f411ba224 100644 --- a/common/nym-lp-data/src/lib.rs +++ b/common/nym-lp-data/src/lib.rs @@ -137,6 +137,21 @@ impl PipelineData { } } + /// Apply `op` to the options component, leaving the timestamp, data, and + /// destination unchanged. + /// + /// `No` can differ from `O`, so this also acts as a type transform. + pub fn options_transform(self, mut op: F) -> PipelineData + where + F: FnMut(Opts) -> No, + { + PipelineData { + data: self.data, + options: op(self.options), + dst: self.dst, + } + } + /// Drop the pipeline options, producing a plain addressed payload. pub fn into_addressed(self) -> AddressedTimedData { AddressedTimedData { diff --git a/common/nym-lp-data/src/packet/frame.rs b/common/nym-lp-data/src/packet/frame.rs index 72fb131a37..07b0f3d19a 100644 --- a/common/nym-lp-data/src/packet/frame.rs +++ b/common/nym-lp-data/src/packet/frame.rs @@ -15,8 +15,10 @@ pub enum LpFrameKind { Forward = 2, SphinxStream = 3, FragmentedData = 4, - SphinxPacket = 5, - OutfoxPacket = 6, + SphinxPacket = 5, // Sphinx Packet to process, delay and forward + OutfoxPacket = 6, // Outfox Packet to process, delay and forward + ForwardSphinxPacket = 7, // Sphinx Packet to immediately forward + ForwardOutfoxPacket = 8, // Outfox Packet to immediately forward #[num_enum(catch_all)] Unknown(u16), diff --git a/nym-node/src/node/lp/data/handler/messages/gateway.rs b/nym-node/src/node/lp/data/handler/messages/gateway.rs new file mode 100644 index 0000000000..e503102e43 --- /dev/null +++ b/nym-node/src/node/lp/data/handler/messages/gateway.rs @@ -0,0 +1,113 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_lp_data::packet::frame::{LpFrameAttributes, LpFrameHeader, LpFrameKind}; +use nym_node_metrics::mixnet::PacketKind; +use nym_sphinx_forwarding::packet::MixPacketFormattingError; +use nym_sphinx_params::SphinxKeyRotation; +use nym_topology::NodeId; + +use crate::node::lp::data::handler::{ + error::LpDataHandlerError, + 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. +#[derive(Debug, Clone, Copy)] +pub enum GatewayMessage { + Mix(MixMessage), + ForwardSphinx(ForwardSphinxMixMessage), + ForwardOutfox(ForwardOutfoxMixMessage), +} + +impl GatewayMessage { + pub fn from_frame_header(header: LpFrameHeader) -> Result { + match header.kind { + LpFrameKind::SphinxPacket | LpFrameKind::OutfoxPacket => { + Ok(GatewayMessage::Mix(MixMessage::from_frame_header(header)?)) + } + LpFrameKind::ForwardSphinxPacket => Ok(GatewayMessage::ForwardSphinx( + header.frame_attributes.try_into()?, + )), + LpFrameKind::ForwardOutfoxPacket => Ok(GatewayMessage::ForwardOutfox( + header.frame_attributes.try_into()?, + )), + _ => Err(LpDataHandlerError::UnsupportedLpFrameKind { typ: header.kind }), + } + } +} + +impl From for PacketKind { + fn from(value: GatewayMessage) -> Self { + match value { + GatewayMessage::Mix(msg) => msg.into(), + GatewayMessage::ForwardSphinx(_) => PacketKind::LpSphinx, + GatewayMessage::ForwardOutfox(_) => PacketKind::LpOutfox, + } + } +} + +impl From for GatewayMessage { + fn from(value: MixMessage) -> Self { + GatewayMessage::Mix(value) + } +} + +impl From for LpFrameHeader { + fn from(value: GatewayMessage) -> Self { + match value { + GatewayMessage::Mix(msg) => msg.into(), + GatewayMessage::ForwardSphinx(msg) => { + LpFrameHeader::new(LpFrameKind::ForwardSphinxPacket, msg) + } + GatewayMessage::ForwardOutfox(msg) => { + LpFrameHeader::new(LpFrameKind::ForwardOutfoxPacket, msg) + } + } + } +} + +#[derive(Debug, Clone, Copy)] +pub struct ForwardSphinxMixMessage { + pub key_rotation: SphinxKeyRotation, + pub next_hop: NodeId, +} + +impl TryFrom for ForwardSphinxMixMessage { + type Error = LpDataHandlerError; + + fn try_from(value: LpFrameAttributes) -> Result { + let key_rotation = value[0] + .try_into() + .map_err(MixPacketFormattingError::InvalidKeyRotation)?; + // SAFETY : slice to array conversion with correct size + #[allow(clippy::unwrap_used)] + let next_hop = NodeId::from_be_bytes(value[1..5].try_into().unwrap()); + Ok(ForwardSphinxMixMessage { + key_rotation, + next_hop, + }) + } +} + +impl From for LpFrameAttributes { + fn from(value: ForwardSphinxMixMessage) -> Self { + let mut attrs = [0; 14]; + attrs[0] = value.key_rotation as u8; + attrs[1..5].copy_from_slice(&value.next_hop.to_be_bytes()); + attrs + } +} + +impl From for SphinxMixMessage { + fn from(value: ForwardSphinxMixMessage) -> Self { + SphinxMixMessage { + key_rotation: value.key_rotation, + } + } +} + +// For now there are no differences. We can augment this variant when we will need it +pub type ForwardOutfoxMixMessage = ForwardSphinxMixMessage; diff --git a/nym-node/src/node/lp/data/handler/messages.rs b/nym-node/src/node/lp/data/handler/messages/mixnode.rs similarity index 66% rename from nym-node/src/node/lp/data/handler/messages.rs rename to nym-node/src/node/lp/data/handler/messages/mixnode.rs index 1b595902eb..79755b834f 100644 --- a/nym-node/src/node/lp/data/handler/messages.rs +++ b/nym-node/src/node/lp/data/handler/messages/mixnode.rs @@ -2,18 +2,42 @@ // SPDX-License-Identifier: Apache-2.0 use nym_lp_data::packet::frame::{LpFrameAttributes, LpFrameHeader, LpFrameKind}; +use nym_node_metrics::mixnet::PacketKind; use nym_sphinx_forwarding::packet::MixPacketFormattingError; use nym_sphinx_params::SphinxKeyRotation; use crate::node::lp::data::handler::error::LpDataHandlerError; -/// Message types supported by mixnodes +/// Message types supported by mixnodes. #[derive(Debug, Clone, Copy)] pub enum MixMessage { Sphinx(SphinxMixMessage), Outfox(OutfoxMixMessage), } +impl MixMessage { + pub fn from_frame_header(header: LpFrameHeader) -> Result { + match header.kind { + LpFrameKind::SphinxPacket => { + Ok(MixMessage::Sphinx(header.frame_attributes.try_into()?)) + } + LpFrameKind::OutfoxPacket => { + Ok(MixMessage::Outfox(header.frame_attributes.try_into()?)) + } + _ => Err(LpDataHandlerError::UnsupportedLpFrameKind { typ: header.kind }), + } + } +} + +impl From for PacketKind { + fn from(value: MixMessage) -> Self { + match value { + MixMessage::Sphinx(_) => PacketKind::LpSphinx, + MixMessage::Outfox(_) => PacketKind::LpOutfox, + } + } +} + impl From for LpFrameHeader { fn from(value: MixMessage) -> Self { match value { @@ -22,17 +46,6 @@ impl From for LpFrameHeader { } } } -impl TryFrom for MixMessage { - type Error = LpDataHandlerError; - - fn try_from(value: LpFrameHeader) -> Result { - match value.kind { - LpFrameKind::SphinxPacket => Ok(MixMessage::Sphinx(value.frame_attributes.try_into()?)), - LpFrameKind::OutfoxPacket => Ok(MixMessage::Outfox(value.frame_attributes.try_into()?)), - other => Err(LpDataHandlerError::UnsupportedLpFrameKind { typ: other })?, - } - } -} #[derive(Debug, Clone, Copy)] pub struct 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 new file mode 100644 index 0000000000..138e871762 --- /dev/null +++ b/nym-node/src/node/lp/data/handler/messages/mod.rs @@ -0,0 +1,8 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +mod gateway; +mod mixnode; + +pub use gateway::{ForwardOutfoxMixMessage, ForwardSphinxMixMessage, GatewayMessage}; +pub use mixnode::{MixMessage, OutfoxMixMessage, SphinxMixMessage}; diff --git a/nym-node/src/node/lp/data/handler/mod.rs b/nym-node/src/node/lp/data/handler/mod.rs index 09cab0ca00..04bc407d56 100644 --- a/nym-node/src/node/lp/data/handler/mod.rs +++ b/nym-node/src/node/lp/data/handler/mod.rs @@ -19,7 +19,7 @@ use crate::node::lp::data::PACKET_BUFFER_SIZE; use crate::node::lp::data::handler::pipeline::MixnodeDataPipeline; use crate::node::lp::data::shared::SharedLpDataState; use nym_lp_data::AddressedTimedData; -use nym_lp_data::mixnodes::traits::MixnodeProcessingPipeline; +use nym_lp_data::nymnodes::traits::NymNodeProcessingPipeline; use nym_lp_data::packet::{EncryptedLpPacket, MalformedLpPacketError}; use nym_metrics::inc; use rand::rngs::OsRng; diff --git a/nym-node/src/node/lp/data/handler/pipeline/gateway.rs b/nym-node/src/node/lp/data/handler/pipeline/gateway.rs new file mode 100644 index 0000000000..904994b935 --- /dev/null +++ b/nym-node/src/node/lp/data/handler/pipeline/gateway.rs @@ -0,0 +1,161 @@ +// 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.rs b/nym-node/src/node/lp/data/handler/pipeline/mixnode.rs similarity index 86% rename from nym-node/src/node/lp/data/handler/pipeline.rs rename to nym-node/src/node/lp/data/handler/pipeline/mixnode.rs index 4da6b062ef..a1012fb344 100644 --- a/nym-node/src/node/lp/data/handler/pipeline.rs +++ b/nym-node/src/node/lp/data/handler/pipeline/mixnode.rs @@ -9,64 +9,50 @@ use nym_lp_data::{ Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline, WireWrappingPipeline, }, - fragmentation::fragment::fragment_lp_message, - mixnodes::traits::MixnodeProcessingPipeline, - packet::{ - EncryptedLpPacket, LpFrame, LpHeader, LpPacket, MalformedLpPacketError, - frame::{LpFrameHeader, LpFrameKind}, - }, + nymnodes::traits::NymNodeProcessingPipeline, + packet::{EncryptedLpPacket, LpFrame, LpHeader, MalformedLpPacketError, frame::LpFrameHeader}, }; use rand::Rng; use tracing::warn; use crate::node::{ lp::data::{ - handler::{messages::MixMessage, processing}, + handler::{messages::MixMessage, pipeline::wire::WirePipeline, processing}, shared::SharedLpDataState, }, routing_filter::RoutingFilter, }; -pub struct MixnodeDataPipeline -where - R: Rng, -{ - /// Shared data state +pub(crate) struct MixnodeDataPipeline { state: Arc, - rng: R, + wire: WirePipeline, } -impl MixnodeDataPipeline -where - R: Rng, -{ - pub fn new(state: Arc, rng: R) -> Self { - Self { state, rng } +impl MixnodeDataPipeline { + pub(crate) fn new(state: Arc, rng: R) -> Self { + Self { + state: state.clone(), + wire: WirePipeline::new(state, rng), + } } -} -// Mixing logic -impl MixnodeProcessingPipeline - for MixnodeDataPipeline -where - R: Rng, -{ - fn mix( - &mut self, + // Processing logic so gateways can reuse them + pub fn process_mix_packet( + shared_state: &SharedLpDataState, message_kind: MixMessage, payload: TimedPayload, _: Instant, ) -> Vec> { let processing_result = match message_kind { MixMessage::Sphinx(metadata) => { - processing::sphinx::process(&self.state, payload, metadata) + processing::sphinx::process(shared_state, payload, metadata) } MixMessage::Outfox(metadata) => { - processing::outfox::process(&self.state, payload, metadata) + processing::outfox::process(shared_state, payload, metadata) } }; - self.state.update_processing_metrics(&processing_result); + shared_state.update_processing_metrics(&processing_result); let packet_to_forward = match processing_result { Ok(packet) => packet, @@ -77,13 +63,13 @@ where }; let next_hop = packet_to_forward.dst; - if !self.state.routing_filter.should_route(next_hop.ip()) { + 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" ); - self.state.routing_filter_dropped(next_hop); + shared_state.routing_filter_dropped(next_hop); Vec::new() } else { vec![packet_to_forward] @@ -91,12 +77,25 @@ where } } -impl Framing for MixnodeDataPipeline -where - R: Rng, +// Mixing logic +impl + NymNodeProcessingPipeline + for MixnodeDataPipeline { - type Frame = LpFrame; + fn mix( + &mut self, + message_kind: MixMessage, + payload: TimedPayload, + timestamp: Instant, + ) -> Vec> { + MixnodeDataPipeline::::process_mix_packet(&self.state, message_kind, payload, timestamp) + } +} +// ============== Wire wrap/unwrap: pure delegation to WirePipeline ============== + +impl Framing for MixnodeDataPipeline { + type Frame = LpFrame; const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE; fn to_frame( @@ -108,54 +107,32 @@ where header: payload.options.into(), content: payload.data.data.into(), }; - - let output_frames = if frame.len() > frame_size { - fragment_lp_message(&mut self.rng, frame, frame_size) - .into_iter() - .map(|f| f.into_lp_frame()) - .collect() - } else { - vec![frame] - }; - - output_frames - .into_iter() - .map(|f| AddressedTimedData::new_addressed(payload.data.timestamp, f, payload.dst)) - .collect() + self.wire + .message_to_frame(payload.data.timestamp, frame, payload.dst, frame_size) } } -impl Transport for MixnodeDataPipeline -where - R: Rng, -{ +impl Transport for MixnodeDataPipeline { type Frame = LpFrame; - const OVERHEAD_SIZE: usize = LpHeader::SIZE; fn to_transport_packet( &mut self, frame: AddressedTimedData, ) -> AddressedTimedData { - // Here be LP encryption. For not, just wrap into an EncryptedLpPacket, we don't care at reception anyway - frame.data_transform(|f| LpPacket::new(LpHeader::new(0, 0, 0), f).encode()) + self.wire.frame_to_packet(frame) } } -impl WireWrappingPipeline +impl WireWrappingPipeline for MixnodeDataPipeline -where - R: Rng, { fn packet_size(&self) -> usize { nym_lp_data::packet::MTU } } -impl TransportUnwrap for MixnodeDataPipeline -where - R: Rng, -{ +impl TransportUnwrap for MixnodeDataPipeline { type Frame = LpFrame; type Error = MalformedLpPacketError; @@ -164,79 +141,31 @@ where packet: EncryptedLpPacket, timestamp: Instant, ) -> Result, Self::Error> { - // Here be LP decryption. For now we do as is it's not encrypted - let lp_packet = LpPacket::decode(packet).inspect_err(|_| { - self.state.malformed_packet(); - })?; - Ok(TimedData { - timestamp, - data: lp_packet.into_frame(), - }) + self.wire.packet_to_frame(packet, timestamp) } } -impl FramingUnwrap for MixnodeDataPipeline -where - R: Rng, -{ +impl FramingUnwrap for MixnodeDataPipeline { type Frame = LpFrame; + fn frame_to_message( &mut self, frame: TimedData, ) -> Option<(TimedPayload, MixMessage)> { - let reassembled_frame = if frame.data.kind() == LpFrameKind::FragmentedData { - let fragment = frame - .data - .try_into() - .inspect_err(|e| { - tracing::error!("Failed to recover a fragment : {e}"); - self.state.malformed_packet(); - }) - .ok()?; - let message = self - .state - .message_reconstructor - .insert_new_fragment(fragment, frame.timestamp)? - .inspect_err(|e| { - tracing::error!("Failed to recover a frame : {e}"); - self.state.malformed_packet(); - }) - .ok()?; - - TimedData::new(frame.timestamp, message) - } else { - frame - }; - - match reassembled_frame.data.kind() { - LpFrameKind::FragmentedData => { - warn!( - "Fragmented data inside fragmented data, it shouldn't happen. Dropping the message" - ); - None - } - _ => { - let message_kind = reassembled_frame - .data - .header - .try_into() - .inspect_err(|e| warn!("{e}")) - .ok()?; - self.state.message_received(&message_kind); - Some(( - TimedPayload::new( - reassembled_frame.timestamp, - reassembled_frame.data.content.to_vec(), - ), - message_kind, - )) - } - } + let reassembled = self.wire.frame_to_maybe_message(frame)?; + let message_kind = MixMessage::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 MixnodeDataPipeline where - R: Rng +impl WireUnwrappingPipeline + for MixnodeDataPipeline { } @@ -251,7 +180,7 @@ mod tests { use nym_lp_data::common::traits::WireWrappingPipeline; use nym_lp_data::fragmentation::fragment::fragment_lp_message; use nym_lp_data::fragmentation::reconstruction::MessageReconstructor; - use nym_lp_data::mixnodes::traits::MixnodeProcessingPipeline; + use nym_lp_data::nymnodes::traits::NymNodeProcessingPipeline; use nym_lp_data::packet::{ EncryptedLpPacket, LpFrame, LpHeader, LpPacket, OuterHeader, frame::LpFrameHeader, version, }; diff --git a/nym-node/src/node/lp/data/handler/pipeline/mod.rs b/nym-node/src/node/lp/data/handler/pipeline/mod.rs new file mode 100644 index 0000000000..bf0e5f45b8 --- /dev/null +++ b/nym-node/src/node/lp/data/handler/pipeline/mod.rs @@ -0,0 +1,12 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! Per-role data pipelines (mixnode, gateway). The wire wrapping/unwrapping +//! shared between them lives in [`wire`]. + +mod gateway; +mod mixnode; +mod wire; + +pub(crate) use gateway::GatewayDataPipeline; +pub(crate) use mixnode::MixnodeDataPipeline; diff --git a/nym-node/src/node/lp/data/handler/pipeline/wire.rs b/nym-node/src/node/lp/data/handler/pipeline/wire.rs new file mode 100644 index 0000000000..c0960584db --- /dev/null +++ b/nym-node/src/node/lp/data/handler/pipeline/wire.rs @@ -0,0 +1,130 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +//! Wire wrapping / unwrapping shared by the mixnode and gateway data pipelines. +//! +//! Everything below the application-message layer (LP packet encode/decode and +//! fragment reassembly) is identical between roles, so it lives here. The +//! per-role pipelines embed a [`WirePipeline`] and delegate their +//! [`Framing`]/[`Transport`]/[`*Unwrap`] trait methods to it. +//! +//! [`Framing`]: nym_lp_data::common::traits::Framing +//! [`Transport`]: nym_lp_data::common::traits::Transport + +use std::{net::SocketAddr, sync::Arc, time::Instant}; + +use nym_lp_data::{ + AddressedTimedData, TimedData, + fragmentation::fragment::fragment_lp_message, + packet::{ + EncryptedLpPacket, LpFrame, LpHeader, LpPacket, MalformedLpPacketError, frame::LpFrameKind, + }, +}; +use rand::Rng; +use tracing::warn; + +use crate::node::lp::data::shared::SharedLpDataState; + +/// Wire-layer pipeline: handles LP packet encode/decode and fragment +/// reassembly. Has no knowledge of the application message type carried in +/// the frame. +pub struct WirePipeline { + state: Arc, + rng: R, +} + +impl WirePipeline { + pub fn new(state: Arc, rng: R) -> Self { + Self { state, rng } + } + + /// Wrap an [`LpFrame`] into an [`EncryptedLpPacket`] for the wire. + pub fn frame_to_packet( + &mut self, + frame: AddressedTimedData, + ) -> AddressedTimedData { + // Here be LP encryption. For now, just wrap into an EncryptedLpPacket; we don't care at reception anyway + frame.data_transform(|f| LpPacket::new(LpHeader::new(0, 0, 0), f).encode()) + } + + /// Unwrap an [`EncryptedLpPacket`] off the wire into an [`LpFrame`]. + pub fn packet_to_frame( + &mut self, + packet: EncryptedLpPacket, + timestamp: Instant, + ) -> Result, MalformedLpPacketError> { + // Here be LP decryption. For now we do as is, it's not encrypted + let lp_packet = LpPacket::decode(packet).inspect_err(|_| { + self.state.malformed_packet(); + })?; + Ok(TimedData { + timestamp, + data: lp_packet.into_frame(), + }) + } + + /// Wrap an [`LpFrame`] into one or more addressed frames, fragmenting it + /// if its serialized length would exceed `frame_size`. + pub fn message_to_frame( + &mut self, + timestamp: Instant, + frame: LpFrame, + dst: SocketAddr, + frame_size: usize, + ) -> Vec> { + let output_frames = if frame.len() > frame_size { + fragment_lp_message(&mut self.rng, frame, frame_size) + .into_iter() + .map(|f| f.into_lp_frame()) + .collect() + } else { + vec![frame] + }; + + output_frames + .into_iter() + .map(|f| AddressedTimedData::new_addressed(timestamp, f, dst)) + .collect() + } + + /// If the frame carries a fragment, attempt reassembly; otherwise return + /// the frame as-is. Returns `None` when more fragments are needed or + /// reassembly fails. The returned frame is guaranteed not to be a + /// fragment. + pub fn frame_to_maybe_message( + &mut self, + frame: TimedData, + ) -> Option> { + let reassembled = if frame.data.kind() == LpFrameKind::FragmentedData { + let fragment = frame + .data + .try_into() + .inspect_err(|e| { + tracing::error!("Failed to recover a fragment : {e}"); + self.state.malformed_packet(); + }) + .ok()?; + let message = self + .state + .message_reconstructor + .insert_new_fragment(fragment, frame.timestamp)? + .inspect_err(|e| { + tracing::error!("Failed to recover a frame : {e}"); + self.state.malformed_packet(); + }) + .ok()?; + TimedData::new(frame.timestamp, message) + } else { + frame + }; + + if reassembled.data.kind() == LpFrameKind::FragmentedData { + warn!( + "Fragmented data inside fragmented data, it shouldn't happen. Dropping the message" + ); + None + } else { + Some(reassembled) + } + } +} 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 2de1dbd1aa..df29b239f7 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::{MixMessage, OutfoxMixMessage}, + messages::{ForwardOutfoxMixMessage, GatewayMessage, MixMessage, OutfoxMixMessage}, }, shared::SharedLpDataState, }; @@ -42,3 +42,11 @@ pub(crate) fn process( )) } } + +pub(crate) fn process_forward( + shared_state: &SharedLpDataState, + outfox_packet: TimedPayload, + metadata: ForwardOutfoxMixMessage, +) -> Result, LpDataHandlerError> { + 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 1995b773f6..7537372ea7 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::{MixMessage, SphinxMixMessage}, + messages::{ForwardSphinxMixMessage, GatewayMessage, MixMessage, SphinxMixMessage}, }, shared::SharedLpDataState, }; @@ -87,3 +87,11 @@ pub(crate) fn process( } } } + +pub(crate) fn process_forward( + shared_state: &SharedLpDataState, + sphinx_packet: TimedPayload, + metadata: ForwardSphinxMixMessage, +) -> Result, LpDataHandlerError> { + todo!() +} diff --git a/nym-node/src/node/lp/data/shared.rs b/nym-node/src/node/lp/data/shared.rs index 708e8824cd..7be9e319bf 100644 --- a/nym-node/src/node/lp/data/shared.rs +++ b/nym-node/src/node/lp/data/shared.rs @@ -1,26 +1,25 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -use crate::config::Config; -use crate::config::LpConfig; -use crate::node::key_rotation::active_keys::ActiveSphinxKeys; -use crate::node::key_rotation::active_keys::SphinxKeyGuard; -use crate::node::lp::data::handler::error::LpDataHandlerError; -use crate::node::lp::data::handler::messages::MixMessage; -use crate::node::replay_protection::bloomfilter::ReplayProtectionBloomfilters; -use crate::node::routing_filter::network_filter::NetworkRoutingFilter; -use nym_lp_data::PipelinePayload; -use nym_lp_data::fragmentation::reconstruction::MessageReconstructor; -use nym_node_metrics::NymNodeMetrics; -use nym_node_metrics::mixnet::PacketKind; +use crate::{ + config::{Config, LpConfig, NodeModes}, + node::{ + key_rotation::active_keys::{ActiveSphinxKeys, SphinxKeyGuard}, + lp::data::handler::error::LpDataHandlerError, + replay_protection::bloomfilter::ReplayProtectionBloomfilters, + routing_filter::network_filter::NetworkRoutingFilter, + }, +}; +use nym_lp_data::{PipelinePayload, fragmentation::reconstruction::MessageReconstructor}; +use nym_node_metrics::{NymNodeMetrics, mixnet::PacketKind}; use nym_sphinx_framing::processing::PacketProcessingError; use nym_sphinx_params::SphinxKeyRotation; use nym_task::ShutdownToken; -use std::net::SocketAddr; -use std::time::Duration; -use std::time::Instant; -use tracing::Span; -use tracing::warn; +use std::{ + net::SocketAddr, + time::{Duration, Instant}, +}; +use tracing::{Span, warn}; #[derive(Clone, Copy)] pub(crate) struct ProcessingConfig { @@ -56,13 +55,6 @@ pub(crate) struct SharedLpDataState { pub shutdown_token: ShutdownToken, } -fn message_kind_to_packet_kind(message_kind: &MixMessage) -> PacketKind { - match message_kind { - MixMessage::Sphinx { .. } => PacketKind::LpSphinx, - MixMessage::Outfox { .. } => PacketKind::LpOutfox, - } -} - impl SharedLpDataState { pub(crate) fn new( config: &Config, @@ -123,10 +115,8 @@ impl SharedLpDataState { self.metrics.mixnet.lp_malformed_packet() } - pub(super) fn message_received(&self, message_kind: &MixMessage) { - self.metrics - .mixnet - .lp_message_received(message_kind_to_packet_kind(message_kind)) + pub(super) fn message_received(&self, message_kind: impl Into) { + self.metrics.mixnet.lp_message_received(message_kind.into()) } pub(super) fn packet_forwarded(&self, dst: SocketAddr) { @@ -163,7 +153,7 @@ impl SharedLpDataState { pub(super) fn update_processing_metrics( &self, processing_result: &Result< - PipelinePayload, + PipelinePayload, SocketAddr>, LpDataHandlerError, >, ) { @@ -171,7 +161,7 @@ impl SharedLpDataState { Ok(packet) => { self.metrics .mixnet - .lp_processed_message(message_kind_to_packet_kind(&packet.options)); + .lp_processed_message(packet.options.clone().into()); } Err(LpDataHandlerError::PacketProcessingError(PacketProcessingError::PacketReplay)) => { self.metrics.mixnet.lp_processing_replayed_packet();