From ed5ddf0170afd72d8a7ac0f17f8336b0a2702b4f Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Wed, 20 May 2026 11:33:14 +0200 Subject: [PATCH] nym-lp-data crate --- Cargo.lock | 30 +- Cargo.toml | 4 + common/nym-lp-data/Cargo.toml | 7 +- common/nym-lp-data/README.md | 103 +++++++ common/nym-lp-data/src/clients/driver.rs | 95 ++++++ common/nym-lp-data/src/clients/helpers.rs | 72 +++++ common/nym-lp-data/src/clients/mod.rs | 78 +++++ common/nym-lp-data/src/clients/traits.rs | 286 ++++++++++++++++++ common/nym-lp-data/src/clients/types.rs | 155 ++++++++++ common/nym-lp-data/src/common/helpers.rs | 100 ++++++ common/nym-lp-data/src/common/mod.rs | 5 + common/nym-lp-data/src/common/traits.rs | 169 +++++++++++ common/nym-lp-data/src/lib.rs | 145 +++++++++ common/nym-lp-data/src/mixnodes/mod.rs | 5 + common/nym-lp-data/src/mixnodes/traits.rs | 61 ++++ common/nym-lp-data/src/mixnodes/types.rs | 56 ++++ .../tests/integration/common/mod.rs | 220 ++++++++++++++ common/nym-lp-data/tests/integration/main.rs | 34 +++ 18 files changed, 1620 insertions(+), 5 deletions(-) create mode 100644 common/nym-lp-data/README.md create mode 100644 common/nym-lp-data/src/clients/driver.rs create mode 100644 common/nym-lp-data/src/clients/helpers.rs create mode 100644 common/nym-lp-data/src/clients/mod.rs create mode 100644 common/nym-lp-data/src/clients/traits.rs create mode 100644 common/nym-lp-data/src/clients/types.rs create mode 100644 common/nym-lp-data/src/common/helpers.rs create mode 100644 common/nym-lp-data/src/common/mod.rs create mode 100644 common/nym-lp-data/src/common/traits.rs create mode 100644 common/nym-lp-data/src/mixnodes/mod.rs create mode 100644 common/nym-lp-data/src/mixnodes/traits.rs create mode 100644 common/nym-lp-data/src/mixnodes/types.rs create mode 100644 common/nym-lp-data/tests/integration/common/mod.rs create mode 100644 common/nym-lp-data/tests/integration/main.rs diff --git a/Cargo.lock b/Cargo.lock index 5718dab961..bc42a8138e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7487,12 +7487,12 @@ dependencies = [ [[package]] name = "nym-lp-data" -version = "1.21.0" +version = "1.20.4" dependencies = [ - "bytes", - "num_enum", + "anyhow", + "nym-client-core", "nym-common", - "thiserror 2.0.18", + "nym-lp", "tracing", ] @@ -7506,6 +7506,28 @@ dependencies = [ "tracing", ] +[[package]] +name = "nym-mix-sim" +version = "1.20.4" +dependencies = [ + "anyhow", + "clap", + "nym-bin-common", + "nym-common", + "nym-crypto", + "nym-lp-data", + "nym-sphinx", + "rand 0.8.5", + "rand_distr", + "serde", + "serde_json", + "strum 0.28.0", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "nym-mixnet-client" version = "1.21.0" diff --git a/Cargo.toml b/Cargo.toml index 95d08ed7be..79b5caaf27 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -136,6 +136,7 @@ members = [ "nym-data-observatory", "nym-gateway-probe", "nym-ip-packet-client", + "nym-mix-sim", "nym-network-monitor", "nym-node", "nym-node-status-api/nym-node-status-agent", @@ -186,6 +187,7 @@ default-members = [ "nym-api", "nym-authenticator-client", "nym-credential-proxy/nym-credential-proxy", + "nym-mix-sim", "nym-node", "nym-registration-client", "nym-statistics-api", @@ -195,6 +197,8 @@ default-members = [ "service-providers/network-requester", "tools/internal/localnet-orchestrator", "tools/nymvisor", + "nym-registration-client", + "tools/internal/localnet-orchestrator", ] exclude = ["contracts", "nym-wallet", "cpu-cycles"] diff --git a/common/nym-lp-data/Cargo.toml b/common/nym-lp-data/Cargo.toml index 5a4f47aafa..a5882e4ef9 100644 --- a/common/nym-lp-data/Cargo.toml +++ b/common/nym-lp-data/Cargo.toml @@ -13,11 +13,16 @@ documentation.workspace = true # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +anyhow.workspace = true +tracing.workspace = true bytes.workspace = true num_enum.workspace = true -tracing.workspace = true thiserror.workspace = true + +[dev-dependencies] +nym-client-core.workspace = true +nym-lp.workspace = true nym-common.workspace = true diff --git a/common/nym-lp-data/README.md b/common/nym-lp-data/README.md new file mode 100644 index 0000000000..49212888d6 --- /dev/null +++ b/common/nym-lp-data/README.md @@ -0,0 +1,103 @@ +# nym-lp-data + +Trait definitions and data structures for Lewes Protocol (LP) processing pipelines in the Nym mixnet. + +This crate is a *vocabulary* crate — it defines the traits that clients and mix nodes implement to compose a packet-processing pipeline, plus a few generic data wrappers (`TimedData`, `AddressedTimedData`, `PipelineData`) that thread per-packet state through every stage. It contains no concrete cryptography, transport, or network code. A concrete implementation live in [`nym-mix-sim`](../../nym-mix-sim). + +## Crate layout + +| Module | Purpose | +|--------|---------| +| [`common`](src/common) | Wire-layer traits ([`Framing`], [`FramingUnwrap`], [`Transport`], [`TransportUnwrap`]) and their composed supertraits ([`WireWrappingPipeline`], [`WireUnwrappingPipeline`]) shared by both clients and mixnodes, plus [`NoOpWireWrapper`] / [`NoOpWireUnwrapper`] marker traits for opting into a pass-through wire layer | +| [`clients`](src/clients) | Client-side outbound/inbound pipeline traits: [`Chunking`], [`Reliability`], [`Obfuscation`], [`RoutingSecurity`], plus the supertraits [`ClientWrappingPipeline`] / [`ClientUnwrappingPipeline`], a `Pipeline` composition struct, no-op marker traits, and a tick-driven [`ClientWrappingPipelineDriver`] | +| [`mixnodes`](src/mixnodes) | Mixnode processing trait [`MixnodeProcessingPipeline`] (unwrap → mix → re-wrap) and a `Pipeline` composition struct | + +[`Framing`]: src/common/traits.rs +[`FramingUnwrap`]: src/common/traits.rs +[`Transport`]: src/common/traits.rs +[`TransportUnwrap`]: src/common/traits.rs +[`WireWrappingPipeline`]: src/common/traits.rs +[`WireUnwrappingPipeline`]: src/common/traits.rs +[`NoOpWireWrapper`]: src/common/helpers.rs +[`NoOpWireUnwrapper`]: src/common/helpers.rs +[`Chunking`]: src/clients/traits.rs +[`Reliability`]: src/clients/traits.rs +[`Obfuscation`]: src/clients/traits.rs +[`RoutingSecurity`]: src/clients/traits.rs +[`ClientWrappingPipeline`]: src/clients/traits.rs +[`ClientUnwrappingPipeline`]: src/clients/traits.rs +[`ClientWrappingPipelineDriver`]: src/clients/driver.rs +[`MixnodeProcessingPipeline`]: src/mixnodes/traits.rs + +## Core data types + +```text +TimedData ── pairs a value of type D with a timestamp Ts +TimedPayload ── alias for TimedData> + +AddressedTimedData ── TimedData plus a destination address +AddressedTimedPayload ── alias for AddressedTimedData, NdId> + +PipelineData ── TimedData plus per-message Opts + (used inside the client wrapping pipeline) +PipelinePayload ── alias for PipelineData, Opts, NdId> +``` + +`Ts` is the timestamp / tick-context type, `NdId` is the next-hop identifier type, and `Opts` is an [`InputOptions`](src/clients/mod.rs)-implementing per-message marker that toggles which optional pipeline stages run for a given payload (reliability, obfuscation, routing security). + +## Client wrapping pipeline + +The outbound client pipeline composes six stages, each represented by its own trait: + +```text + Vec ──▶ Chunking ──▶ Reliability ──▶ Obfuscation + │ + ▼ + AddressedTimedData ◀── Transport ◀── Framing ◀── RoutingSecurity +``` + +[`ClientWrappingPipeline`] is the supertrait that ties them together and provides a default `process()` method which runs all six stages in order on every tick. Each stage is opt-in per message via the active [`InputOptions`]. + +### Pipeline tick semantics + +`process()` is intended to be called on every tick (with or without an input payload): + +- [`Reliability::reliable_encode`] is always called once with `Some(input)` (when present), then once more with `None` so that timer-driven retransmissions can fire even when no new payload arrived. +- [`Obfuscation::obfuscate`] follows the same pattern — once with the real input and once with `None` so that cover-traffic loops can fire on idle ticks. +- [`Chunking`] and [`RoutingSecurity`] only run when a payload is actually present. + +This convention is what allows pipelines to support Poisson cover traffic and SURB-ACK retransmission without the caller having to know whether anything is in flight. + +## Mixnode processing pipeline + +The mixnode pipeline is simpler — three stages that consume a packet and emit zero or more re-wrapped output packets: + +```text +Pkt ──▶ WireUnwrappingPipeline ──▶ mix ──▶ WireWrappingPipeline ──▶ Vec> + (TransportUnwrap + ▲ (Framing + Transport) + FramingUnwrap) │ + └── implementor decrypts, routes, + schedules delays, etc. +``` + +Implementors fill in `mix()`; everything else is provided by the [`MixnodeProcessingPipeline`] supertrait's default `process()`. + +## Helpers + +- **Client-stage no-op marker traits** ([`NoOpReliability`], [`NoOpRoutingSecurity`], [`NoOpObfuscation`] in [`clients/helpers.rs`](src/clients/helpers.rs)) — implement these to opt out of a pipeline stage with zero overhead. Useful for stub or testing pipelines. +- **Wire-layer no-op marker traits** ([`NoOpWireWrapper`], [`NoOpWireUnwrapper`] in [`common/helpers.rs`](src/common/helpers.rs)) — collapse the entire wire layer (framing + transport, or their inverses) to a pass-through. Use these when your packet type is already self-contained on the wire (e.g. a Sphinx packet) and needs no extra framing or transport header. `NoOpWireWrapper` requires `Pkt: From>`; `NoOpWireUnwrapper` requires `Pkt: Into>` and `Mk: Default`. +- **`Pipeline` composition structs** (in [`clients/types.rs`](src/clients/types.rs) and [`mixnodes/types.rs`](src/mixnodes/types.rs)) — generic structs that aggregate one component per pipeline stage and provide blanket impls of the relevant supertraits, so you can build a working pipeline by plugging in any combination of stage implementations. +- **[`ClientWrappingPipelineDriver`](src/clients/driver.rs)** — wraps a dyn-compatible client pipeline behind a tick-driven `tick(timestamp) -> Vec<(Pkt, NdId)>` interface, with an internal mpsc channel for application-supplied input payloads. Reads new input only when the internal buffer is empty so buffered packets do not stack additional latency on top. + +[`NoOpReliability`]: src/clients/helpers.rs +[`NoOpRoutingSecurity`]: src/clients/helpers.rs +[`NoOpObfuscation`]: src/clients/helpers.rs +[`InputOptions`]: src/clients/mod.rs +[`Reliability::reliable_encode`]: src/clients/traits.rs +[`Obfuscation::obfuscate`]: src/clients/traits.rs + +## Example users + +[`nym-mix-sim`](../../nym-mix-sim) is the reference consumer: it ships two complete pipeline implementations (a pass-through `Simple*` family and a full Sphinx + Poisson + SURB-ACK family) on top of the traits defined here. See its source for end-to-end examples of implementing each pipeline stage. + +The integration test under [`tests/integration`](tests/integration) wires together a small synthetic pipeline (`MockChunking`, `KcpReliability`, `SphinxSecurity`, `KekwObfuscation`, `LpFraming`, `LpTransport`) against the [`nym-lp`](../nym-lp) packet types — a useful starting point if you want to read a self-contained example of every trait being implemented. diff --git a/common/nym-lp-data/src/clients/driver.rs b/common/nym-lp-data/src/clients/driver.rs new file mode 100644 index 0000000000..e5e2d485ac --- /dev/null +++ b/common/nym-lp-data/src/clients/driver.rs @@ -0,0 +1,95 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use std::sync::mpsc; + +use crate::AddressedTimedData; +use crate::clients::InputOptions; +use crate::clients::traits::DynClientWrappingPipeline; + +/// Drives a [`DynClientWrappingPipeline`] tick-by-tick, feeding it raw application +/// payloads and emitting transport packets whose scheduled timestamp is due. +/// +/// ## How it works +/// +/// 1. The caller submits raw byte payloads via [`PipelineDriver::input_sender`]. +/// 2. On each call to [`PipelineDriver::tick`], the driver reads one pending +/// payload (only when both the packet buffer and the obfuscation buffer are +/// empty, to avoid adding extra latency on top of buffered data), runs it +/// through the pipeline, and appends the resulting timestamped packets to an +/// internal buffer. +/// 3. Packets whose `timestamp ≤ now` are extracted from the buffer and +/// returned to the caller for sending. +/// +/// `Ts` must implement `Clone + PartialOrd` so that timestamps can be compared +/// to decide which packets are due. +/// +pub struct ClientWrappingPipelineDriver +where + Ts: Clone + PartialOrd, + Opts: InputOptions, +{ + pipeline: Box>, + + packet_buffer: Vec>, + + input: mpsc::Receiver<(Vec, Opts)>, + + // Keeping a ref so we don't have problem about it being dropped + input_sender: mpsc::SyncSender<(Vec, Opts)>, +} + +impl ClientWrappingPipelineDriver +where + Ts: Clone + PartialOrd, + Opts: InputOptions, +{ + /// Create a new driver wrapping `pipeline`. + /// + /// Internally allocates a zero-capacity `sync_channel` for input payloads. + pub fn new( + pipeline: impl DynClientWrappingPipeline + 'static, + ) -> Self { + let (input_sender, input_receiver) = mpsc::sync_channel(0); + + Self { + pipeline: Box::new(pipeline), + packet_buffer: Vec::new(), + input: input_receiver, + input_sender, + } + } + + /// Return a clone of the sender half of the input channel. + /// + /// Send raw application payloads here; they will be picked up on the next + /// tick when the pipeline's internal buffers are empty. + pub fn input_sender(&self) -> mpsc::SyncSender<(Vec, Opts)> { + self.input_sender.clone() + } + + /// Advance the driver by one tick. + /// + /// Reads a pending input payload (if both the packet buffer and the + /// obfuscation buffer are empty), runs it through the pipeline, then + /// returns all packets whose `timestamp ≤ now`. + pub fn tick(&mut self, timestamp: Ts) -> Vec<(Pkt, NdId)> { + // We're reading a message only if our buffer is empty + // Otherwise, we will have buffers adding latencies to data + let next_message = if self.packet_buffer.is_empty() { + self.input + .try_recv() + .inspect_err(|_| tracing::trace!("No message in the queue")) + .ok() + } else { + None + }; + self.packet_buffer + .extend(self.pipeline.process(next_message, timestamp.clone())); + + self.packet_buffer + .extract_if(.., |p| p.data.timestamp <= timestamp) + .map(|pkt| (pkt.data.data, pkt.dst)) + .collect() + } +} diff --git a/common/nym-lp-data/src/clients/helpers.rs b/common/nym-lp-data/src/clients/helpers.rs new file mode 100644 index 0000000000..21eff70fb3 --- /dev/null +++ b/common/nym-lp-data/src/clients/helpers.rs @@ -0,0 +1,72 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::clients::traits::{Obfuscation, Reliability, RoutingSecurity}; +use crate::clients::{InputOptions, PipelinePayload}; + +/// Marker trait for a no-op [`Reliability`] implementation. +/// +/// Implement this for your pipeline type to get a [`Reliability`] impl that +/// passes the payload through unchanged with zero byte overhead. +pub trait NoOpReliability {} + +impl Reliability for T +where + T: NoOpReliability, + Opts: InputOptions, +{ + const OVERHEAD_SIZE: usize = 0; + fn reliable_encode( + &mut self, + input: Option>, + _: Ts, + ) -> Vec> { + input.map(|payload| vec![payload]).unwrap_or_default() + } +} + +/// Marker trait for a no-op [`RoutingSecurity`] implementation. +/// +/// Implement this for your pipeline type to get a [`RoutingSecurity`] impl that +/// passes the payload through unchanged with zero byte overhead and `nb_frames() == 1`. +pub trait NoOpRoutingSecurity {} + +impl RoutingSecurity for T +where + T: NoOpRoutingSecurity, + Opts: InputOptions, +{ + const OVERHEAD_SIZE: usize = 0; + + fn nb_frames(&self) -> usize { + 1 + } + + fn encrypt( + &mut self, + input: PipelinePayload, + ) -> PipelinePayload { + input + } +} + +/// Marker trait for a no-op [`Obfuscation`] implementation. +/// +/// Implement this for your pipeline type to get an [`Obfuscation`] impl that +/// passes the input through unchanged with no cover traffic, delay, or +/// buffering. +pub trait NoOpObfuscation {} + +impl Obfuscation for T +where + T: NoOpObfuscation, + Opts: InputOptions, +{ + fn obfuscate( + &mut self, + input: Option>, + _: Ts, + ) -> Vec> { + input.map(|payload| vec![payload]).unwrap_or_default() + } +} diff --git a/common/nym-lp-data/src/clients/mod.rs b/common/nym-lp-data/src/clients/mod.rs new file mode 100644 index 0000000000..700476c032 --- /dev/null +++ b/common/nym-lp-data/src/clients/mod.rs @@ -0,0 +1,78 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::{AddressedTimedData, TimedData}; + +pub mod driver; +pub mod helpers; +pub mod traits; +pub mod types; + +pub trait InputOptions: Clone { + fn reliability(&self) -> bool; + fn routing_security(&self) -> bool; + fn obfuscation(&self) -> bool; + + fn next_hop(&self) -> NdId; +} + +pub struct PipelineData +where + Opts: InputOptions, +{ + pub data: TimedData, + pub options: Opts, + _marker: std::marker::PhantomData, +} + +impl PipelineData +where + Opts: InputOptions, +{ + pub fn new(timestamp: Ts, data: D, options: Opts) -> Self { + PipelineData { + data: TimedData::new(timestamp, data), + options, + _marker: std::marker::PhantomData, + } + } + + /// Apply `op` to the data component, leaving the timestamp unchanged. + /// `Nd` can be a different type to allow type transform as well + pub fn data_transform(self, op: F) -> PipelineData + where + F: FnMut(D) -> Nd, + { + PipelineData { + data: self.data.data_transform(op), + options: self.options, + _marker: self._marker, + } + } + + /// Apply `op` to the timestamp component, leaving the data unchanged. + pub fn ts_transform(self, op: F) -> Self + where + F: FnMut(Ts) -> Ts, + { + PipelineData { + data: self.data.ts_transform(op), + options: self.options, + _marker: self._marker, + } + } +} + +impl From> for AddressedTimedData +where + Opts: InputOptions, +{ + fn from(value: PipelineData) -> Self { + AddressedTimedData { + dst: value.options.next_hop(), + data: value.data, + } + } +} + +pub type PipelinePayload = PipelineData, Opts, NdId>; diff --git a/common/nym-lp-data/src/clients/traits.rs b/common/nym-lp-data/src/clients/traits.rs new file mode 100644 index 0000000000..330601eff4 --- /dev/null +++ b/common/nym-lp-data/src/clients/traits.rs @@ -0,0 +1,286 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::clients::{InputOptions, PipelinePayload}; +use crate::common::traits::{WireUnwrappingPipeline, WireWrappingPipeline}; +use crate::{AddressedTimedData, TimedPayload}; + +/// Trait for splitting an incoming payload into timestamped chunks. +/// +/// # Type Parameters +/// - `Ts`: Timestamp type associated with each produced `TimedPayload`. +/// +/// # Parameters +/// - `input`: Raw payload to split into chunks. +/// - `chunk_size`: Maximum size of each chunk in bytes. +/// - `timestamp`: Timestamp to assign to the produced chunks. +/// +/// # Returns +/// - A vector of `TimedPayload`s representing the chunked payload. +pub trait Chunking +where + Opts: InputOptions, +{ + fn chunked( + &mut self, + input: Vec, + input_options: Opts, + chunk_size: usize, + timestamp: Ts, + ) -> Vec>; +} + +/// Trait for applying reliability encoding to a timed payload. +/// +/// # Type Parameters +/// - `Ts`: Timestamp type carried by the `TimedPayload`. +/// +/// # Associated Constants +/// - `OVERHEAD_SIZE`: Number of additional bytes added by the reliability scheme. +/// +/// # Parameters +/// - `input`: Payload to encode with the reliability mechanism. +/// # Returns +/// - A vector of `TimedPayload` containing the reliability-encoded data and potential retransmissions. +pub trait Reliability +where + Opts: InputOptions, +{ + const OVERHEAD_SIZE: usize; + fn reliable_encode( + &mut self, + input: Option>, + timestamp: Ts, + ) -> Vec>; +} + +/// Trait for applying obfuscation to a timed payload. +/// If obfuscation is used, `obfuscate` should be called at every `Ts` not just the ones with input +/// +/// # Type Parameters +/// - `Ts`: Timestamp type carried by the `TimedPayload`. +pub trait Obfuscation +where + Opts: InputOptions, +{ + /// Obfuscate a given timed payload + /// # Parameters + /// - `input`: Optional payload to obfusctate + /// - `timestamp` : Current timestamp + /// + /// # Returns + /// - An `Vec`, result of the obfuscation algorithm + /// - The vector can be empty if there is nothing to return right away + fn obfuscate( + &mut self, + input: Option>, + timestamp: Ts, + ) -> Vec>; +} + +/// Trait for applying routing-security encryption to a timed payload. +/// +/// # Type Parameters +/// - `Ts`: Timestamp type carried by the `TimedPayload`. +/// +/// # Associated Constants +/// - `OVERHEAD_SIZE`: Number of additional bytes added by the encryption scheme. +/// +/// # Required Methods +/// - `encrypt`: Encrypt the given payload, returning a new `TimedPayload`. +/// +/// # Provided Methods +/// - `nb_frames`: Number of transport frames that one encrypted payload expands +/// into; defaults to `1`. Override when the encryption scheme (e.g. Sphinx) +/// produces multiple frames per input chunk. +pub trait RoutingSecurity +where + Opts: InputOptions, +{ + const OVERHEAD_SIZE: usize; + fn nb_frames(&self) -> usize { + 1 + } + fn encrypt( + &mut self, + input: PipelinePayload, + ) -> PipelinePayload; +} + +/// Full client-side outbound message pipeline. +/// +/// Composes all six processing stages — [`Chunking`], [`Reliability`], +/// [`Obfuscation`], [`RoutingSecurity`], and the shared [`WireWrappingPipeline`] +/// (framing + transport) — into a single `process` call that takes a raw byte +/// payload and returns a list of timestamped transport packets ready for sending. +/// +/// # Type Parameters +/// - `Ts`: Timestamp type carried through the pipeline. +/// - `Pkt`: Final transport packet type produced by transport. +/// +/// # Provided Methods +/// - `chunk_size`: Derived from `frame_size` (via [`WireWrappingPipeline`]) minus +/// routing-security and reliability overheads, accounting for `nb_frames` expansion. +/// - `process`: Runs the full pipeline in order: +/// chunk → reliability encode → obfuscate → encrypt → frame → transport. +pub trait ClientWrappingPipeline: + Chunking + + Reliability + + Obfuscation + + RoutingSecurity + + WireWrappingPipeline +where + Ts: Clone, + NdId: Clone, + Opts: InputOptions, +{ + fn chunk_size(&self, input_options: Opts) -> usize { + // Frame size comes from WireWrappingPipeline + let mut chunk_size = self.frame_size(); + + if input_options.routing_security() { + chunk_size = + chunk_size * self.nb_frames() - >::OVERHEAD_SIZE; + } + + if input_options.reliability() { + chunk_size -= >::OVERHEAD_SIZE; + } + + chunk_size + } + + fn process( + &mut self, + input: Option<(Vec, Opts)>, // Optional to be able to tick the pipeline without input + timestamp: Ts, + ) -> Vec> { + let mut chunks = if let Some((input_data, input_options)) = input { + self.chunked( + input_data, + input_options.clone(), + self.chunk_size(input_options.clone()), + timestamp.clone(), + ) + } else { + Vec::new() + }; + + // Reliability stage with chunks that needs reliability + chunks = chunks + .into_iter() + .flat_map(|chunk| { + if chunk.options.reliability() { + self.reliable_encode(Some(chunk), timestamp.clone()) + } else { + vec![chunk] + } + }) + .collect(); + + // Even if we had nothing go into the reliability stage, we need to catch potential retransmissions + // If we had, this should be a no-op, since it already has been called with the same timestamp + chunks.append(&mut self.reliable_encode(None, timestamp.clone())); + + chunks = chunks + .into_iter() + .flat_map(|chunk| { + if chunk.options.obfuscation() { + self.obfuscate(Some(chunk), timestamp.clone()) + } else { + vec![chunk] + } + }) + .collect(); + + // Even if we had nothing go into the obfuscation stage, we need to catch potential cover traffic + // If we had, this should be a no-op, since it already has been called with the same timestamp + chunks.append(&mut self.obfuscate(None, timestamp.clone())); + + chunks = chunks + .into_iter() + .map(|chunk| { + if chunk.options.routing_security() { + self.encrypt(chunk) + } else { + chunk + } + }) + .collect(); + + chunks + .into_iter() + .flat_map(|payload| self.wire_wrap(payload.into())) + .collect::>() + } +} + +/// Dyn-compatible mirror of [`ClientWrappingPipeline`]. +/// +/// All associated constants from the sub-traits are exposed as methods so the +/// trait can be used as `dyn DynClientWrappingPipeline`, erasing the +/// concrete pipeline type while keeping `Ts`, `Fr`, and `Pkt`. +/// +/// Implement [`ClientWrappingPipeline`] on your concrete type; the blanket impl +/// below provides `DynClientWrappingPipeline` for free. +pub trait DynClientWrappingPipeline { + fn packet_size(&self) -> usize; + + fn process( + &mut self, + input: Option<(Vec, Opts)>, + timestamp: Ts, + ) -> Vec>; +} + +impl DynClientWrappingPipeline for T +where + Ts: Clone, + NdId: Clone, + Opts: InputOptions, + T: ClientWrappingPipeline, +{ + fn packet_size(&self) -> usize { + WireWrappingPipeline::packet_size(self) + } + + fn process( + &mut self, + input: Option<(Vec, Opts)>, + timestamp: Ts, + ) -> Vec> { + ClientWrappingPipeline::process(self, input, timestamp) + } +} + +/// Full client-side inbound pipeline. +/// +/// Combines the shared [`WireUnwrappingPipeline`] (transport + framing unwrap) with a +/// blank [`process_unwrapped`] step that the implementor fills in (routing-security +/// decrypt, reliability decode, chunk reassembly, etc.). +/// +/// # Type Parameters +/// - `Ts`: Timestamp type. +/// - `Pkt`: Transport packet type consumed as input. +/// - `Mk`: Message-kind marker returned alongside reassembled payloads. +/// +/// # Required Methods +/// - `process_unwrapped`: Called with the reassembled payload and its message kind +/// once a complete message is available. Returns the decoded application bytes, +/// or `None` if reassembly is still in progress. +/// +/// # Provided Methods +/// - `unwrap`: Strips the wire layers via [`WireUnwrappingPipeline::wire_unwrap`], +/// then delegates to `process_unwrapped`. +pub trait ClientUnwrappingPipeline: WireUnwrappingPipeline +where + Ts: Clone, +{ + fn process_unwrapped(&mut self, payload: TimedPayload, kind: Mk) -> Option>; + + fn unwrap(&mut self, input: Pkt, timestamp: Ts) -> anyhow::Result>> { + Ok(self + .wire_unwrap(input, timestamp)? + .and_then(|(payload, kind)| self.process_unwrapped(payload, kind))) + } +} diff --git a/common/nym-lp-data/src/clients/types.rs b/common/nym-lp-data/src/clients/types.rs new file mode 100644 index 0000000000..9360a8bf33 --- /dev/null +++ b/common/nym-lp-data/src/clients/types.rs @@ -0,0 +1,155 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::clients::traits::{ + Chunking, ClientWrappingPipeline, Obfuscation, Reliability, RoutingSecurity, +}; +use crate::clients::{InputOptions, PipelinePayload}; +use crate::common::traits::{Framing, Transport, WireWrappingPipeline}; +use crate::{AddressedTimedData, AddressedTimedPayload}; + +/// Generic composition struct that implements [`ClientWrappingPipeline`] by +/// delegating each stage to a held component. +/// +/// Type parameters correspond to the six pipeline stages: +/// - `C`: [`Chunking`] +/// - `R`: [`Reliability`] +/// - `O`: [`Obfuscation`] +/// - `Rs`: [`RoutingSecurity`] +/// - `F`: [`Framing`] +/// - `T`: [`Transport`] +pub struct Pipeline { + pub packet_size: usize, + pub chunking: C, + pub reliability: R, + pub obfuscation: O, + pub security: Rs, + pub framing: F, + pub transport: T, +} + +impl Chunking for Pipeline +where + Opts: InputOptions, + C: Chunking, +{ + fn chunked( + &mut self, + input: Vec, + input_options: Opts, + chunk_size: usize, + timestamp: Ts, + ) -> Vec> { + self.chunking + .chunked(input, input_options, chunk_size, timestamp) + } +} + +impl Reliability for Pipeline +where + Opts: InputOptions, + R: Reliability, +{ + const OVERHEAD_SIZE: usize = R::OVERHEAD_SIZE; + + fn reliable_encode( + &mut self, + input: Option>, + timestamp: Ts, + ) -> Vec> { + self.reliability.reliable_encode(input, timestamp) + } +} + +impl Obfuscation for Pipeline +where + Opts: InputOptions, + O: Obfuscation, +{ + fn obfuscate( + &mut self, + input: Option>, + timestamp: Ts, + ) -> Vec> { + self.obfuscation.obfuscate(input, timestamp) + } +} + +impl RoutingSecurity + for Pipeline +where + Opts: InputOptions, + Rs: RoutingSecurity, +{ + const OVERHEAD_SIZE: usize = Rs::OVERHEAD_SIZE; + + fn nb_frames(&self) -> usize { + self.security.nb_frames() + } + + fn encrypt( + &mut self, + input: PipelinePayload, + ) -> PipelinePayload { + self.security.encrypt(input) + } +} + +impl Framing for Pipeline +where + F: Framing, +{ + type Frame = F::Frame; + const OVERHEAD_SIZE: usize = F::OVERHEAD_SIZE; + + fn to_frame( + &self, + payload: AddressedTimedPayload, + frame_size: usize, + ) -> Vec> { + self.framing.to_frame(payload, frame_size) + } +} + +impl Transport for Pipeline +where + T: Transport, +{ + type Frame = T::Frame; + const OVERHEAD_SIZE: usize = T::OVERHEAD_SIZE; + + fn to_transport_packet( + &self, + frame: AddressedTimedData, + ) -> AddressedTimedData { + self.transport.to_transport_packet(frame) + } +} + +impl WireWrappingPipeline + for Pipeline +where + Ts: Clone, + NdId: Clone, + F: Framing, + T: Transport, +{ + fn packet_size(&self) -> usize { + self.packet_size + } +} + +impl ClientWrappingPipeline + for Pipeline +where + Ts: Clone, + NdId: Clone, + Opts: InputOptions, + C: Chunking, + R: Reliability, + O: Obfuscation, + Rs: RoutingSecurity, + F: Framing, + T: Transport, +{ +} diff --git a/common/nym-lp-data/src/common/helpers.rs b/common/nym-lp-data/src/common/helpers.rs new file mode 100644 index 0000000000..695a933ab9 --- /dev/null +++ b/common/nym-lp-data/src/common/helpers.rs @@ -0,0 +1,100 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::{ + AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload, + common::traits::{ + Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline, + WireWrappingPipeline, + }, +}; + +/// Marker trait for a no-op [`WireWrappingPipeline`] implementation. +/// +/// Implement this for your pipeline type to get a [`WireWrappingPipeline`] impl that +/// passes the payload through unchanged with zero byte overhead. +pub trait NoOpWireWrapper { + const PACKET_SIZE: usize = 1500; +} + +impl Framing for T +where + T: NoOpWireWrapper, +{ + type Frame = Vec; + const OVERHEAD_SIZE: usize = 0; + fn to_frame( + &self, + payload: AddressedTimedPayload, + _: usize, + ) -> Vec> { + vec![payload] + } +} + +impl Transport for T +where + T: NoOpWireWrapper, + Pkt: From>, +{ + type Frame = Vec; + const OVERHEAD_SIZE: usize = 0; + fn to_transport_packet( + &self, + frame: AddressedTimedPayload, + ) -> AddressedTimedData { + frame.data_transform(|data| data.into()) + } +} + +impl WireWrappingPipeline for T +where + T: NoOpWireWrapper, + Ts: Clone, + Pkt: From>, + NdId: Clone, +{ + fn packet_size(&self) -> usize { + T::PACKET_SIZE + } +} + +/// Marker trait for a no-op [`WireUnwrappingPipeline`] implementation. +/// +/// Implement this for your pipeline type to get a [`WireUnwrappingPipeline`] impl that +/// passes the payload through unchanged. +pub trait NoOpWireUnwrapper {} + +impl FramingUnwrap for T +where + T: NoOpWireUnwrapper, + Mk: Default, +{ + type Frame = Vec; + fn frame_to_message(&mut self, frame: TimedPayload) -> Option<(TimedPayload, Mk)> { + Some((frame, Default::default())) + } +} + +impl TransportUnwrap for T +where + T: NoOpWireUnwrapper, + Pkt: Into>, +{ + type Frame = Vec; + fn packet_to_frame(&self, packet: Pkt, timestamp: Ts) -> anyhow::Result> { + Ok(TimedData { + timestamp, + data: packet.into(), + }) + } +} + +impl WireUnwrappingPipeline for T +where + T: NoOpWireUnwrapper, + Ts: Clone, + Pkt: Into>, + Mk: Default, +{ +} diff --git a/common/nym-lp-data/src/common/mod.rs b/common/nym-lp-data/src/common/mod.rs new file mode 100644 index 0000000000..5712be8ee1 --- /dev/null +++ b/common/nym-lp-data/src/common/mod.rs @@ -0,0 +1,5 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +pub mod helpers; +pub mod traits; diff --git a/common/nym-lp-data/src/common/traits.rs b/common/nym-lp-data/src/common/traits.rs new file mode 100644 index 0000000000..59636aaf00 --- /dev/null +++ b/common/nym-lp-data/src/common/traits.rs @@ -0,0 +1,169 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::{AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload}; + +/// Trait for applying framing to a timed payload. +/// +/// # Type Parameters +/// - `Ts`: Timestamp type carried by the `TimedPayload`. +/// +/// # Associated Types +/// - `Frame`: Frame type produced by the framing operation. +/// +/// # Associated Constants +/// - `OVERHEAD_SIZE`: Number of additional bytes added by the framing scheme. +/// +/// # Required Methods +/// - `to_frame`: Splits the payload into a `Vec>` of frames of the given size. +pub trait Framing { + type Frame; + const OVERHEAD_SIZE: usize; + fn to_frame( + &self, + payload: AddressedTimedPayload, + frame_size: usize, + ) -> Vec>; +} + +/// Trait for unwrapping framing from a frame back into a payload. +/// +/// # Type Parameters +/// - `Ts`: Timestamp type carried by the `TimedPayload`. +/// - `Mk`: Enum describing the kind of message that can be returned. +/// +/// # Associated Types +/// - `Frame`: Frame type consumed as input. +/// +/// # Required Methods +/// - `frame_to_message`: Attempts to reassemble a payload from the given frame, returning +/// `Some((payload, kind))` when a complete message is available, or `None` otherwise. +pub trait FramingUnwrap { + type Frame; + fn frame_to_message( + &mut self, + frame: TimedData, + ) -> Option<(TimedPayload, Mk)>; +} + +/// Trait for applying a transport layer to a framed payload. +/// +/// # Type Parameters +/// - `Ts`: Timestamp type carried by the `TimedPayload`. +/// - `Pkt`: Transport packet type produced as output. +/// +/// # Associated Types +/// - `Frame`: Frame type consumed as input. +/// +/// # Associated Constants +/// - `OVERHEAD_SIZE`: Number of additional bytes added by the transport scheme. +/// +/// # Required Methods +/// - `to_transport_packet`: Wraps a frame into a transport packet. +pub trait Transport { + type Frame; + const OVERHEAD_SIZE: usize; + fn to_transport_packet( + &self, + frame: AddressedTimedData, + ) -> AddressedTimedData; +} + +/// Trait for unwrapping a transport packet back into a frame. +/// +/// # Type Parameters +/// - `Ts`: Timestamp type carried by the `TimedPayload`. +/// - `Pkt`: Transport packet type consumed as input. +/// +/// # Associated Types +/// - `Frame`: Frame type produced as output. +/// +/// # Required Methods +/// - `packet_to_frame`: Strips the transport layer from a packet, returning the inner frame +/// tagged with the given timestamp. +pub trait TransportUnwrap { + type Frame; + fn packet_to_frame( + &self, + packet: Pkt, + timestamp: Ts, + ) -> anyhow::Result>; +} + +/// Supertrait combining [`Framing`] and [`Transport`] into a reusable wire-wrapping layer. +/// +/// Used as the bottom stage of any outbound pipeline (client or mixnode). +/// +/// # Type Parameters +/// - `Ts`: Timestamp type. +/// - `Pkt`: Final transport packet type. +/// +/// Both [`Framing`] and [`Transport`] declare their own `type Frame`; this +/// supertrait cross-constrains them so `to_frame`'s output feeds directly into +/// `to_transport_packet`. +/// +/// # Required Methods +/// - `packet_size`: Total on-wire size of an output packet in bytes. +/// +/// # Provided Methods +/// - `frame_size`: Derived from `packet_size` minus transport and framing overheads. +/// - `wire_wrap`: Frames a payload and wraps each frame into a transport packet. +pub trait WireWrappingPipeline: + Transport + + Framing>::Frame> +where + Ts: Clone, + NdId: Clone, +{ + fn packet_size(&self) -> usize; + + fn frame_size(&self) -> usize { + self.packet_size() + - >::OVERHEAD_SIZE + - >::OVERHEAD_SIZE + } + + fn wire_wrap( + &self, + payload: AddressedTimedPayload, + ) -> Vec> { + let frame_size = self.frame_size(); + self.to_frame(payload, frame_size) + .into_iter() + .map(|frame| self.to_transport_packet(frame)) + .collect() + } +} + +/// Supertrait combining [`TransportUnwrap`] and [`FramingUnwrap`] into a reusable +/// wire-unwrapping layer. +/// +/// Used as the bottom stage of any inbound pipeline (client or mixnode). +/// +/// # Type Parameters +/// - `Ts`: Timestamp type. +/// - `Pkt`: Transport packet type consumed as input. +/// - `Mk`: Message-kind marker returned alongside the reassembled payload. +/// +/// Both [`TransportUnwrap`] and [`FramingUnwrap`] declare their own `type Frame`; +/// this supertrait cross-constrains them so `packet_to_frame`'s output feeds +/// directly into `frame_to_message`. +/// +/// # Provided Methods +/// - `wire_unwrap`: Strips the transport layer from a packet and attempts to reassemble +/// a payload, returning `Some((payload, kind))` when a complete message is available. +pub trait WireUnwrappingPipeline: + TransportUnwrap + + FramingUnwrap>::Frame> +where + Ts: Clone, +{ + fn wire_unwrap( + &mut self, + input: Pkt, + timestamp: Ts, + ) -> anyhow::Result, Mk)>> { + let frame = self.packet_to_frame(input, timestamp)?; + Ok(self.frame_to_message(frame)) + } +} diff --git a/common/nym-lp-data/src/lib.rs b/common/nym-lp-data/src/lib.rs index 46f172a2fd..5b100cc86a 100644 --- a/common/nym-lp-data/src/lib.rs +++ b/common/nym-lp-data/src/lib.rs @@ -1,4 +1,149 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +//! Trait definitions and data structures for low-level packet (LP) processing +//! pipelines in the Nym mixnet. +//! +//! ## Crate layout +//! +//! | Module | Purpose | +//! |--------|---------| +//! | [`clients`] | Client-side pipeline traits and types: chunking, reliability, obfuscation, routing security, framing, transport | +//! | [`common`] | Shared framing and transport traits used by both clients and mixnodes | +//! | [`mixnodes`] | Mixnode-side pipeline traits: unwrap incoming packets, re-wrap and forward them | +//! +//! ## Core types +//! +//! [`TimedData`] is the foundational wrapper that pairs any piece of data with a +//! timestamp, threading timing information through every stage of the pipeline. +//! [`TimedPayload`] is a convenience alias for `TimedData>`. + +use std::fmt::Debug; + +pub mod clients; +pub mod common; +pub mod mixnodes; pub mod packet; + +/// A value of type `D` tagged with a timestamp of type `Ts`. +/// +/// `TimedData` threads timing information through every stage of the LP +/// pipeline. It is produced by [`clients::traits::Chunking`] and propagated +/// unchanged (or with the timestamp transformed) through every subsequent +/// pipeline stage until the packet is sent on the wire. +pub struct TimedData { + pub timestamp: Ts, + pub data: D, +} + +impl Debug for TimedData +where + D: Debug, + Ts: Debug, +{ + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + writeln!(f, "TimedData {{")?; + writeln!(f, " data:")?; + let data_debug = format!("{:#?}", &self.data); + for line in data_debug.lines() { + writeln!(f, " {}", line)?; + } + writeln!(f, " timestamp: {:#?},", &self.timestamp)?; + write!(f, "}}") + } +} + +impl TimedData { + pub fn new(timestamp: Ts, data: D) -> Self { + TimedData { timestamp, data } + } + /// Apply `op` to the data component, leaving the timestamp unchanged. + /// `Nd` can be a different type to allow type transform as well + pub fn data_transform(self, mut op: F) -> TimedData + where + F: FnMut(D) -> Nd, + { + TimedData { + data: op(self.data), + timestamp: self.timestamp, + } + } + + /// Apply `op` to the timestamp component, leaving the data unchanged. + pub fn ts_transform(self, mut op: F) -> Self + where + F: FnMut(Ts) -> Ts, + { + TimedData { + data: self.data, + timestamp: op(self.timestamp), + } + } +} + +pub struct AddressedTimedData { + pub dst: NdId, + pub data: TimedData, +} + +impl AddressedTimedData { + pub fn new(timestamp: Ts, data: D, dst: NdId) -> Self { + AddressedTimedData { + dst, + data: TimedData::new(timestamp, data), + } + } + + /// Apply `op` to the data component, leaving the timestamp unchanged. + /// `Nd` can be a different type to allow type transform as well + pub fn data_transform(self, op: F) -> AddressedTimedData + where + F: FnMut(D) -> Nd, + { + AddressedTimedData { + dst: self.dst, + data: self.data.data_transform(op), + } + } + + /// Apply `op` to the timestamp component, leaving the data unchanged. + pub fn ts_transform(self, op: F) -> Self + where + F: FnMut(Ts) -> Ts, + { + AddressedTimedData { + dst: self.dst, + data: self.data.ts_transform(op), + } + } +} + +impl Debug for AddressedTimedData +where + D: Debug, + Ts: Debug, + NdId: Debug, +{ + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + writeln!(f, "AddressedTimedData {{")?; + writeln!(f, " dst: {:#?}", &self.dst)?; + writeln!(f, " data: {:#?}", &self.data)?; + write!(f, "}}") + } +} + +impl From<(NdId, TimedData)> for AddressedTimedData { + fn from(value: (NdId, TimedData)) -> Self { + AddressedTimedData { + dst: value.0, + data: value.1, + } + } +} + +/// Convenience alias for a [`TimedData`] whose payload is a raw byte buffer. +/// +/// Used as the input and output type for most pipeline stages before the data +/// is wrapped into a typed frame or packet. +pub type TimedPayload = TimedData>; +pub type AddressedTimedPayload = AddressedTimedData, NdId>; diff --git a/common/nym-lp-data/src/mixnodes/mod.rs b/common/nym-lp-data/src/mixnodes/mod.rs new file mode 100644 index 0000000000..ecb11512ec --- /dev/null +++ b/common/nym-lp-data/src/mixnodes/mod.rs @@ -0,0 +1,5 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +pub mod traits; +pub mod types; diff --git a/common/nym-lp-data/src/mixnodes/traits.rs b/common/nym-lp-data/src/mixnodes/traits.rs new file mode 100644 index 0000000000..34eb1383b7 --- /dev/null +++ b/common/nym-lp-data/src/mixnodes/traits.rs @@ -0,0 +1,61 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::{AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload}; + +use crate::common::traits::{WireUnwrappingPipeline, WireWrappingPipeline}; + +/// Top-level processing trait for a mix node. +/// +/// Combines [`WireUnwrappingPipeline`] and [`WireWrappingPipeline`] with a blank [`mix`] +/// step that the implementor fills in (decrypt, route, re-encrypt, cover traffic, etc.). +/// +/// # Type Parameters +/// - `Ts`: Timestamp / tick-context type. +/// - `Pkt`: Transport packet type; the same type is consumed and produced. +/// - `Mk`: Message-kind marker returned by the unwrap side. +/// - `NdId`: Identifier type for the next-hop destination. +/// +/// Frame types are owned by the wire sub-traits as associated items and do not +/// appear in this trait's parameter list. +/// +/// # Required Methods +/// - `mix`: Given a reassembled payload and the current timestamp, return zero or more +/// `(next_hop, payload)` pairs to be re-wrapped and forwarded. +/// +/// # Provided Methods +/// - `process`: Unwraps the incoming packet via [`WireUnwrappingPipeline::wire_unwrap`], +/// passes the result to [`mix`], and re-wraps each output payload via +/// [`WireWrappingPipeline::wire_wrap`]. +pub trait MixnodeProcessingPipeline: + WireUnwrappingPipeline + WireWrappingPipeline +where + Ts: Clone, + NdId: Clone, +{ + fn mix( + &mut self, + message_kind: Mk, + payload: TimedPayload, + timestamp: Ts, + ) -> Vec>; + + fn process( + &mut self, + input: TimedData, + timestamp: Ts, + ) -> anyhow::Result>> { + let TimedData { + data: packet, + timestamp: ts, + } = input; + let Some((payload, kind)) = self.wire_unwrap(packet, ts)? else { + return Ok(Vec::new()); + }; + let mixed = self.mix(kind, payload, timestamp); + Ok(mixed + .into_iter() + .flat_map(|addressed_data| self.wire_wrap(addressed_data).into_iter()) + .collect()) + } +} diff --git a/common/nym-lp-data/src/mixnodes/types.rs b/common/nym-lp-data/src/mixnodes/types.rs new file mode 100644 index 0000000000..5d53f83a0b --- /dev/null +++ b/common/nym-lp-data/src/mixnodes/types.rs @@ -0,0 +1,56 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::common::traits::{Framing, Transport, WireWrappingPipeline}; +use crate::{AddressedTimedData, AddressedTimedPayload}; + +/// The generic pipeline struct for a mixnode +pub struct Pipeline { + pub packet_size: usize, + pub framing: F, + pub transport: T, + _marker: std::marker::PhantomData, +} + +impl Framing for Pipeline +where + F: Framing, +{ + type Frame = F::Frame; + const OVERHEAD_SIZE: usize = F::OVERHEAD_SIZE; + + fn to_frame( + &self, + payload: AddressedTimedPayload, + frame_size: usize, + ) -> Vec> { + self.framing.to_frame(payload, frame_size) + } +} + +impl Transport for Pipeline +where + T: Transport, +{ + type Frame = T::Frame; + const OVERHEAD_SIZE: usize = T::OVERHEAD_SIZE; + + fn to_transport_packet( + &self, + frame: AddressedTimedData, + ) -> AddressedTimedData { + self.transport.to_transport_packet(frame) + } +} + +impl WireWrappingPipeline for Pipeline +where + Ts: Clone, + NdId: Clone, + F: Framing, + T: Transport, +{ + fn packet_size(&self) -> usize { + self.packet_size + } +} diff --git a/common/nym-lp-data/tests/integration/common/mod.rs b/common/nym-lp-data/tests/integration/common/mod.rs new file mode 100644 index 0000000000..22b2885ca6 --- /dev/null +++ b/common/nym-lp-data/tests/integration/common/mod.rs @@ -0,0 +1,220 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_lp::packet::{ + LpFrame, LpHeader, LpPacket, + frame::{LpFrameHeader, LpFrameKind}, +}; + +use nym_lp_data::{ + AddressedTimedData, + common::traits::{Framing, Transport}, +}; +use nym_lp_data::{ + AddressedTimedPayload, + clients::{ + InputOptions, PipelinePayload, + traits::{Chunking, Obfuscation, Reliability, RoutingSecurity}, + }, +}; + +#[derive(Clone, Copy)] +pub struct BasicOptions { + pub reliability: bool, + pub security: bool, + pub obfuscation: bool, + pub next_hop: u8, +} + +impl InputOptions for BasicOptions { + fn reliability(&self) -> bool { + self.reliability + } + + fn routing_security(&self) -> bool { + self.security + } + + fn obfuscation(&self) -> bool { + self.obfuscation + } + + fn next_hop(&self) -> u8 { + self.next_hop + } +} + +pub type BasicPipelinePayload = PipelinePayload; + +pub struct MockChunking; +impl Chunking for MockChunking +where + Ts: Clone, +{ + fn chunked( + &mut self, + input: Vec, + input_options: BasicOptions, + chunk_size: usize, + timestamp: Ts, + ) -> Vec> { + input + .chunks(chunk_size) + .map(|chunk| { + BasicPipelinePayload::new(timestamp.clone(), chunk.to_vec(), input_options) + }) + .collect() + } +} + +pub struct KcpReliability; + +impl KcpReliability { + const HEADER: &[u8; 5] = b"0KCP0"; +} + +impl Reliability for KcpReliability { + const OVERHEAD_SIZE: usize = Self::HEADER.len(); + fn reliable_encode( + &mut self, + input: Option>, + _: Ts, + ) -> Vec> { + input + .map(|data| { + vec![data.data_transform(|data| { + let mut packet = Self::HEADER.to_vec(); + packet.extend(data); + packet + })] + }) + .unwrap_or_default() + } +} + +pub struct SphinxSecurity { + pub nb_frames: usize, +} + +impl SphinxSecurity { + const HEADER: &[u8; 8] = b"0SPHINX0"; +} + +impl RoutingSecurity for SphinxSecurity { + const OVERHEAD_SIZE: usize = Self::HEADER.len(); + + fn nb_frames(&self) -> usize { + self.nb_frames + } + + fn encrypt(&mut self, input: BasicPipelinePayload) -> BasicPipelinePayload { + input.data_transform(|data| { + let mut packet = Self::HEADER.to_vec(); + packet.extend(data); + packet + }) + } +} + +pub struct KekwObfuscation; + +impl Obfuscation for KekwObfuscation { + fn obfuscate( + &mut self, + input: Option>, + _timestamp: u32, + ) -> Vec> { + if let Some(input) = input { + vec![input.ts_transform(|ts| ts + 1)] + } else { + Vec::new() + } + } +} + +#[allow(dead_code)] +pub struct ReallyOddObfuscation { + next_ts: u32, +} + +impl ReallyOddObfuscation { + #[allow(dead_code)] + pub fn new(start_ts: u32) -> Self { + let next_ts = if !start_ts.is_multiple_of(2) { + start_ts + } else { + start_ts + 1 + }; + Self { next_ts } + } +} + +impl Obfuscation for ReallyOddObfuscation { + fn obfuscate( + &mut self, + input: Option>, + _timestamp: u32, + ) -> Vec> { + if let Some(input) = input { + let pkt = input.ts_transform(|_| self.next_ts); + self.next_ts += 2; + vec![pkt] + } else { + Vec::new() + } + } +} + +pub struct LpFraming; + +impl LpFraming { + const FRAME_ATTRIBUTES: &[u8; 14] = b"0LpFrameAttrs0"; +} + +impl Framing for LpFraming +where + Ts: Clone, +{ + type Frame = LpFrame; + const OVERHEAD_SIZE: usize = LpFrameHeader::SIZE; + fn to_frame( + &self, + input: AddressedTimedPayload, + frame_size: usize, + ) -> Vec> { + input + .data + .data + .chunks(frame_size) + .map(|frame_payload| { + let header = LpFrameHeader::new(LpFrameKind::Opaque, *Self::FRAME_ATTRIBUTES); + + AddressedTimedData::new( + input.data.timestamp.clone(), + LpFrame { + header, + content: frame_payload.to_vec().into(), + }, + input.dst, + ) + }) + .collect() + } +} + +pub struct LpTransport; + +impl Transport for LpTransport { + type Frame = LpFrame; + const OVERHEAD_SIZE: usize = LpHeader::SIZE; + fn to_transport_packet( + &self, + input: AddressedTimedData, + ) -> AddressedTimedData { + AddressedTimedData::new( + input.data.timestamp, + LpPacket::new(LpHeader::new(7, 7, 7), input.data.data), + input.dst, + ) + } +} diff --git a/common/nym-lp-data/tests/integration/main.rs b/common/nym-lp-data/tests/integration/main.rs new file mode 100644 index 0000000000..2a2e8ba82f --- /dev/null +++ b/common/nym-lp-data/tests/integration/main.rs @@ -0,0 +1,34 @@ +// Copyright 2026 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_lp_data::clients::{traits::ClientWrappingPipeline, types::Pipeline}; + +use crate::common::{ + KcpReliability, KekwObfuscation, LpFraming, LpTransport, MockChunking, SphinxSecurity, +}; + +mod common; + +#[test] +fn empty_input_yields_empty_output() { + let packet_size = 64; + let security_layer_nb_frames = 2; + + let mut mock_pipeline = Pipeline { + chunking: MockChunking, + reliability: KcpReliability, + security: SphinxSecurity { + nb_frames: security_layer_nb_frames, + }, + obfuscation: KekwObfuscation, + framing: LpFraming, + transport: LpTransport, + packet_size, + }; + + let output = mock_pipeline.process(None, 1); + + assert!(output.is_empty()); +} + +// TODO More test to come later