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

different fragmentation

Browse Source
This commit is contained in:
Simon Wicky
2026-05-19 11:31:15 +02:00
parent 804b17517f
commit c0a8f97a20
9 changed files with 303 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nym-lp-data/src/fragmentation/fragment.rs
+41 -57
View File
@@ -11,32 +11,14 @@ use crate::{
#[derive(Copy, Clone, Debug, Hash, PartialEq, Eq)]
/// Key for reconstruction hashmap
pub struct FragmentHashKey(u64, LpFrameKind);
pub struct FragmentHashKey(LpFrameKind, u64);
impl From<(u64, LpFrameKind)> for FragmentHashKey {
fn from(value: (u64, LpFrameKind)) -> Self {
impl From<(LpFrameKind, u64)> for FragmentHashKey {
fn from(value: (LpFrameKind, u64)) -> Self {
FragmentHashKey(value.0, value.1)
}
}
#[derive(Copy, Clone, Debug)]
pub struct FragmentMetadata(LpFrameKind, [u8; 4]);
impl FragmentMetadata {
pub fn kind(&self) -> LpFrameKind {
self.0
}
pub fn metadata(&self) -> [u8; 4] {
self.1
}
}
impl From<(LpFrameKind, [u8; 4])> for FragmentMetadata {
fn from(value: (LpFrameKind, [u8; 4])) -> Self {
FragmentMetadata(value.0, value.1)
}
}
#[derive(PartialEq, Clone, Debug)]
pub struct FragmentHeader {
/// ID associated to this particular `Fragment`.
@@ -49,18 +31,26 @@ pub struct FragmentHeader {
/// Index of this fragment, in (0..total_fragments)
current_fragment: u8,
/// Additional metadata, parsing depends on the frame_kind of the fragment.
kind_metadata: [u8; 4],
/// The kind of the inner frame
next_frame_kind: LpFrameKind,
reserved: [u8; 2],
}
impl FragmentHeader {
// It's up to the caller to make sure values are valid
fn new(id: u64, total_fragments: u8, current_fragment: u8, kind_metadata: [u8; 4]) -> Self {
fn new(
id: u64,
total_fragments: u8,
current_fragment: u8,
next_frame_kind: LpFrameKind,
) -> Self {
FragmentHeader {
id,
total_fragments,
current_fragment,
kind_metadata,
next_frame_kind,
reserved: [0; 2],
}
}
}
@@ -71,7 +61,8 @@ impl From<FragmentHeader> for LpFrameAttributes {
buf[0..8].copy_from_slice(&value.id.to_be_bytes());
buf[8] = value.total_fragments;
buf[9] = value.current_fragment;
buf[10..14].copy_from_slice(&value.kind_metadata);
buf[10..12].copy_from_slice(&u16::to_be_bytes(value.next_frame_kind.into()));
buf[12..14].copy_from_slice(&value.reserved);
buf
}
}
@@ -79,27 +70,28 @@ impl From<FragmentHeader> for LpFrameAttributes {
impl TryFrom<LpFrameAttributes> for FragmentHeader {
type Error = FragmentationError;
fn try_from(value: LpFrameAttributes) -> Result<Self, Self::Error> {
// SAFETY : Three conversion from slices to arrays with correct size
let total_fragments = value[8];
let current_fragment = value[9];
if current_fragment >= total_fragments {
return Err(FragmentationError::FragmentIndexOutOfBounds);
}
// SAFETY : Three conversion from slices to arrays with correct size
Ok(FragmentHeader {
#[allow(clippy::unwrap_used)]
id: u64::from_be_bytes(value[0..8].try_into().unwrap()),
total_fragments,
current_fragment,
#[allow(clippy::unwrap_used)]
kind_metadata: value[10..14].try_into().unwrap(),
next_frame_kind: u16::from_be_bytes(value[10..12].try_into().unwrap()).into(),
#[allow(clippy::unwrap_used)]
reserved: value[12..14].try_into().unwrap(),
})
}
}
#[derive(PartialEq, Clone, Debug)]
pub struct Fragment {
frame_kind: LpFrameKind,
header: FragmentHeader,
payload: Vec<u8>,
}
@@ -111,19 +103,17 @@ impl Fragment {
id: u64,
total_fragments: u8,
current_fragment: u8,
kind_metadata: [u8; 4],
frame_kind: LpFrameKind,
next_frame_kind: LpFrameKind,
) -> Self {
let header = FragmentHeader::new(id, total_fragments, current_fragment, kind_metadata);
let header = FragmentHeader::new(id, total_fragments, current_fragment, next_frame_kind);
Fragment {
header,
payload: payload.to_vec(),
frame_kind,
}
}
pub fn into_lp_frame(self) -> LpFrame {
LpFrame::new_with_attributes(self.frame_kind, self.header, self.payload)
LpFrame::new_with_attributes(LpFrameKind::FragmentedData, self.header, self.payload)
}
/// Extracts id of this `Fragment`.
@@ -142,16 +132,12 @@ impl Fragment {
self.header.current_fragment
}
pub fn frame_kind(&self) -> LpFrameKind {
self.frame_kind
}
pub fn kind_metadata(&self) -> [u8; 4] {
self.header.kind_metadata
pub fn next_frame_kind(&self) -> LpFrameKind {
self.header.next_frame_kind
}
pub fn hash_key(&self) -> FragmentHashKey {
(self.header.id, self.frame_kind).into()
(self.header.next_frame_kind, self.header.id).into()
}
/// Consumes `self` to obtain payload (i.e. part of original message) associated with this
@@ -164,45 +150,43 @@ impl Fragment {
impl TryFrom<LpFrame> for Fragment {
type Error = FragmentationError;
fn try_from(value: LpFrame) -> Result<Self, Self::Error> {
if value.kind().is_fragmented() {
Ok(Fragment {
match value.kind() {
LpFrameKind::FragmentedData => Ok(Fragment {
header: value.header.frame_attributes.try_into()?,
payload: value.content.to_vec(),
frame_kind: value.kind(),
})
} else {
Err(FragmentationError::InvalidFrameKind)
}),
_ => Err(FragmentationError::InvalidFrameKind),
}
}
}
/// Splits a payload into multiple `Fragment`s
/// Splits an LpFrame into multiple `Fragment`s
/// This is meant to be used during Framing, not Chunking. This way we can ensure it fits in less than 255 fragments
pub fn fragment_payload<R: rand::Rng>(
pub fn fragment_lp_message<R: rand::Rng>(
rng: &mut R,
message: &[u8],
fragment_metadata: FragmentMetadata,
message: LpFrame,
fragment_payload_size: usize,
) -> Vec<Fragment> {
debug_assert!(message.len() <= u8::MAX as usize * fragment_payload_size);
debug_assert!(fragment_metadata.kind().is_fragmented());
let message_kind = message.kind();
let message_bytes = message.to_bytes();
let id = rng.r#gen();
let num_fragments = (message.len() as f64 / fragment_payload_size as f64).ceil() as u8;
let num_fragments = (message_bytes.len() as f64 / fragment_payload_size as f64).ceil() as u8;
let mut fragments = Vec::with_capacity(num_fragments as usize);
for i in 0..num_fragments as usize {
let lb = i * fragment_payload_size;
let ub = usize::min(message.len(), (i + 1) * fragment_payload_size);
let ub = usize::min(message_bytes.len(), (i + 1) * fragment_payload_size);
fragments.push(Fragment::new(
&message[lb..ub],
&message_bytes[lb..ub],
id,
num_fragments,
i as u8,
fragment_metadata.metadata(),
fragment_metadata.kind(),
message_kind,
))
}
common/nym-lp-data/src/fragmentation/reconstruction.rs
+50 -63
View File
@@ -1,7 +1,8 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::fragmentation::fragment::{Fragment, FragmentHashKey, FragmentMetadata};
use crate::fragmentation::fragment::{Fragment, FragmentHashKey};
use crate::packet::{LpFrame, MalformedLpPacketError};
use dashmap::DashMap;
use dashmap::mapref::entry::Entry;
@@ -155,36 +156,30 @@ where
}
/// Insert `fragment` into the buffer for its message and, if it was the
/// last outstanding fragment, return the reassembled message bytes
/// alongside the original metadata.
/// last outstanding fragment, return the reassembled LpFrame
///
/// Stale incomplete messages are evicted on every call.
pub fn insert_new_fragment(
&self,
fragment: Fragment,
timestamp: Ts,
) -> Option<(Vec<u8>, FragmentMetadata)> {
) -> Option<Result<LpFrame, MalformedLpPacketError>> {
let key = fragment.hash_key();
let total_fragments = fragment.total_fragments();
// Captured before insertion since `insert_fragment` consumes
// `fragment`. The kind is also part of `key`, so reading it back
// from any of the buffer's slots would be equivalent.
let metadata = (fragment.frame_kind(), fragment.kind_metadata()).into();
let maybe_message = match self.in_flight_messages.entry(key) {
Entry::Occupied(mut entry) => {
entry.get_mut().insert_fragment(fragment, timestamp.clone());
entry
.get()
.is_complete
.then(|| (entry.remove().into_message(), metadata))
.then(|| LpFrame::decode(&entry.remove().into_message()))
}
Entry::Vacant(entry) => {
let mut buf = MessageBuffer::new(total_fragments, timestamp.clone());
buf.insert_fragment(fragment, timestamp.clone());
if buf.is_complete {
Some((buf.into_message(), metadata))
Some(LpFrame::decode(&buf.into_message()))
} else {
entry.insert(buf);
None
@@ -229,14 +224,14 @@ mod tests {
#![allow(clippy::unwrap_used)]
use super::*;
use crate::fragmentation::fragment::fragment_payload;
use crate::fragmentation::fragment::fragment_lp_message;
use crate::packet::LpFrame;
use crate::packet::frame::LpFrameKind;
use rand::SeedableRng;
use rand::rngs::StdRng;
const SPHINX: LpFrameKind = LpFrameKind::FragmentedSphinxPacket;
const OUTFOX: LpFrameKind = LpFrameKind::FragmentedOutfoxPacket;
const SPHINX: LpFrameKind = LpFrameKind::SphinxPacket;
const OUTFOX: LpFrameKind = LpFrameKind::OutfoxPacket;
/// Build a `Fragment` with explicit header values via the public
/// `LpFrame` round-trip, so tests can craft duplicates, out-of-order
@@ -245,20 +240,21 @@ mod tests {
id: u64,
total_fragments: u8,
current_fragment: u8,
kind: LpFrameKind,
inner_kind: LpFrameKind,
payload: Vec<u8>,
) -> Fragment {
let mut attrs = [0u8; 14];
attrs[0..8].copy_from_slice(&id.to_be_bytes());
attrs[8] = total_fragments;
attrs[9] = current_fragment;
let frame = LpFrame::new_with_attributes(kind, attrs, payload);
attrs[10..12].copy_from_slice(&u16::to_be_bytes(inner_kind.into()));
let frame = LpFrame::new_with_attributes(LpFrameKind::FragmentedData, attrs, payload);
Fragment::try_from(frame).unwrap()
}
fn split(message: &[u8], kind: LpFrameKind, fragment_size: usize) -> Vec<Fragment> {
fn split(message: LpFrame, fragment_size: usize) -> Vec<Fragment> {
let mut rng = StdRng::seed_from_u64(0xdead_beef);
fragment_payload(&mut rng, message, (kind, [0; 4]).into(), fragment_size)
fragment_lp_message(&mut rng, message, fragment_size)
}
// ---------- MessageBuffer ----------
@@ -331,64 +327,55 @@ mod tests {
#[test]
fn reconstructor_round_trip_single_fragment_message() {
let message = b"small";
let mut fragments = split(message, SPHINX, 64);
let message = LpFrame::new(SPHINX, b"small".as_slice());
let mut fragments = split(message.clone(), 64);
assert_eq!(fragments.len(), 1);
let rec = MessageReconstructor::<u64, u64>::new(60);
let out = rec.insert_new_fragment(fragments.pop().unwrap(), 0);
let (bytes, metadata) = out.expect("single fragment must complete the message");
assert_eq!(bytes, message);
assert_eq!(metadata.kind(), SPHINX);
let recovered_frame = out
.expect("single fragment must complete the message")
.unwrap();
assert_eq!(recovered_frame, message);
}
#[test]
fn reconstructor_round_trip_multi_fragment_message() {
let message: Vec<u8> = (0u8..=200).collect();
let fragments = split(&message, SPHINX, 16);
let message = LpFrame::new(SPHINX, (0u8..=200).collect::<Vec<_>>());
let fragments = split(message.clone(), 16);
assert!(fragments.len() > 1);
let rec = MessageReconstructor::<u64, u64>::new(60);
let total = fragments.len();
let mut completed = None;
let mut out = None;
for (i, f) in fragments.into_iter().enumerate() {
let out = rec.insert_new_fragment(f, i as u64);
out = rec.insert_new_fragment(f, i as u64);
if i + 1 < total {
assert!(out.is_none(), "premature completion at fragment {i}");
} else {
completed = out;
}
}
let (bytes, metadata) = completed.expect("last fragment must complete the message");
assert_eq!(bytes, message);
assert_eq!(metadata.kind(), SPHINX);
let recovered_frame = out
.expect("last fragment must complete the message")
.unwrap();
assert_eq!(recovered_frame, message);
}
#[test]
fn reconstructor_handles_out_of_order_arrival() {
let message: Vec<u8> = (0u8..50).collect();
let mut fragments = split(&message, SPHINX, 8);
let message = LpFrame::new(SPHINX, (0u8..=200).collect::<Vec<_>>());
let mut fragments = split(message.clone(), 18);
// Reverse arrival order.
fragments.reverse();
let rec = MessageReconstructor::<u64, u64>::new(60);
let mut last = None;
let mut out = None;
for (i, f) in fragments.into_iter().enumerate() {
last = rec.insert_new_fragment(f, i as u64);
out = rec.insert_new_fragment(f, i as u64);
}
let (bytes, _) = last.expect("must complete after all fragments arrive");
assert_eq!(bytes, message);
}
#[test]
fn reconstructor_propagates_frame_kind() {
let message = b"outfox bytes";
let mut fragments = split(message, OUTFOX, 64);
let rec = MessageReconstructor::<u64, u64>::new(60);
let (_, metadata) = rec
.insert_new_fragment(fragments.pop().unwrap(), 0)
.expect("complete");
assert_eq!(metadata.kind(), OUTFOX);
let recovered_frame = out
.expect("last fragment must complete the message")
.unwrap();
assert_eq!(recovered_frame, message);
}
#[test]
@@ -407,11 +394,11 @@ mod tests {
// Interleave.
assert!(rec.insert_new_fragment(a.remove(0), 0).is_none());
assert!(rec.insert_new_fragment(b.remove(0), 1).is_none());
let (msg_a, _) = rec.insert_new_fragment(a.remove(0), 2).unwrap();
let (msg_b, _) = rec.insert_new_fragment(b.remove(0), 3).unwrap();
let msg_a = rec.insert_new_fragment(a.remove(0), 2).unwrap().unwrap();
let msg_b = rec.insert_new_fragment(b.remove(0), 3).unwrap().unwrap();
assert_eq!(msg_a, vec![0xa1, 0xa2]);
assert_eq!(msg_b, vec![0xb1, 0xb2]);
assert_eq!(msg_a.content, vec![0xa1, 0xa2]);
assert_eq!(msg_b.content, vec![0xb1, 0xb2]);
}
#[test]
@@ -425,13 +412,13 @@ mod tests {
let rec = MessageReconstructor::<u64, u64>::new(60);
assert!(rec.insert_new_fragment(s1, 0).is_none());
assert!(rec.insert_new_fragment(o1, 1).is_none());
let (s_msg, s_metadata) = rec.insert_new_fragment(s2, 2).unwrap();
let (o_msg, o_metadata) = rec.insert_new_fragment(o2, 3).unwrap();
let s_msg = rec.insert_new_fragment(s2, 2).unwrap().unwrap();
let o_msg = rec.insert_new_fragment(o2, 3).unwrap().unwrap();
assert_eq!(s_msg, vec![0x10, 0x11]);
assert_eq!(s_metadata.kind(), SPHINX);
assert_eq!(o_msg, vec![0x20, 0x21]);
assert_eq!(o_metadata.kind(), OUTFOX);
assert_eq!(s_msg.content, vec![0x10, 0x11]);
assert_eq!(s_msg.kind(), SPHINX);
assert_eq!(o_msg.content, vec![0x20, 0x21]);
assert_eq!(o_msg.kind(), OUTFOX);
}
#[test]
@@ -477,8 +464,8 @@ mod tests {
assert!(rec.insert_new_fragment(stale, 0).is_none());
assert_eq!(rec.in_flight_messages.len(), 1);
let (msg, _) = rec.insert_new_fragment(fresh, 1_000).unwrap();
assert_eq!(msg, vec![0xff]);
let msg = rec.insert_new_fragment(fresh, 1_000).unwrap().unwrap();
assert_eq!(msg.content, vec![0xff]);
// `fresh` was a single-fragment message and is removed on emission;
// the stale buffer must also be gone.
assert!(rec.in_flight_messages.is_empty());
@@ -502,7 +489,7 @@ mod tests {
// Absolute time is now 16 (> 10), but the gap from the previous
// fragment (8) to now (16) is 8, still within the 10-tick timeout.
let out = rec.insert_new_fragment(make_fragment(1, 3, 2, SPHINX, vec![0xc]), 16);
let (msg, _) = out.expect("buffer must still be alive");
assert_eq!(msg, vec![0xa, 0xb, 0xc]);
let msg = out.expect("buffer must still be alive").unwrap();
assert_eq!(msg.content, vec![0xa, 0xb, 0xc]);
}
}
common/nym-lp-data/src/packet/frame.rs
+9 -18
View File
@@ -14,23 +14,14 @@ pub enum LpFrameKind {
Registration = 1,
Forward = 2,
SphinxStream = 3,
FragmentedSphinxPacket = 4,
FragmentedOutfoxPacket = 5,
FragmentedData = 4,
SphinxPacket = 5,
OutfoxPacket = 6,
#[num_enum(catch_all)]
Unknown(u16),
}
impl LpFrameKind {
// Indicate if the frame attributes can be parsed as fragments attributess
pub fn is_fragmented(&self) -> bool {
matches!(
self,
Self::FragmentedSphinxPacket | Self::FragmentedOutfoxPacket
)
}
}
/// Raw 14-byte frame attributes field in every [`LpFrameHeader`].
/// Interpretation depends on the [`LpFrameKind`].
pub type LpFrameAttributes = [u8; 14];
@@ -88,12 +79,6 @@ pub struct LpFrame {
pub content: Bytes,
}
impl AsRef<[u8]> for LpFrame {
fn as_ref(&self) -> &[u8] {
&self.content
}
}
impl LpFrame {
pub fn new(kind: LpFrameKind, content: impl Into<Bytes>) -> Self {
Self {
@@ -126,6 +111,12 @@ impl LpFrame {
Ok(Self { header, content })
}
pub fn to_bytes(self) -> Vec<u8> {
let mut bytes = BytesMut::new();
self.encode(&mut bytes);
bytes.freeze().to_vec()
}
pub fn kind(&self) -> LpFrameKind {
self.header.kind
}
nym-node/src/node/lp/data/handler/error.rs
+3
View File
@@ -28,6 +28,9 @@ pub enum LpDataHandlerError {
#[error("received payload type of an unexpected type: {typ:?}")]
UnexpectedLpPayload { typ: LpFrameKind },
#[error("received an Lp Frame kind that we don't support: {typ:?}")]
UnsupportedLpFrameKind { typ: LpFrameKind },
#[error("unwrapped a packet into a final hop packet. This is no longer supported")]
FinalHop,
nym-node/src/node/lp/data/handler/messages.rs
+45 -48
View File
@@ -1,65 +1,62 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nym_lp_data::{fragmentation::fragment::FragmentMetadata, packet::frame::LpFrameKind};
use nym_lp_data::packet::frame::{LpFrameAttributes, LpFrameHeader, LpFrameKind};
use nym_sphinx_forwarding::packet::MixPacketFormattingError;
use nym_sphinx_params::SphinxKeyRotation;
use crate::node::lp::data::handler::error::LpDataHandlerError;
/// Message types supported by mixnodes
#[derive(Debug, Clone, Copy)]
pub enum MixMessage {
Sphinx {
key_rotation: SphinxKeyRotation,
reserved: [u8; 3],
},
Outfox {
key_rotation: SphinxKeyRotation,
reserved: [u8; 3],
},
Sphinx(SphinxMixMessage),
Outfox(OutfoxMixMessage),
}
impl TryFrom<FragmentMetadata> for MixMessage {
type Error = LpDataHandlerError;
fn try_from(value: FragmentMetadata) -> Result<Self, Self::Error> {
let key_rotation = value.metadata()[0]
.try_into()
.map_err(MixPacketFormattingError::InvalidKeyRotation)?;
// SAFETY : correct length casting
#[allow(clippy::unwrap_used)]
let reserved = value.metadata()[1..4].try_into().unwrap();
match value.kind() {
LpFrameKind::FragmentedOutfoxPacket => Ok(MixMessage::Outfox {
key_rotation,
reserved,
}),
LpFrameKind::FragmentedSphinxPacket => Ok(MixMessage::Sphinx {
key_rotation,
reserved,
}),
_ => Err(LpDataHandlerError::UnexpectedLpPayload { typ: value.kind() }),
}
}
}
impl From<MixMessage> for FragmentMetadata {
impl From<MixMessage> for LpFrameHeader {
fn from(value: MixMessage) -> Self {
match value {
MixMessage::Outfox {
key_rotation,
reserved,
} => {
let metadata = [key_rotation as u8, reserved[0], reserved[1], reserved[2]];
(LpFrameKind::FragmentedOutfoxPacket, metadata).into()
}
MixMessage::Sphinx {
key_rotation,
reserved,
} => {
let metadata = [key_rotation as u8, reserved[0], reserved[1], reserved[2]];
(LpFrameKind::FragmentedSphinxPacket, metadata).into()
}
MixMessage::Sphinx(msg) => LpFrameHeader::new(LpFrameKind::SphinxPacket, msg),
MixMessage::Outfox(msg) => LpFrameHeader::new(LpFrameKind::OutfoxPacket, msg),
}
}
}
impl TryFrom<LpFrameHeader> for MixMessage {
type Error = LpDataHandlerError;
fn try_from(value: LpFrameHeader) -> Result<Self, Self::Error> {
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 {
pub key_rotation: SphinxKeyRotation,
}
impl TryFrom<LpFrameAttributes> for SphinxMixMessage {
type Error = LpDataHandlerError;
fn try_from(value: LpFrameAttributes) -> Result<Self, Self::Error> {
let key_rotation = value[0]
.try_into()
.map_err(MixPacketFormattingError::InvalidKeyRotation)?;
Ok(SphinxMixMessage { key_rotation })
}
}
impl From<SphinxMixMessage> for LpFrameAttributes {
fn from(value: SphinxMixMessage) -> Self {
let mut attrs = [0; 14];
attrs[0] = value.key_rotation as u8;
attrs
}
}
// For now there are no differences. We can augment this variant when we will need it
pub type OutfoxMixMessage = SphinxMixMessage;
nym-node/src/node/lp/data/handler/pipeline.rs
+120 -96
View File
@@ -9,11 +9,11 @@ use nym_lp_data::{
Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline,
WireWrappingPipeline,
},
fragmentation::fragment::fragment_payload,
fragmentation::fragment::fragment_lp_message,
mixnodes::traits::MixnodeProcessingPipeline,
packet::{
EncryptedLpPacket, LpFrame, LpHeader, LpPacket, MalformedLpPacketError,
frame::LpFrameHeader,
frame::{LpFrameHeader, LpFrameKind},
},
};
use rand::Rng;
@@ -58,18 +58,15 @@ where
_: Instant,
) -> Vec<PipelinePayload<Instant, MixMessage, SocketAddr>> {
let processing_result = match message_kind {
MixMessage::Sphinx {
key_rotation,
reserved: _,
} => processing::sphinx::process(&self.state, payload, key_rotation),
MixMessage::Outfox {
key_rotation,
reserved: _,
} => processing::outfox::process(&self.state, payload, key_rotation),
MixMessage::Sphinx(metadata) => {
processing::sphinx::process(&self.state, payload, metadata)
}
MixMessage::Outfox(metadata) => {
processing::outfox::process(&self.state, payload, metadata)
}
};
self.state
.update_processing_metrics(&processing_result, message_kind);
self.state.update_processing_metrics(&processing_result);
let packet_to_forward = match processing_result {
Ok(packet) => packet,
@@ -89,7 +86,7 @@ where
self.state.routing_filter_dropped(next_hop);
Vec::new()
} else {
vec![packet_to_forward.with_options(message_kind)]
vec![packet_to_forward]
}
}
}
@@ -107,19 +104,23 @@ where
payload: PipelinePayload<Instant, MixMessage, SocketAddr>,
frame_size: usize,
) -> Vec<AddressedTimedData<Instant, Self::Frame, SocketAddr>> {
let content = payload.data.data;
let fragments =
fragment_payload(&mut self.rng, &content, payload.options.into(), frame_size);
let frame = LpFrame {
header: payload.options.into(),
content: payload.data.data.into(),
};
fragments
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.into_lp_frame(),
payload.dst,
)
})
.map(|f| AddressedTimedData::new_addressed(payload.data.timestamp, f, payload.dst))
.collect()
}
}
@@ -183,7 +184,7 @@ where
&mut self,
frame: TimedData<Instant, Self::Frame>,
) -> Option<(TimedPayload<Instant>, MixMessage)> {
if frame.data.kind().is_fragmented() {
let reassembled_frame = if frame.data.kind() == LpFrameKind::FragmentedData {
let fragment = frame
.data
.try_into()
@@ -192,23 +193,44 @@ where
self.state.malformed_packet();
})
.ok()?;
let (payload, metadata) = self
let message = self
.state
.message_reconstructor
.insert_new_fragment(fragment, frame.timestamp)?;
let message_kind = metadata
.try_into()
.insert_new_fragment(fragment, frame.timestamp)?
.inspect_err(|e| {
tracing::warn!(
"Somehow got a non fragmented message kind from reconstruction buffer : {e}"
);
tracing::error!("Failed to recover a frame : {e}");
self.state.malformed_packet();
})
.ok()?;
self.state.message_received(message_kind);
Some((TimedPayload::new(frame.timestamp, payload), message_kind))
TimedData::new(frame.timestamp, message)
} else {
warn!("unimplemented yet");
None
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,
))
}
}
}
}
@@ -227,11 +249,11 @@ mod tests {
use std::time::{Duration, Instant};
use nym_lp_data::common::traits::WireWrappingPipeline;
use nym_lp_data::fragmentation::fragment::{FragmentMetadata, fragment_payload};
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::packet::{
EncryptedLpPacket, LpFrame, LpHeader, LpPacket, OuterHeader, version,
EncryptedLpPacket, LpFrame, LpHeader, LpPacket, OuterHeader, frame::LpFrameHeader, version,
};
use nym_node_metrics::NymNodeMetrics;
use nym_node_metrics::mixnet::PacketKind;
@@ -248,7 +270,7 @@ mod tests {
use crate::config::{LpConfig, ReplayProtectionDebug};
use crate::node::key_rotation::active_keys::ActiveSphinxKeys;
use crate::node::key_rotation::key::SphinxPrivateKey;
use crate::node::lp::data::handler::messages::MixMessage;
use crate::node::lp::data::handler::messages::{MixMessage, SphinxMixMessage};
use crate::node::lp::data::handler::pipeline::MixnodeDataPipeline;
use crate::node::lp::data::shared::{ProcessingConfig, SharedLpDataState};
use crate::node::replay_protection::bloomfilter::{
@@ -427,37 +449,48 @@ mod tests {
LpPacket::new(LpHeader::new(0, 0, version::CURRENT), frame).encode()
}
/// Fragment `bytes` into `EncryptedLpPacket`s carrying the given mix-message
/// metadata. `fragment_payload_size` controls how the payload is split:
/// pass at least `bytes.len()` to get a single fragment, or smaller to force
/// multiple fragments.
/// Wrap `bytes` into an [`LpFrame`] using `message` as the header, then
/// either pass it through unfragmented (if it fits within `frame_size`)
/// or split it into fragments. Mirrors the pipeline's framing logic so
/// tests can build the exact packet sequence a peer would emit.
fn fragment_into_lp_packets(
bytes: &[u8],
message: MixMessage,
fragment_payload_size: usize,
frame_size: usize,
rng: &mut DeterministicRng,
) -> Vec<EncryptedLpPacket> {
let metadata: FragmentMetadata = message.into();
fragment_payload(rng, bytes, metadata, fragment_payload_size.max(1))
let frame = LpFrame {
header: message.into(),
content: bytes.to_vec().into(),
};
let frames = if frame.len() > frame_size {
fragment_lp_message(rng, frame, frame_size)
.into_iter()
.map(|f| f.into_lp_frame())
.collect()
} else {
vec![frame]
};
frames
.into_iter()
.map(|f| lp_frame_to_encrypted_packet(f.into_lp_frame()))
.map(lp_frame_to_encrypted_packet)
.collect()
}
/// Default sphinx mix-message metadata used by tests (rotation matching the
/// even primary key in [`mock_shared_state`]).
fn sphinx_mix_message() -> MixMessage {
MixMessage::Sphinx {
MixMessage::Sphinx(SphinxMixMessage {
key_rotation: SphinxKeyRotation::EvenRotation,
reserved: [0u8; 3],
}
})
}
fn outfox_mix_message() -> MixMessage {
MixMessage::Outfox {
MixMessage::Outfox(SphinxMixMessage {
key_rotation: SphinxKeyRotation::EvenRotation,
reserved: [0u8; 3],
}
})
}
// ==================== Tests ====================
@@ -476,38 +509,35 @@ mod tests {
state.sphinx_keys.primary().x25519_pubkey().into(),
delay,
next_hop,
pipeline.frame_size(),
pipeline.frame_size() - LpFrameHeader::SIZE,
&mut rng,
);
// Sanity check
assert_eq!(sphinx_bytes.len(), pipeline.frame_size());
let inputs = fragment_into_lp_packets(
&sphinx_bytes,
sphinx_mix_message(),
pipeline.frame_size(),
&mut rng,
);
assert_eq!(inputs.len(), 1, "expected a single input fragment");
assert_eq!(inputs.len(), 1, "expected a single input frame");
let input_packet = inputs[0].clone();
let arrival = Instant::now();
let outputs = pipeline.process(input_packet, arrival).unwrap();
assert_eq!(outputs.len(), 1, "expected exactly one output fragment");
assert_eq!(outputs.len(), 1, "expected exactly one output frame");
let output_packet = outputs[0].clone();
assert_eq!(
output_packet.dst, next_hop,
"output fragment must target the next hop"
"output frame must target the next hop"
);
assert_eq!(
output_packet.data.timestamp,
arrival + delay.to_duration(),
"output fragment delay must match arrival + delay"
"output frame delay must match arrival + delay"
);
assert_eq!(
state
@@ -531,37 +561,34 @@ mod tests {
let outfox_bytes = build_outfox_bytes(
state.sphinx_keys.primary().x25519_pubkey().into(),
next_hop,
pipeline.frame_size(),
pipeline.frame_size() - LpFrameHeader::SIZE,
&mut rng,
);
// Sanity check
assert_eq!(outfox_bytes.len(), pipeline.frame_size());
let inputs = fragment_into_lp_packets(
&outfox_bytes,
outfox_mix_message(),
pipeline.frame_size(),
&mut rng,
);
assert_eq!(inputs.len(), 1, "expected a single input fragment");
assert_eq!(inputs.len(), 1, "expected a single input frame");
let input_packet = inputs[0].clone();
let arrival = Instant::now();
let outputs = pipeline.process(input_packet, arrival).unwrap();
assert_eq!(outputs.len(), 1, "expected exactly one output fragment");
assert_eq!(outputs.len(), 1, "expected exactly one output frame");
let output_packet = outputs[0].clone();
assert_eq!(
output_packet.dst, next_hop,
"output fragment must target the next hop"
"output frame must target the next hop"
);
assert_eq!(
output_packet.data.timestamp, arrival,
"outfox output fragment should not have any delay"
"outfox output frame should not have any delay"
);
assert_eq!(
state
@@ -582,19 +609,16 @@ mod tests {
let sphinx_bytes = build_final_hop_sphinx_bytes(
state.sphinx_keys.primary().x25519_pubkey().into(),
Delay::new_from_millis(50),
pipeline.frame_size(),
pipeline.frame_size() - LpFrameHeader::SIZE,
);
// Sanity check
assert_eq!(sphinx_bytes.len(), pipeline.frame_size());
let inputs = fragment_into_lp_packets(
&sphinx_bytes,
sphinx_mix_message(),
pipeline.frame_size(),
&mut rng,
);
assert_eq!(inputs.len(), 1, "expected a single input fragment");
assert_eq!(inputs.len(), 1, "expected a single input frame");
let input_packet = inputs[0].clone();
@@ -622,7 +646,7 @@ mod tests {
state.sphinx_keys.primary().x25519_pubkey().into(),
delay,
next_hop,
pipeline.frame_size(),
pipeline.frame_size() - LpFrameHeader::SIZE,
&mut rng,
);
@@ -632,7 +656,7 @@ mod tests {
pipeline.frame_size(),
&mut rng,
);
assert_eq!(inputs.len(), 1, "expected a single input fragment");
assert_eq!(inputs.len(), 1, "expected a single input frame");
let input_packet = inputs[0].clone();
// This also replays the LP encryption. This is fine for now since there is none, but once LP has replay protection by itself, we should test sphinx replay here
@@ -643,7 +667,7 @@ mod tests {
assert_eq!(
first.len(),
1,
"first send should be forwarded in one fragment"
"first send should be forwarded in one frame"
);
assert_eq!(
state
@@ -688,9 +712,14 @@ mod tests {
let (mut pipeline, state) = mock_pipeline();
let mut rng = deterministic_rng();
let garbage = vec![0xAAu8; pipeline.frame_size()];
let inputs =
fragment_into_lp_packets(&garbage, sphinx_mix_message(), garbage.len(), &mut rng);
let garbage = vec![0xAAu8; pipeline.frame_size() - LpFrameHeader::SIZE];
let inputs = fragment_into_lp_packets(
&garbage,
sphinx_mix_message(),
pipeline.frame_size(),
&mut rng,
);
assert_eq!(inputs.len(), 1, "expected a single input frame");
let outputs = pipeline
.process(inputs.into_iter().next().unwrap(), Instant::now())
@@ -715,12 +744,13 @@ mod tests {
let nb_fragments = 3;
// Packet fits exactly in one frame
// Sized so the wrapping LpFrame is exactly `nb_fragments * frame_size`
// bytes once serialized, which fragments into `nb_fragments` pieces.
let sphinx_bytes = build_sphinx_bytes(
state.sphinx_keys.primary().x25519_pubkey().into(),
delay,
next_hop,
nb_fragments * pipeline.frame_size(),
nb_fragments * pipeline.frame_size() - LpFrameHeader::SIZE,
&mut rng,
);
@@ -766,14 +796,14 @@ mod tests {
for out_pkt in out {
assert_eq!(
out_pkt.dst, next_hop,
"output fragment must target the next hop"
"output frame must target the next hop"
);
// All fragments should have a ts of the last arrival plus delay
assert_eq!(
out_pkt.data.timestamp,
arrivals[nb_fragments - 1] + delay.to_duration(),
"output fragment delay must match arrival + delay"
"output frame delay must match arrival + delay"
);
}
@@ -801,33 +831,30 @@ mod tests {
state.sphinx_keys.primary().x25519_pubkey().into(),
huge_delay,
next_hop,
pipeline.frame_size(),
pipeline.frame_size() - LpFrameHeader::SIZE,
&mut rng,
);
// Sanity check
assert_eq!(sphinx_bytes.len(), pipeline.frame_size());
let inputs = fragment_into_lp_packets(
&sphinx_bytes,
sphinx_mix_message(),
pipeline.frame_size(),
&mut rng,
);
assert_eq!(inputs.len(), 1, "expected a single input fragment");
assert_eq!(inputs.len(), 1, "expected a single input frame");
let input_packet = inputs[0].clone();
let arrival = Instant::now();
let outputs = pipeline.process(input_packet, arrival).unwrap();
assert_eq!(outputs.len(), 1, "expected exactly one output fragment");
assert_eq!(outputs.len(), 1, "expected exactly one output frame");
let output_packet = outputs[0].clone();
assert_eq!(
output_packet.dst, next_hop,
"output fragment must target the next hop"
"output frame must target the next hop"
);
assert_eq!(
output_packet.data.timestamp,
@@ -862,20 +889,17 @@ mod tests {
state.sphinx_keys.primary().x25519_pubkey().into(),
delay,
next_hop,
pipeline.frame_size(),
pipeline.frame_size() - LpFrameHeader::SIZE,
&mut rng,
);
// Sanity check
assert_eq!(sphinx_bytes.len(), pipeline.frame_size());
let inputs = fragment_into_lp_packets(
&sphinx_bytes,
sphinx_mix_message(),
pipeline.frame_size(),
&mut rng,
);
assert_eq!(inputs.len(), 1, "expected a single input fragment");
assert_eq!(inputs.len(), 1, "expected a single input frame");
let input_packet = inputs[0].clone();
nym-node/src/node/lp/data/handler/processing/outfox.rs
+13 -7
View File
@@ -2,19 +2,24 @@
// SPDX-License-Identifier: Apache-2.0
use std::{net::SocketAddr, time::Instant};
use nym_lp_data::{AddressedTimedPayload, TimedPayload};
use nym_lp_data::{PipelinePayload, TimedPayload};
use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
use nym_sphinx_params::SphinxKeyRotation;
use nym_sphinx_types::OutfoxPacket;
use tracing::warn;
use crate::node::lp::data::{handler::error::LpDataHandlerError, shared::SharedLpDataState};
use crate::node::lp::data::{
handler::{
error::LpDataHandlerError,
messages::{MixMessage, OutfoxMixMessage},
},
shared::SharedLpDataState,
};
pub(crate) fn process(
shared_state: &SharedLpDataState,
outfox_packet: TimedPayload<Instant>,
key_rotation: SphinxKeyRotation,
) -> Result<AddressedTimedPayload<Instant, SocketAddr>, LpDataHandlerError> {
metadata: OutfoxMixMessage,
) -> Result<PipelinePayload<Instant, MixMessage, SocketAddr>, LpDataHandlerError> {
let TimedPayload {
data: outfox_bytes,
timestamp: arrival_timestamp,
@@ -22,16 +27,17 @@ pub(crate) fn process(
let mut outfox_packet = OutfoxPacket::try_from(outfox_bytes.as_slice())?;
let key = shared_state.resolve_rotation_key(key_rotation)?;
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!("Dropping final hop packet as it is no longer supported");
Err(LpDataHandlerError::FinalHop)
} else {
Ok(AddressedTimedPayload::new_addressed(
Ok(PipelinePayload::new(
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(),
))
}
nym-node/src/node/lp/data/handler/processing/sphinx.rs
+13 -7
View File
@@ -3,20 +3,25 @@
use std::{net::SocketAddr, time::Instant};
use nym_lp_data::{AddressedTimedPayload, TimedPayload};
use nym_lp_data::{PipelinePayload, TimedPayload};
use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
use nym_sphinx_framing::processing::PacketProcessingError;
use nym_sphinx_params::SphinxKeyRotation;
use nym_sphinx_types::SphinxPacket;
use tracing::{error, warn};
use crate::node::lp::data::{handler::error::LpDataHandlerError, shared::SharedLpDataState};
use crate::node::lp::data::{
handler::{
error::LpDataHandlerError,
messages::{MixMessage, SphinxMixMessage},
},
shared::SharedLpDataState,
};
pub(crate) fn process(
shared_state: &SharedLpDataState,
sphinx_packet: TimedPayload<Instant>,
key_rotation: SphinxKeyRotation,
) -> Result<AddressedTimedPayload<Instant, SocketAddr>, LpDataHandlerError> {
metadata: SphinxMixMessage,
) -> Result<PipelinePayload<Instant, MixMessage, SocketAddr>, LpDataHandlerError> {
let TimedPayload {
data: sphinx_bytes,
timestamp: arrival_timestamp,
@@ -25,7 +30,7 @@ pub(crate) fn process(
let sphinx_packet = SphinxPacket::from_bytes(&sphinx_bytes)?;
// Extracting shared_secret
let key = shared_state.resolve_rotation_key(key_rotation)?;
let key = shared_state.resolve_rotation_key(metadata.key_rotation)?;
let rotation_id = key.rotation_id();
let expanded_shared_secret = sphinx_packet
.header
@@ -69,9 +74,10 @@ pub(crate) fn process(
shared_state.excessive_delay_packet();
delay = shared_state.processing_config.maximum_packet_delay;
}
Ok(AddressedTimedPayload::new_addressed(
Ok(PipelinePayload::new(
arrival_timestamp + delay,
next_hop_packet.to_bytes(),
MixMessage::Sphinx(metadata),
NymNodeRoutingAddress::try_from(next_hop_address)?.into(),
))
}
nym-node/src/node/lp/data/shared.rs
+9 -7
View File
@@ -9,7 +9,7 @@ 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::AddressedTimedPayload;
use nym_lp_data::PipelinePayload;
use nym_lp_data::fragmentation::reconstruction::MessageReconstructor;
use nym_node_metrics::NymNodeMetrics;
use nym_node_metrics::mixnet::PacketKind;
@@ -56,7 +56,7 @@ pub(crate) struct SharedLpDataState {
pub shutdown_token: ShutdownToken,
}
fn message_kind_to_packet_kind(message_kind: MixMessage) -> PacketKind {
fn message_kind_to_packet_kind(message_kind: &MixMessage) -> PacketKind {
match message_kind {
MixMessage::Sphinx { .. } => PacketKind::LpSphinx,
MixMessage::Outfox { .. } => PacketKind::LpOutfox,
@@ -123,7 +123,7 @@ impl SharedLpDataState {
self.metrics.mixnet.lp_malformed_packet()
}
pub(super) fn message_received(&self, message_kind: MixMessage) {
pub(super) fn message_received(&self, message_kind: &MixMessage) {
self.metrics
.mixnet
.lp_message_received(message_kind_to_packet_kind(message_kind))
@@ -162,14 +162,16 @@ impl SharedLpDataState {
pub(super) fn update_processing_metrics(
&self,
processing_result: &Result<AddressedTimedPayload<Instant, SocketAddr>, LpDataHandlerError>,
message_kind: MixMessage,
processing_result: &Result<
PipelinePayload<Instant, MixMessage, SocketAddr>,
LpDataHandlerError,
>,
) {
match processing_result {
Ok(_) => {
Ok(packet) => {
self.metrics
.mixnet
.lp_processed_message(message_kind_to_packet_kind(message_kind));
.lp_processed_message(message_kind_to_packet_kind(&packet.options));
}
Err(LpDataHandlerError::PacketProcessingError(PacketProcessingError::PacketReplay)) => {
self.metrics.mixnet.lp_processing_replayed_packet();