nym-mix-sim crate
This commit is contained in:
Simon Wicky
@@ -0,0 +1,45 @@
|
||||
[package]
|
||||
name = "nym-mix-sim"
|
||||
version.workspace = true
|
||||
authors.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
homepage.workspace = true
|
||||
documentation.workspace = true
|
||||
|
||||
[[bin]]
|
||||
name = "mix-client"
|
||||
path = "src/bin/mix_client.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "nym-mix-sim"
|
||||
path = "src/main.rs"
|
||||
|
||||
[dependencies]
|
||||
anyhow.workspace = true
|
||||
clap = { workspace = true, features = ["derive"] }
|
||||
rand.workspace = true
|
||||
rand_distr.workspace = true
|
||||
tokio = { workspace = true, features = [
|
||||
"rt-multi-thread",
|
||||
"macros",
|
||||
"time",
|
||||
"signal",
|
||||
] }
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
serde_json.workspace = true
|
||||
tracing.workspace = true
|
||||
tracing-subscriber = { workspace = true, features = ["env-filter"] }
|
||||
strum = { workspace = true, features = ["derive"] }
|
||||
uuid = { workspace = true, features = ["v4", "serde"] }
|
||||
|
||||
nym-bin-common = { workspace = true, features = ["basic_tracing"] }
|
||||
nym-common.workspace = true
|
||||
nym-lp-data.workspace = true
|
||||
nym-sphinx.workspace = true
|
||||
nym-crypto = { workspace = true, features = ["serde", "rand"] }
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
@@ -0,0 +1,162 @@
|
||||
# nym-mix-sim
|
||||
|
||||
A discrete-time simulator for the Nym mixnet, intended for local testing and experimentation. It models a small network of mix nodes and clients exchanging UDP packets on localhost, allowing you to observe packet flow, experiment with different drivers, and debug routing behaviour step by step.
|
||||
|
||||
## Overview
|
||||
|
||||
The simulator runs a configurable number of mix nodes and clients on localhost, each bound to its own UDP port. Time advances in **ticks** — each tick runs the client phase, then drains incoming sockets, processes packets through the mixing pipeline, and dispatches outgoing packets.
|
||||
|
||||
Two binaries are provided:
|
||||
|
||||
| Binary | Purpose |
|
||||
|--------|---------|
|
||||
| `nym-mix-sim` | Main simulator: topology generation and tick-loop execution |
|
||||
| `mix-client` | Standalone tool to inject messages into a running simulation |
|
||||
|
||||
## Quick Start
|
||||
|
||||
```bash
|
||||
# 1. Generate a topology with 6 nodes and 2 clients
|
||||
cargo run --bin nym-mix-sim -- init-topology
|
||||
|
||||
# 2. Run the simulation (automatic mode, 1ms ticks, default discrete-sphinx driver)
|
||||
cargo run --bin nym-mix-sim -- run
|
||||
|
||||
# 3. In a separate terminal, send a message between the two clients
|
||||
cargo run --bin mix-client -- --src 6 --dst 7
|
||||
# Then type a message and press ENTER
|
||||
```
|
||||
|
||||
## Commands
|
||||
|
||||
### `init-topology`
|
||||
|
||||
Generates a `topology.json` file describing nodes and clients.
|
||||
|
||||
```
|
||||
cargo run --bin nym-mix-sim -- init-topology [OPTIONS]
|
||||
```
|
||||
|
||||
| Option | Default | Description |
|
||||
|--------|---------|-------------|
|
||||
| `--nodes <N>` | `6` | Number of mix nodes |
|
||||
| `--clients <N>` | `2` | Number of clients |
|
||||
| `--output <PATH>` | `topology.json` | Output file path |
|
||||
|
||||
Nodes are assigned sequential ports starting at `127.0.0.1:9000`. Clients get two sockets each: a mix-facing socket starting at `127.0.0.1:9500` and an app-facing socket starting at `127.0.0.1:9600`. Each node gets a freshly generated X25519 key pair (used by Sphinx drivers).
|
||||
|
||||
### `run`
|
||||
|
||||
Starts the simulation loop.
|
||||
|
||||
```
|
||||
cargo run --bin nym-mix-sim -- run [OPTIONS]
|
||||
```
|
||||
|
||||
| Option | Default | Description |
|
||||
|--------|---------|-------------|
|
||||
| `--topology <PATH>` | `topology.json` | Topology file to load |
|
||||
| `--driver <DRIVER>` | `discrete-sphinx` | Simulation driver (see below) |
|
||||
| `--tick-duration-ms <MS>` | `1` | Milliseconds between automatic ticks |
|
||||
| `--manual` | off | Enable manual stepping mode (ENTER per tick) |
|
||||
| `--no-display-state` | off | Suppress per-phase state dump in manual mode |
|
||||
|
||||
### `mix-client`
|
||||
|
||||
Injects messages into a running simulation from stdin.
|
||||
|
||||
```
|
||||
cargo run --bin mix-client -- --src <ID> --dst <ID> [--topology <PATH>]
|
||||
```
|
||||
|
||||
Reads lines from stdin and sends each as a payload routed from client `--src` through the mix network to client `--dst`. Client IDs begin where node IDs end (e.g., with 6 nodes and 2 clients, client IDs are `6` and `7`).
|
||||
|
||||
## Drivers
|
||||
|
||||
The driver controls how packets are formatted, encrypted, and routed.
|
||||
|
||||
| Driver | Timestamp | Encryption | Cover traffic | Reliability | Manual mode |
|
||||
|--------|-----------|------------|---------------|-------------|-------------|
|
||||
| `simple` | Discrete (u32 tick counter) | None | No | No | Yes |
|
||||
| `sphinx` | Wall-clock (`Instant`) | Full Sphinx | Yes (Poisson) | SURB ACKs | No |
|
||||
| `discrete-sphinx` | Discrete (u32 tick counter) | Full Sphinx | Yes (Poisson) | SURB ACKs | Yes |
|
||||
|
||||
**`simple`** — Each packet is a fixed 64-byte frame (16-byte UUID + 48-byte payload). Nodes forward to `node_id + 1`. No cryptography. Best for sanity-checking the topology and observing raw packet flow.
|
||||
|
||||
**`sphinx`** — Uses `nym_sphinx::SphinxPacket` for full onion encryption. Clients build a 3-hop route, generate a SURB ACK reliability layer, and run two Poisson cover-traffic loops. Per-hop delays are extracted from the decrypted packet and scheduled using real wall-clock time. Automatic mode only.
|
||||
|
||||
**`discrete-sphinx`** — Same Sphinx encryption, SURB ACKs, and cover traffic as `sphinx`, but uses a u32 tick counter instead of wall-clock time (1 tick = 1 ms). This makes timing deterministic and compatible with `--manual` mode. Default driver.
|
||||
|
||||
## Tick Mechanics
|
||||
|
||||
Each tick runs four phases across all participants:
|
||||
|
||||
1. **Clients** — every client drains its app socket, runs new payloads through the wrapping pipeline, processes any inbound mix packets, and forwards queued packets whose scheduled timestamp is due.
|
||||
2. **Nodes — incoming** — every node drains its UDP socket (non-blocking) and buffers received packets.
|
||||
3. **Nodes — processing** — buffered packets pass through the mixing pipeline. For Sphinx nodes, this means decryption and routing extraction. Each processed packet is queued with a scheduled dispatch timestamp.
|
||||
4. **Nodes — outgoing** — packets whose timestamp ≤ current tick are serialised and sent via UDP to the next hop.
|
||||
|
||||
In manual mode, the node state is pretty-printed between phases 2 and 3, and again between 3 and 4 (unless `--no-display-state` is set).
|
||||
|
||||
## Speed Controls
|
||||
|
||||
**Tick duration** (`--tick-duration-ms`) controls how fast the simulation runs:
|
||||
|
||||
- `0` — maximum speed, no sleep between ticks
|
||||
- `1` (default) — roughly real-time for discrete drivers
|
||||
- Any value `N > 1` — slows the simulation down linearly; in practice a value of `N` will make the simulation `N` times slower than real time
|
||||
|
||||
**Manual mode** (`--manual`) pauses after every tick and waits for ENTER. Completely deterministic — no timing overhead, step through packet sequences one tick at a time. Only available with the `simple` and `discrete-sphinx` drivers.
|
||||
|
||||
**Discrete vs wall-clock timestamps** — Discrete (u32) timestamps have minimal overhead and allow the simulation to run faster than real time. Wall-clock (`Instant`) timestamps tie delays to real elapsed time, which is more realistic but limits simulation speed.
|
||||
|
||||
## Topology File
|
||||
|
||||
`topology.json` is generated by `init-topology` and consumed by `run` and `mix-client`.
|
||||
|
||||
```json
|
||||
{
|
||||
"nodes": [
|
||||
{
|
||||
"node_id": 0,
|
||||
"socket_address": "127.0.0.1:9000",
|
||||
"reliability": 100,
|
||||
"sphinx_private_key": "<bs58-encoded X25519 key>"
|
||||
}
|
||||
],
|
||||
"clients": [
|
||||
{
|
||||
"client_id": 6,
|
||||
"mixnet_address": "127.0.0.1:9506",
|
||||
"app_address": "127.0.0.1:9606"
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
The `reliability` field is reserved for future use.
|
||||
|
||||
## Logging
|
||||
|
||||
Set `RUST_LOG` to control verbosity:
|
||||
|
||||
```bash
|
||||
RUST_LOG=debug cargo run --bin nym-mix-sim -- run
|
||||
RUST_LOG=warn cargo run --bin nym-mix-sim -- run # quiet
|
||||
```
|
||||
|
||||
Default level is `info`. Logs go to stderr; received message content goes to stdout.
|
||||
|
||||
## Example: Manual Sphinx Walk-Through
|
||||
|
||||
```bash
|
||||
# Terminal 1 — run in manual mode, one tick at a time
|
||||
cargo run --bin nym-mix-sim -- run --driver discrete-sphinx --manual
|
||||
|
||||
# Terminal 2 — send a message from client 6 to client 7
|
||||
cargo run --bin mix-client -- --src 6 --dst 7
|
||||
> hello
|
||||
|
||||
# Back in Terminal 1, press ENTER to advance each tick and observe
|
||||
# the encrypted packet hop through each node
|
||||
```
|
||||
@@ -0,0 +1,102 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Standalone client CLI — inject packets into a running mix-sim.
|
||||
//!
|
||||
//! Reads lines from stdin and, on each ENTER, sends the text as a raw payload
|
||||
//! to the app socket of the running client identified by `--src`. The client
|
||||
//! wraps it into the active wire format (e.g. [`SimplePacket`] for the simple
|
||||
//! driver, an onion-encrypted Sphinx packet for the Sphinx drivers) and forwards
|
||||
//! it through the mix network to the destination client `--dst`.
|
||||
//!
|
||||
//! ## Message format (app-socket datagram)
|
||||
//!
|
||||
//! ```text
|
||||
//! ┌────────────────────────┬─────────────────────┐
|
||||
//! │ dst_client_id (1 B) │ raw payload bytes │
|
||||
//! └────────────────────────┴─────────────────────┘
|
||||
//! ```
|
||||
//!
|
||||
//! The running client's `tick_app_incoming` parses this datagram on the next tick.
|
||||
//!
|
||||
//! ## Usage
|
||||
//!
|
||||
//! ```text
|
||||
//! cargo run --bin mix-client -- --topology topology.json --src 6 --dst 7
|
||||
//! ```
|
||||
|
||||
use std::net::UdpSocket;
|
||||
|
||||
use clap::Parser;
|
||||
use nym_mix_sim::{client::ClientId, topology::Topology};
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(
|
||||
name = "mix-client",
|
||||
about = "Send stdin lines into a running nym-mix-sim"
|
||||
)]
|
||||
struct Cli {
|
||||
/// Path to the topology.json file.
|
||||
#[arg(short, long, default_value = "topology.json")]
|
||||
topology: String,
|
||||
|
||||
/// ID of the client (in the topology) to deliver packets through.
|
||||
#[arg(short, long)]
|
||||
src: ClientId,
|
||||
|
||||
/// ID of the destination client packets should be routed toward.
|
||||
#[arg(short, long)]
|
||||
dst: ClientId,
|
||||
}
|
||||
|
||||
fn main() -> anyhow::Result<()> {
|
||||
nym_bin_common::logging::setup_tracing_logger();
|
||||
|
||||
let cli = Cli::parse();
|
||||
|
||||
let topology_data = std::fs::read_to_string(&cli.topology)?;
|
||||
let topology: Topology = serde_json::from_str(&topology_data)?;
|
||||
|
||||
let client = topology
|
||||
.clients
|
||||
.iter()
|
||||
.find(|c| c.client_id == cli.src)
|
||||
.ok_or_else(|| anyhow::anyhow!("no client with id {}", cli.src))?;
|
||||
|
||||
let app_addr = client.app_address;
|
||||
|
||||
// Bind an ephemeral socket to send from.
|
||||
let socket = UdpSocket::bind("127.0.0.1:0")?;
|
||||
|
||||
println!(
|
||||
"Ready — type a message and press ENTER to send to client {} via client {}.",
|
||||
cli.dst, cli.src
|
||||
);
|
||||
println!("(Ctrl-C to quit)");
|
||||
|
||||
let mut line = String::new();
|
||||
loop {
|
||||
line.clear();
|
||||
if std::io::stdin().read_line(&mut line)? == 0 {
|
||||
break; // EOF
|
||||
}
|
||||
|
||||
let text = line.trim_end_matches('\n').trim_end_matches('\r');
|
||||
let bytes = text.as_bytes();
|
||||
|
||||
// Prepend the destination node ID.
|
||||
let mut msg = Vec::with_capacity(1 + bytes.len());
|
||||
msg.push(cli.dst);
|
||||
msg.extend_from_slice(bytes);
|
||||
|
||||
socket.send_to(&msg, app_addr)?;
|
||||
println!(
|
||||
"Sent {} byte(s) of payload to client {} → client {}.",
|
||||
bytes.len(),
|
||||
cli.src,
|
||||
cli.dst
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,273 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use std::{fmt::Debug, io::ErrorKind, net::UdpSocket, sync::Arc};
|
||||
|
||||
use nym_lp_data::AddressedTimedData;
|
||||
|
||||
use crate::{
|
||||
node::NodeId,
|
||||
packet::WirePacketFormat,
|
||||
topology::{TopologyClient, directory::Directory},
|
||||
};
|
||||
|
||||
pub mod simple;
|
||||
pub mod sphinx;
|
||||
|
||||
/// Compact identifier for a simulated client.
|
||||
pub type ClientId = NodeId;
|
||||
|
||||
/// Driver-facing interface for a simulated client.
|
||||
///
|
||||
/// Erases `Fr`, `Pkt`, and `Mk` so that [`MixSimDriver`] only needs `Ts`.
|
||||
/// Implemented by [`SimpleClient<Ts, Fr, Pkt, Mk>`] and any other concrete
|
||||
/// client types.
|
||||
///
|
||||
/// [`MixSimDriver`]: crate::driver::MixSimDriver
|
||||
pub trait MixSimClient<Ts: Clone + PartialOrd + Debug + Send>: Send {
|
||||
fn tick(&mut self, timestamp: Ts);
|
||||
}
|
||||
|
||||
/// Pipeline interface used by [`BaseClient`] to convert raw app payloads into
|
||||
/// wire packets and to unwrap received packets back into plaintext.
|
||||
///
|
||||
/// `SndPkt` is the outgoing packet type (e.g. [`SimplePacket`] or
|
||||
/// [`SimMixPacket`]). `RcvPkt` defaults to `SndPkt` but can differ when the
|
||||
/// inbound and outbound wire formats diverge (e.g. the Sphinx client receives
|
||||
/// raw `Vec<u8>` final-hop payloads from nodes).
|
||||
///
|
||||
/// [`SimplePacket`]: crate::packet::simple::SimplePacket
|
||||
/// [`SimMixPacket`]: crate::packet::sphinx::SimMixPacket
|
||||
pub trait ProcessingClient<Ts, SndPkt, RcvPkt = SndPkt>: Send {
|
||||
/// Wrap `input` into one or more outbound packets addressed toward `dst`.
|
||||
fn process(
|
||||
&mut self,
|
||||
input: Vec<u8>,
|
||||
dst: ClientId,
|
||||
timestamp: Ts,
|
||||
) -> Vec<AddressedTimedData<Ts, SndPkt, NodeId>>;
|
||||
|
||||
/// Unwrap an inbound packet received from the mix network.
|
||||
///
|
||||
/// Returns `Ok(Some(plaintext))` for a real message, `Ok(None)` when the
|
||||
/// packet is cover traffic or an incomplete fragment, and `Err` when
|
||||
/// decryption or deserialisation fails.
|
||||
fn unwrap(&mut self, input: RcvPkt, timestamp: Ts) -> anyhow::Result<Option<Vec<u8>>>;
|
||||
}
|
||||
|
||||
/// Shared UDP transport layer for simulated clients.
|
||||
///
|
||||
/// Encapsulates both sockets, the routing directory, and the client id so that
|
||||
/// multiple concrete client types can reuse `send_to_node`, `recv_from_mix`,
|
||||
/// and `recv_from_app` without duplicating that logic. Packet types are
|
||||
/// method-level generics so `BaseClient` itself has no type parameters.
|
||||
pub struct BaseClient<Ts, Pc, SndPkt, RcvPkt = SndPkt> {
|
||||
id: ClientId,
|
||||
mix_socket: UdpSocket,
|
||||
app_socket: UdpSocket,
|
||||
directory: Arc<Directory>,
|
||||
|
||||
/// Packets that have been processed and are waiting to be forwarded to their
|
||||
/// first-hop node, sorted (loosely) by scheduled send timestamp.
|
||||
outgoing_queue: Vec<AddressedTimedData<Ts, SndPkt, NodeId>>,
|
||||
|
||||
processing_client: Pc,
|
||||
|
||||
/// Phantom data to carry the `RcvPkt` type parameter without storing a value.
|
||||
_marker: std::marker::PhantomData<RcvPkt>,
|
||||
}
|
||||
|
||||
impl<Ts, Pc, SndPkt, RcvPkt> BaseClient<Ts, Pc, SndPkt, RcvPkt> {
|
||||
/// Bind both UDP sockets to the given addresses.
|
||||
pub(crate) fn with_pipeline(
|
||||
topology_client: &TopologyClient,
|
||||
directory: Arc<Directory>,
|
||||
processing_client: Pc,
|
||||
) -> anyhow::Result<Self> {
|
||||
let mix_socket = UdpSocket::bind(topology_client.mixnet_address)?;
|
||||
mix_socket.set_nonblocking(true)?;
|
||||
|
||||
let app_socket = UdpSocket::bind(topology_client.app_address)?;
|
||||
app_socket.set_nonblocking(true)?;
|
||||
|
||||
Ok(Self {
|
||||
id: topology_client.client_id,
|
||||
mix_socket,
|
||||
app_socket,
|
||||
directory,
|
||||
outgoing_queue: Vec::new(),
|
||||
processing_client,
|
||||
_marker: std::marker::PhantomData,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts, Pc, SndPkt, RcvPkt> BaseClient<Ts, Pc, SndPkt, RcvPkt>
|
||||
where
|
||||
SndPkt: WirePacketFormat,
|
||||
RcvPkt: WirePacketFormat,
|
||||
{
|
||||
/// Send `packet` to the mix node identified by `node_id` via `mix_socket`.
|
||||
///
|
||||
/// Resolves `node_id` against the shared [`Directory`], serialises via
|
||||
/// [`WirePacketFormat::to_bytes`], and dispatches with a single `sendto`.
|
||||
/// Errors are logged but not propagated.
|
||||
pub fn send_to_node(&self, node_id: NodeId, packet: SndPkt) {
|
||||
if let Some(node) = self.directory.node(node_id) {
|
||||
if let Err(e) = self.mix_socket.send_to(&packet.to_bytes(), node.addr) {
|
||||
tracing::error!("[Client {}] Failed to send to node {node_id}: {e}", self.id);
|
||||
} else {
|
||||
tracing::debug!(
|
||||
"[Client {}] Sent packet to node {node_id} @ {}",
|
||||
self.id,
|
||||
node.addr
|
||||
);
|
||||
}
|
||||
} else {
|
||||
tracing::error!("[Client {}] Node {node_id} not found in directory", self.id);
|
||||
}
|
||||
}
|
||||
|
||||
/// Attempt to receive one packet from the mix socket and deserialise it.
|
||||
///
|
||||
/// Returns `None` when the socket would block (no datagram waiting).
|
||||
pub fn recv_from_mix(&self) -> Option<anyhow::Result<RcvPkt>> {
|
||||
let mut buf = [0u8; 1500];
|
||||
let (nb, src) = match self.mix_socket.recv_from(&mut buf) {
|
||||
Ok(r) => r,
|
||||
Err(e) if e.kind() == ErrorKind::WouldBlock => return None,
|
||||
Err(e) => {
|
||||
tracing::error!("[Client {}] mix_socket recv error: {e}", self.id);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
tracing::debug!(
|
||||
"[Client {}] Received {nb} byte(s) from mix node {src}",
|
||||
self.id
|
||||
);
|
||||
Some(RcvPkt::try_from_bytes(&buf[..nb]))
|
||||
}
|
||||
|
||||
/// Attempt to receive one raw datagram from the app socket.
|
||||
///
|
||||
/// Returns `None` when the socket would block (no datagram waiting).
|
||||
pub fn recv_from_app(&self) -> Option<anyhow::Result<Vec<u8>>> {
|
||||
let mut buf = [0u8; 15000];
|
||||
let nb = match self.app_socket.recv(&mut buf) {
|
||||
Ok(n) => n,
|
||||
Err(e) if e.kind() == ErrorKind::WouldBlock => return None,
|
||||
Err(e) => {
|
||||
tracing::error!("[Client {}] app_socket recv error: {e}", self.id);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
Some(Ok(buf[..nb].to_vec()))
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts, Pc, SndPkt, RcvPkt> MixSimClient<Ts> for BaseClient<Ts, Pc, SndPkt, RcvPkt>
|
||||
where
|
||||
Ts: Clone + PartialOrd + Debug + Send,
|
||||
SndPkt: WirePacketFormat + Debug + Send,
|
||||
RcvPkt: WirePacketFormat + Debug + Send,
|
||||
Pc: ProcessingClient<Ts, SndPkt, RcvPkt>,
|
||||
{
|
||||
fn tick(&mut self, timestamp: Ts) {
|
||||
self.tick_app_incoming(timestamp.clone());
|
||||
self.tick_outgoing(timestamp.clone());
|
||||
self.tick_mix_incoming(timestamp);
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts, Pc, SndPkt, RcvPkt> BaseClient<Ts, Pc, SndPkt, RcvPkt>
|
||||
where
|
||||
Ts: Clone + PartialOrd + Debug + Send,
|
||||
SndPkt: WirePacketFormat + Debug + Send,
|
||||
RcvPkt: WirePacketFormat + Debug + Send,
|
||||
Pc: ProcessingClient<Ts, SndPkt, RcvPkt>,
|
||||
{
|
||||
/// **Phase 1 — app incoming**: drain the app socket, run each payload
|
||||
/// through the processing pipeline, and enqueue the resulting packets.
|
||||
fn tick_app_incoming(&mut self, timestamp: Ts) {
|
||||
// Collect (dst, payload) pairs from the app socket.
|
||||
let mut inputs = Vec::new();
|
||||
while let Some(result) = self.recv_from_app() {
|
||||
let bytes = match result {
|
||||
Ok(b) => b,
|
||||
Err(e) => {
|
||||
tracing::error!("[Client {}] app_socket recv error: {e}", self.id);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
// We assume format is [dst, payload]
|
||||
if bytes.len() < 2 {
|
||||
tracing::warn!(
|
||||
"[Client {}] app message too short ({} bytes), dropping",
|
||||
self.id,
|
||||
bytes.len()
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
let dst = bytes[0];
|
||||
let payload = bytes[1..].to_vec();
|
||||
tracing::debug!(
|
||||
"[Client {}] App input: {} byte(s) → client {dst}",
|
||||
self.id,
|
||||
payload.len()
|
||||
);
|
||||
inputs.push((dst, payload));
|
||||
}
|
||||
|
||||
// Always call process at least once; use an empty payload to self when idle.
|
||||
// We need to tick cover traffic
|
||||
if inputs.is_empty() {
|
||||
inputs.push((self.id, vec![]));
|
||||
}
|
||||
|
||||
for (dst, payload) in inputs {
|
||||
let packets = self
|
||||
.processing_client
|
||||
.process(payload, dst, timestamp.clone());
|
||||
self.outgoing_queue.extend(packets);
|
||||
}
|
||||
}
|
||||
|
||||
/// **Phase 2 — outgoing**: send all queued packets whose scheduled
|
||||
/// timestamp is ≤ `timestamp` to their first-hop node.
|
||||
fn tick_outgoing(&mut self, timestamp: Ts) {
|
||||
let to_send = self
|
||||
.outgoing_queue
|
||||
.extract_if(.., |pkt| pkt.data.timestamp <= timestamp)
|
||||
.collect::<Vec<_>>();
|
||||
for pkt in to_send {
|
||||
self.send_to_node(pkt.dst, pkt.data.data);
|
||||
}
|
||||
}
|
||||
|
||||
/// **Phase 3 — mix incoming**: drain the mix socket and pass each packet
|
||||
/// through the unwrapping pipeline.
|
||||
fn tick_mix_incoming(&mut self, timestamp: Ts) {
|
||||
while let Some(result) = self.recv_from_mix() {
|
||||
match result {
|
||||
Ok(pkt) => match self.processing_client.unwrap(pkt, timestamp.clone()) {
|
||||
Ok(Some(content)) => {
|
||||
tracing::info!(
|
||||
"[Client {}] Received: {:?}",
|
||||
self.id,
|
||||
String::from_utf8_lossy(&content)
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!("[Client {}] Error unwrapping packet : {e}", self.id);
|
||||
}
|
||||
Ok(None) => {}
|
||||
},
|
||||
Err(e) => {
|
||||
tracing::error!("[Client {}] Failed to deserialize mix packet: {e}", self.id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,254 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Simulated mix-network client.
|
||||
//!
|
||||
//! A [`SimpleClient`] owns a [`BaseClient`] (which manages both UDP sockets
|
||||
//! and the routing directory) plus the mix and unwrapping pipelines.
|
||||
//!
|
||||
//! ## Tick phases
|
||||
//!
|
||||
//! ```text
|
||||
//! tick_app_incoming ──── app_socket ──▶ processing_pipeline ──▶ outgoing_queue
|
||||
//! tick_outgoing ──── outgoing_queue ──▶ mix_socket ──▶ Node N
|
||||
//! tick_mix_incoming ──── mix_socket ◀── Node N ──▶ unwrapping_pipeline
|
||||
//! ```
|
||||
//!
|
||||
//! ## App-socket message format
|
||||
//!
|
||||
//! ```text
|
||||
//! ┌────────────────────────┬─────────────────────┐
|
||||
//! │ dst_client_id (1 B) │ raw payload bytes │
|
||||
//! └────────────────────────┴─────────────────────┘
|
||||
//! ```
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use nym_lp_data::{
|
||||
AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload,
|
||||
clients::{
|
||||
InputOptions, PipelinePayload,
|
||||
helpers::{NoOpObfuscation, NoOpReliability, NoOpRoutingSecurity},
|
||||
traits::{Chunking, ClientUnwrappingPipeline, ClientWrappingPipeline},
|
||||
},
|
||||
common::traits::{
|
||||
Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline,
|
||||
WireWrappingPipeline,
|
||||
},
|
||||
};
|
||||
|
||||
use crate::{
|
||||
client::{BaseClient, ClientId, ProcessingClient},
|
||||
node::NodeId,
|
||||
packet::simple::{
|
||||
SimpleFrame, SimpleMessage, SimplePacket, SimpleWireUnwrapper, SimpleWireWrapper,
|
||||
},
|
||||
topology::{TopologyClient, directory::Directory},
|
||||
};
|
||||
|
||||
/// A simulated client that injects packets into the mix network.
|
||||
///
|
||||
/// `Ts` is the timestamp / tick-context type. Packet type, frame type, and
|
||||
/// message marker are fixed to the `Simple*` concrete types.
|
||||
///
|
||||
/// UDP transport and routing are handled by the embedded [`BaseClient`]; this
|
||||
/// struct adds the outgoing queue and the wrapping/unwrapping pipelines.
|
||||
pub type SimpleClient<Ts> = BaseClient<Ts, SimpleProcessingClient, SimplePacket>;
|
||||
|
||||
impl<Ts> SimpleClient<Ts> {
|
||||
/// Bind both UDP sockets and return a new client.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if either socket fails to bind or set non-blocking.
|
||||
pub fn new(topology_client: TopologyClient, directory: Arc<Directory>) -> anyhow::Result<Self> {
|
||||
let processing_client = SimpleProcessingClient {
|
||||
wrapper: SimpleClientWrappingPipeline::default(),
|
||||
unwrapper: SimpleClientUnwrapping::default(),
|
||||
};
|
||||
BaseClient::with_pipeline(&topology_client, directory, processing_client)
|
||||
}
|
||||
}
|
||||
|
||||
/// [`InputOptions`] for the simple pipeline — all optional features disabled,
|
||||
/// next hop is always node 0, really simple routing
|
||||
#[derive(Clone, Copy)]
|
||||
pub struct SimpleInputOptions;
|
||||
|
||||
impl InputOptions<NodeId> for SimpleInputOptions {
|
||||
fn reliability(&self) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
fn routing_security(&self) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
fn obfuscation(&self) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
fn next_hop(&self) -> NodeId {
|
||||
0
|
||||
}
|
||||
}
|
||||
|
||||
/// Bridges [`BaseClient`] to the simple wrapping and unwrapping pipelines.
|
||||
pub struct SimpleProcessingClient {
|
||||
wrapper: SimpleClientWrappingPipeline,
|
||||
unwrapper: SimpleClientUnwrapping,
|
||||
}
|
||||
|
||||
impl<Ts: Clone> ProcessingClient<Ts, SimplePacket> for SimpleProcessingClient {
|
||||
fn process(
|
||||
&mut self,
|
||||
input: Vec<u8>,
|
||||
_: ClientId,
|
||||
timestamp: Ts,
|
||||
) -> Vec<AddressedTimedData<Ts, SimplePacket, NodeId>> {
|
||||
self.wrapper
|
||||
.process(Some((input, SimpleInputOptions)), timestamp)
|
||||
}
|
||||
|
||||
fn unwrap(&mut self, input: SimplePacket, timestamp: Ts) -> anyhow::Result<Option<Vec<u8>>> {
|
||||
self.unwrapper.unwrap(input, timestamp)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Concrete pipelines
|
||||
|
||||
/// Stub client processing pipeline for [`SimplePacket`].
|
||||
///
|
||||
/// A no-op pass-through: returns the payload as a single packet with no
|
||||
/// Sphinx layering, chunking, reliability encoding, or obfuscation.
|
||||
///
|
||||
/// All required sub-traits of [`ClientWrappingPipeline`] are implemented here;
|
||||
/// [`ClientWrappingPipeline`] is then provided automatically via the blanket
|
||||
/// impl in `nym_lp_data`.
|
||||
pub struct SimpleClientWrappingPipeline(SimpleWireWrapper);
|
||||
|
||||
pub(crate) type SimplePipelinePayload<Ts> = PipelinePayload<Ts, SimpleInputOptions, NodeId>;
|
||||
|
||||
impl Default for SimpleClientWrappingPipeline {
|
||||
fn default() -> Self {
|
||||
Self(SimpleWireWrapper)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> Chunking<Ts, SimpleInputOptions, NodeId> for SimpleClientWrappingPipeline {
|
||||
/// Split `input` into chunks of `chunk_size` bytes, padding the last chunk
|
||||
/// with zero bytes if necessary.
|
||||
///
|
||||
/// A `0x01` marker byte is appended before padding so the unwrapper can
|
||||
/// strip trailing zeros.
|
||||
fn chunked(
|
||||
&mut self,
|
||||
mut input: Vec<u8>,
|
||||
options: SimpleInputOptions,
|
||||
chunk_size: usize,
|
||||
timestamp: Ts,
|
||||
) -> Vec<SimplePipelinePayload<Ts>> {
|
||||
input.push(1);
|
||||
if !input.len().is_multiple_of(chunk_size) {
|
||||
let padding = vec![0; chunk_size - input.len() % chunk_size];
|
||||
input.extend_from_slice(&padding);
|
||||
}
|
||||
|
||||
input
|
||||
.chunks(chunk_size)
|
||||
.map(|chunk| SimplePipelinePayload::new(timestamp.clone(), chunk.to_vec(), options))
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
impl NoOpReliability for SimpleClientWrappingPipeline {}
|
||||
impl NoOpObfuscation for SimpleClientWrappingPipeline {}
|
||||
impl NoOpRoutingSecurity for SimpleClientWrappingPipeline {}
|
||||
|
||||
// Delegation to SimpleWireWrapper
|
||||
impl<Ts: Clone> Framing<Ts, NodeId> for SimpleClientWrappingPipeline {
|
||||
type Frame = SimpleFrame;
|
||||
const OVERHEAD_SIZE: usize = <SimpleWireWrapper as Framing<Ts, _>>::OVERHEAD_SIZE;
|
||||
fn to_frame(
|
||||
&self,
|
||||
payload: AddressedTimedPayload<Ts, NodeId>,
|
||||
frame_size: usize,
|
||||
) -> Vec<AddressedTimedData<Ts, SimpleFrame, NodeId>> {
|
||||
self.0.to_frame(payload, frame_size)
|
||||
}
|
||||
}
|
||||
|
||||
// Delegation to SimpleWireWrapper
|
||||
impl<Ts: Clone> Transport<Ts, SimplePacket, NodeId> for SimpleClientWrappingPipeline {
|
||||
type Frame = SimpleFrame;
|
||||
const OVERHEAD_SIZE: usize = <SimpleWireWrapper as Transport<Ts, _, _>>::OVERHEAD_SIZE;
|
||||
fn to_transport_packet(
|
||||
&self,
|
||||
frame: AddressedTimedData<Ts, SimpleFrame, NodeId>,
|
||||
) -> AddressedTimedData<Ts, SimplePacket, NodeId> {
|
||||
self.0.to_transport_packet(frame)
|
||||
}
|
||||
}
|
||||
|
||||
// Delegation to SimpleWireWrapper
|
||||
impl<Ts: Clone> WireWrappingPipeline<Ts, SimplePacket, NodeId> for SimpleClientWrappingPipeline {
|
||||
fn packet_size(&self) -> usize {
|
||||
<SimpleWireWrapper as WireWrappingPipeline<Ts, _, _>>::packet_size(&self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> ClientWrappingPipeline<Ts, SimplePacket, SimpleInputOptions, NodeId>
|
||||
for SimpleClientWrappingPipeline
|
||||
{
|
||||
}
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Unwrapping pipeline for [`SimpleClient`]: strips the frame header and
|
||||
/// removes padding from the recovered payload.
|
||||
pub struct SimpleClientUnwrapping(SimpleWireUnwrapper);
|
||||
|
||||
impl Default for SimpleClientUnwrapping {
|
||||
fn default() -> Self {
|
||||
Self(SimpleWireUnwrapper)
|
||||
}
|
||||
}
|
||||
|
||||
// Delegation to SimpleWireUnwrapper
|
||||
impl<Ts> FramingUnwrap<Ts, SimpleMessage> for SimpleClientUnwrapping {
|
||||
type Frame = SimpleFrame;
|
||||
fn frame_to_message(
|
||||
&mut self,
|
||||
frame: TimedData<Ts, SimpleFrame>,
|
||||
) -> Option<(TimedPayload<Ts>, SimpleMessage)> {
|
||||
self.0.frame_to_message(frame)
|
||||
}
|
||||
}
|
||||
|
||||
// Delegation to SimpleWireUnwrapper
|
||||
impl<Ts: Clone> TransportUnwrap<Ts, SimplePacket> for SimpleClientUnwrapping {
|
||||
type Frame = SimpleFrame;
|
||||
fn packet_to_frame(
|
||||
&self,
|
||||
packet: SimplePacket,
|
||||
timestamp: Ts,
|
||||
) -> anyhow::Result<TimedData<Ts, SimpleFrame>> {
|
||||
self.0.packet_to_frame(packet, timestamp)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> WireUnwrappingPipeline<Ts, SimplePacket, SimpleMessage> for SimpleClientUnwrapping {}
|
||||
|
||||
impl<Ts: Clone> ClientUnwrappingPipeline<Ts, SimplePacket, SimpleMessage> for SimpleClientUnwrapping {
|
||||
fn process_unwrapped(
|
||||
&mut self,
|
||||
payload: TimedPayload<Ts>,
|
||||
_kind: SimpleMessage,
|
||||
) -> Option<Vec<u8>> {
|
||||
let mut data = payload.data;
|
||||
if let Some(pos) = data.iter().rposition(|&b| b == 1) {
|
||||
data.truncate(pos);
|
||||
}
|
||||
Some(data)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,344 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! [`SphinxClient`] — simulated client using full Sphinx encryption.
|
||||
//!
|
||||
//! The wrapping pipeline applies chunking, Sphinx encryption (routing security),
|
||||
//! and Poisson cover traffic obfuscation. The unwrapping pipeline reconstructs
|
||||
//! fragmented messages and filters out cover traffic.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use nym_lp_data::{
|
||||
AddressedTimedData, TimedPayload,
|
||||
clients::{
|
||||
InputOptions, PipelinePayload,
|
||||
traits::{
|
||||
Chunking, ClientUnwrappingPipeline, ClientWrappingPipeline, Obfuscation, Reliability,
|
||||
RoutingSecurity,
|
||||
},
|
||||
},
|
||||
common::{
|
||||
helpers::{NoOpWireUnwrapper, NoOpWireWrapper},
|
||||
traits::{Framing, Transport, WireWrappingPipeline},
|
||||
},
|
||||
};
|
||||
use nym_sphinx::{
|
||||
Delay, Destination, DestinationAddressBytes, SphinxPacketBuilder,
|
||||
chunking::{fragment::Fragment, reconstruction::MessageReconstructor},
|
||||
message::{NymMessage, PaddedMessage},
|
||||
};
|
||||
use rand::Rng;
|
||||
|
||||
use crate::{
|
||||
client::{
|
||||
BaseClient, ClientId, ProcessingClient,
|
||||
sphinx::{poisson_cover_traffic::PoissonCoverTraffic, surb_acks::SurbAcksReliability},
|
||||
},
|
||||
node::NodeId,
|
||||
packet::sphinx::{GenerateDelay, SimMixPacket, SphinxMessage, SurbAck},
|
||||
topology::{TopologyClient, directory::Directory},
|
||||
};
|
||||
|
||||
mod poisson_cover_traffic;
|
||||
mod surb_acks;
|
||||
|
||||
/// A simulated client that injects packets into the mix network.
|
||||
///
|
||||
/// `Ts` is the timestamp / tick-context type. Packet type, frame type, and
|
||||
/// message marker are fixed to the `Sphinx*` concrete types.
|
||||
///
|
||||
/// UDP transport and routing are handled by the embedded [`BaseClient`]; this
|
||||
/// struct adds the outgoing queue and the wrapping/unwrapping pipelines.
|
||||
pub type SphinxClient<Ts, R> = BaseClient<Ts, SphinxProcessingClient<Ts, R>, SimMixPacket, Vec<u8>>;
|
||||
|
||||
impl<Ts: Clone + GenerateDelay + PartialOrd + Send, R: Rng + Clone + Send> SphinxClient<Ts, R> {
|
||||
/// Bind both UDP sockets and return a new client.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if either socket fails to bind or set non-blocking.
|
||||
pub fn new(
|
||||
topology_client: TopologyClient,
|
||||
directory: Arc<Directory>,
|
||||
current_timestamp: Ts,
|
||||
rng: R,
|
||||
) -> anyhow::Result<Self> {
|
||||
let processing_client = SphinxProcessingClient {
|
||||
wrapper: SphinxClientWrappingPipeline {
|
||||
cover_traffic: PoissonCoverTraffic::new(
|
||||
topology_client.client_id,
|
||||
directory.clone(),
|
||||
current_timestamp,
|
||||
rng.clone(),
|
||||
),
|
||||
reliability: SurbAcksReliability::new(
|
||||
rng.clone(),
|
||||
topology_client.client_id,
|
||||
directory.clone(),
|
||||
),
|
||||
directory: directory.clone(),
|
||||
rng,
|
||||
},
|
||||
unwrapper: SphinxClientUnwrapping::default(),
|
||||
};
|
||||
BaseClient::with_pipeline(&topology_client, directory, processing_client)
|
||||
}
|
||||
}
|
||||
|
||||
/// [`InputOptions`] for the Sphinx pipeline — reliability, routing security,
|
||||
/// and obfuscation are all enabled.
|
||||
#[derive(Clone, Copy)]
|
||||
pub struct SphinxInputOptions {
|
||||
/// Destination client ID, embedded in the Sphinx destination address.
|
||||
dst: ClientId,
|
||||
/// First-hop node ID. In a real Nym network this would be the client's
|
||||
/// gateway; here it is chosen at random from the topology because there is
|
||||
/// no gateway concept in the simulation.
|
||||
next_hop: NodeId,
|
||||
}
|
||||
|
||||
impl InputOptions<NodeId> for SphinxInputOptions {
|
||||
fn reliability(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn routing_security(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn obfuscation(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn next_hop(&self) -> NodeId {
|
||||
self.next_hop
|
||||
}
|
||||
}
|
||||
|
||||
/// Bridges [`BaseClient`] to the Sphinx wrapping and unwrapping pipelines.
|
||||
pub struct SphinxProcessingClient<Ts: Clone + GenerateDelay + PartialOrd, R: Rng> {
|
||||
wrapper: SphinxClientWrappingPipeline<Ts, R>,
|
||||
unwrapper: SphinxClientUnwrapping,
|
||||
}
|
||||
|
||||
impl<Ts: Clone + GenerateDelay + PartialOrd + Send, R: Rng + Send>
|
||||
ProcessingClient<Ts, SimMixPacket, Vec<u8>> for SphinxProcessingClient<Ts, R>
|
||||
{
|
||||
fn process(
|
||||
&mut self,
|
||||
input: Vec<u8>,
|
||||
dst: ClientId,
|
||||
timestamp: Ts,
|
||||
) -> Vec<AddressedTimedData<Ts, SimMixPacket, NodeId>> {
|
||||
let input_options = SphinxInputOptions {
|
||||
dst,
|
||||
next_hop: self
|
||||
.wrapper
|
||||
.directory
|
||||
.random_next_hop(&mut self.wrapper.rng), // This substitutes for a real gateway selection — in the simulation every node is equally eligible as a first hop
|
||||
};
|
||||
self.wrapper
|
||||
.process(Some((input, input_options)), timestamp)
|
||||
}
|
||||
|
||||
fn unwrap(&mut self, input: Vec<u8>, timestamp: Ts) -> anyhow::Result<Option<Vec<u8>>> {
|
||||
self.unwrapper.unwrap(input, timestamp)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Concrete pipelines
|
||||
|
||||
/// Full wrapping pipeline for [`SphinxClient`].
|
||||
///
|
||||
/// Applies, in order: chunking (using standard Sphinx fragmentation), SURB-ACK
|
||||
/// reliability prefix, Poisson cover traffic obfuscation, Sphinx onion
|
||||
/// encryption, and a no-op wire wrapper (a Sphinx packet is already its own
|
||||
/// wire unit).
|
||||
pub struct SphinxClientWrappingPipeline<Ts: Clone + GenerateDelay + PartialOrd, R: Rng> {
|
||||
cover_traffic: PoissonCoverTraffic<Ts, R>,
|
||||
reliability: SurbAcksReliability<R>,
|
||||
directory: Arc<Directory>,
|
||||
rng: R,
|
||||
}
|
||||
|
||||
pub(crate) type SphinxPipelinePayload<Ts> = PipelinePayload<Ts, SphinxInputOptions, NodeId>;
|
||||
|
||||
impl<Ts: Clone + GenerateDelay + PartialOrd, R: Rng> Chunking<Ts, SphinxInputOptions, NodeId>
|
||||
for SphinxClientWrappingPipeline<Ts, R>
|
||||
{
|
||||
fn chunked(
|
||||
&mut self,
|
||||
input: Vec<u8>,
|
||||
options: SphinxInputOptions,
|
||||
chunk_size: usize,
|
||||
timestamp: Ts,
|
||||
) -> Vec<SphinxPipelinePayload<Ts>> {
|
||||
if input.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
|
||||
// This is using standard sphinx chunking. Proper LP should use a different one
|
||||
let fragments = NymMessage::new_plain(input)
|
||||
.pad_to_full_packet_lengths(chunk_size)
|
||||
.split_into_fragments(&mut self.rng, chunk_size);
|
||||
|
||||
fragments
|
||||
.into_iter()
|
||||
.map(|fragment| {
|
||||
SphinxPipelinePayload::new(timestamp.clone(), fragment.into_bytes(), options)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone + GenerateDelay + PartialOrd, R: Rng> Reliability<Ts, SphinxInputOptions, NodeId>
|
||||
for SphinxClientWrappingPipeline<Ts, R>
|
||||
{
|
||||
const OVERHEAD_SIZE: usize =
|
||||
<SurbAcksReliability<R> as Reliability<Ts, SphinxInputOptions, _>>::OVERHEAD_SIZE;
|
||||
fn reliable_encode(
|
||||
&mut self,
|
||||
input: Option<SphinxPipelinePayload<Ts>>,
|
||||
timestamp: Ts,
|
||||
) -> Vec<SphinxPipelinePayload<Ts>> {
|
||||
self.reliability.reliable_encode(input, timestamp)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone + GenerateDelay + PartialOrd, R: Rng> Obfuscation<Ts, SphinxInputOptions, NodeId>
|
||||
for SphinxClientWrappingPipeline<Ts, R>
|
||||
{
|
||||
fn obfuscate(
|
||||
&mut self,
|
||||
input: Option<SphinxPipelinePayload<Ts>>,
|
||||
timestamp: Ts,
|
||||
) -> Vec<SphinxPipelinePayload<Ts>> {
|
||||
self.cover_traffic.obfuscate(input, timestamp)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone + GenerateDelay + PartialOrd, R: Rng> RoutingSecurity<Ts, SphinxInputOptions, NodeId>
|
||||
for SphinxClientWrappingPipeline<Ts, R>
|
||||
{
|
||||
const OVERHEAD_SIZE: usize = nym_sphinx::HEADER_SIZE + nym_sphinx::PAYLOAD_OVERHEAD_SIZE;
|
||||
fn nb_frames(&self) -> usize {
|
||||
1
|
||||
}
|
||||
/// Wrap `input` in a Sphinx onion packet with a 3-hop route.
|
||||
///
|
||||
/// The route is built by taking `input_options.next_hop` as the first hop
|
||||
/// and choosing two additional hops at random from the directory (repeats are
|
||||
/// allowed). The final destination is the client identified by
|
||||
/// `input_options.dst`. Per-hop delays are drawn from
|
||||
/// [`GenerateDelay::generate_mix_delay`].
|
||||
fn encrypt(&mut self, input: SphinxPipelinePayload<Ts>) -> SphinxPipelinePayload<Ts> {
|
||||
// SAFETY: IDs were sampled from the directory, so they are guaranteed to exist.
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let first_hop = self.directory.node(input.options.next_hop).unwrap().into();
|
||||
|
||||
let route = std::iter::once(first_hop)
|
||||
.chain(
|
||||
self.directory
|
||||
.random_route(2, &mut self.rng)
|
||||
.iter()
|
||||
.map(Into::into),
|
||||
)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let destination = Destination::new(
|
||||
DestinationAddressBytes::from_bytes([input.options.dst; 32]),
|
||||
[input.options.dst; 16],
|
||||
);
|
||||
|
||||
let delays = (0..route.len())
|
||||
.map(|_| Delay::new_from_millis(Ts::generate_mix_delay(&mut self.rng)))
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
// Useful payload size is packet size - transport overhead - framing overhead - routing overhead
|
||||
let plaintext_size =
|
||||
<Self as WireWrappingPipeline<Ts, SimMixPacket, NodeId>>::packet_size(self)
|
||||
- <Self as Framing<Ts, NodeId>>::OVERHEAD_SIZE
|
||||
- <Self as Transport<Ts, SimMixPacket, NodeId>>::OVERHEAD_SIZE
|
||||
- <Self as RoutingSecurity<Ts, _, _>>::OVERHEAD_SIZE;
|
||||
|
||||
// Packet builder's size includes the payload overhead so we have to add it
|
||||
let packet_builder = SphinxPacketBuilder::new()
|
||||
.with_payload_size(plaintext_size + nym_sphinx::PAYLOAD_OVERHEAD_SIZE);
|
||||
|
||||
// SAFETY : If the pipeline is built correctly, the packet building should not fail.
|
||||
// If it does, something is wrong with the code. If it crashes it's fine since it's a simulator anyway
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let packet = packet_builder
|
||||
.build_packet(input.data.data, &route, &destination, &delays)
|
||||
.unwrap();
|
||||
|
||||
SphinxPipelinePayload::new(input.data.timestamp, packet.to_bytes(), input.options)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone + GenerateDelay + PartialOrd, R: Rng> NoOpWireWrapper
|
||||
for SphinxClientWrappingPipeline<Ts, R>
|
||||
{
|
||||
}
|
||||
|
||||
impl<Ts: Clone + GenerateDelay + PartialOrd, R: Rng>
|
||||
ClientWrappingPipeline<Ts, SimMixPacket, SphinxInputOptions, NodeId>
|
||||
for SphinxClientWrappingPipeline<Ts, R>
|
||||
{
|
||||
}
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Unwrapping pipeline for [`SphinxClient`].
|
||||
///
|
||||
/// Receives the raw final-hop payload (the last Sphinx layer has already been
|
||||
/// stripped by the terminal mix node), recovers the plaintext, filters cover
|
||||
/// traffic, and reassembles Sphinx fragments into complete messages.
|
||||
#[derive(Default)]
|
||||
pub struct SphinxClientUnwrapping {
|
||||
message_reconstructor: MessageReconstructor,
|
||||
}
|
||||
|
||||
impl NoOpWireUnwrapper for SphinxClientUnwrapping {}
|
||||
|
||||
impl<Ts: Clone> ClientUnwrappingPipeline<Ts, Vec<u8>, SphinxMessage> for SphinxClientUnwrapping {
|
||||
fn process_unwrapped(
|
||||
&mut self,
|
||||
timed_plaintext: TimedPayload<Ts>,
|
||||
_kind: SphinxMessage,
|
||||
) -> Option<Vec<u8>> {
|
||||
let plaintext = timed_plaintext.data;
|
||||
|
||||
// Ditch cover traffic
|
||||
if nym_sphinx::cover::is_cover(&plaintext) {
|
||||
tracing::debug!("Received cover traffic packet");
|
||||
return None;
|
||||
}
|
||||
|
||||
// TODO Route acks elsewhere HERE
|
||||
if SurbAck::is_surb_ack(&plaintext) {
|
||||
// SAFETY : casting slice of len 8 into array of len 8
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let id = u64::from_le_bytes(plaintext[8..16].try_into().unwrap());
|
||||
tracing::debug!("Received a SURB_ACK for id : {id}");
|
||||
return None;
|
||||
}
|
||||
|
||||
let fragment = Fragment::try_from_bytes(&plaintext)
|
||||
.inspect_err(|e| tracing::warn!("Failed to deserialize fragment : {e}"))
|
||||
.ok()?;
|
||||
|
||||
if let Some(reconstructed_message) =
|
||||
self.message_reconstructor.insert_new_fragment(fragment)
|
||||
{
|
||||
let message = PaddedMessage::from(reconstructed_message.0)
|
||||
.remove_padding()
|
||||
.inspect_err(|e| tracing::warn!("Failed to remove padding : {e}"))
|
||||
.ok()?;
|
||||
Some(message.into_inner_data())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,148 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Poisson cover traffic generator.
|
||||
//!
|
||||
//! Implements the [`Obfuscation`] trait for [`SphinxClient`] using two
|
||||
//! independent Poisson processes:
|
||||
//!
|
||||
//! * **Main loop** — schedules one slot per inter-arrival time drawn from an
|
||||
//! exponential distribution. Real messages are injected into these slots; if
|
||||
//! no real message is ready when a slot fires, a cover-traffic payload is sent
|
||||
//! instead.
|
||||
//! * **Secondary loop** — independently fires cover-traffic packets at a lower
|
||||
//! rate, providing additional traffic volume that is independent of the main
|
||||
//! loop's cadence.
|
||||
//!
|
||||
//! Together the two loops ensure that an observer cannot determine from traffic
|
||||
//! patterns alone whether the client is actively sending real messages.
|
||||
//!
|
||||
//! [`SphinxClient`]: super::SphinxClient
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use nym_lp_data::clients::traits::Obfuscation;
|
||||
use nym_sphinx::cover::LOOP_COVER_MESSAGE_PAYLOAD;
|
||||
use rand::Rng;
|
||||
|
||||
use crate::{
|
||||
client::{
|
||||
ClientId,
|
||||
sphinx::{SphinxInputOptions, SphinxPipelinePayload},
|
||||
},
|
||||
node::NodeId,
|
||||
packet::sphinx::GenerateDelay,
|
||||
topology::directory::Directory,
|
||||
};
|
||||
|
||||
/// Two-loop Poisson cover traffic generator.
|
||||
///
|
||||
/// Maintains two independent next-fire timestamps — one for the main sending
|
||||
/// loop and one for the secondary cover loop — and advances them by independent
|
||||
/// exponential delays on each firing.
|
||||
pub struct PoissonCoverTraffic<Ts, R>
|
||||
where
|
||||
Ts: Clone + GenerateDelay + PartialOrd,
|
||||
R: Rng,
|
||||
{
|
||||
address: ClientId,
|
||||
directory: Arc<Directory>,
|
||||
/// Timestamp at which the main loop next fires (real or cover packet).
|
||||
main_loop_next_timestamp: Ts,
|
||||
/// Timestamp at which the secondary cover loop next fires.
|
||||
secondary_loop_next_timestamp: Ts,
|
||||
/// Random number generator used for exponential delay sampling.
|
||||
rng: R,
|
||||
}
|
||||
|
||||
impl<Ts, R> PoissonCoverTraffic<Ts, R>
|
||||
where
|
||||
Ts: Clone + GenerateDelay + PartialOrd,
|
||||
R: Rng,
|
||||
{
|
||||
/// Construct a new cover traffic generator.
|
||||
///
|
||||
/// Both loops are initialised to fire immediately at `current_timestamp` so
|
||||
/// that cover traffic begins on the very first tick.
|
||||
pub fn new(
|
||||
address: ClientId,
|
||||
directory: Arc<Directory>,
|
||||
current_timestamp: Ts,
|
||||
rng: R,
|
||||
) -> Self {
|
||||
Self {
|
||||
address,
|
||||
directory,
|
||||
main_loop_next_timestamp: current_timestamp.clone(),
|
||||
secondary_loop_next_timestamp: current_timestamp,
|
||||
rng,
|
||||
}
|
||||
}
|
||||
|
||||
/// Build [`SphinxInputOptions`] for a self-addressed cover-traffic packet.
|
||||
///
|
||||
/// The destination is set to this client's own address and the first hop is
|
||||
/// chosen at random from the directory, matching the real-message behaviour.
|
||||
pub fn cover_traffic_options(&mut self) -> SphinxInputOptions {
|
||||
SphinxInputOptions {
|
||||
dst: self.address,
|
||||
next_hop: self.directory.random_next_hop(&mut self.rng),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts, R> Obfuscation<Ts, SphinxInputOptions, NodeId> for PoissonCoverTraffic<Ts, R>
|
||||
where
|
||||
Ts: Clone + GenerateDelay + PartialOrd,
|
||||
R: Rng,
|
||||
{
|
||||
/// Produce the set of payloads to send at `timestamp`.
|
||||
///
|
||||
/// Called once per tick with an optional real message (`input`). May
|
||||
/// return zero, one, or two payloads depending on which loops fire and
|
||||
/// whether a real message is available.
|
||||
fn obfuscate(
|
||||
&mut self,
|
||||
input: Option<SphinxPipelinePayload<Ts>>,
|
||||
timestamp: Ts,
|
||||
) -> Vec<SphinxPipelinePayload<Ts>> {
|
||||
let mut output = Vec::new();
|
||||
|
||||
// Secondary cover traffic loop
|
||||
// We should not schedule those in advance, because backpressure can't tell if it has real or cover traffic.
|
||||
if timestamp >= self.secondary_loop_next_timestamp {
|
||||
output.push(SphinxPipelinePayload::new(
|
||||
timestamp.clone(),
|
||||
LOOP_COVER_MESSAGE_PAYLOAD.to_vec(),
|
||||
self.cover_traffic_options(),
|
||||
));
|
||||
self.secondary_loop_next_timestamp = self.secondary_loop_next_timestamp.clone()
|
||||
+ Ts::generate_cover_traffic_delay(&mut self.rng);
|
||||
}
|
||||
|
||||
// Main cover traffic loop
|
||||
|
||||
match input {
|
||||
// If we have a message, schedule it for the next timestamp, prepare the following one
|
||||
Some(real_message) => {
|
||||
output.push(real_message.ts_transform(|_| self.main_loop_next_timestamp.clone()));
|
||||
self.main_loop_next_timestamp = self.main_loop_next_timestamp.clone()
|
||||
+ Ts::generate_sending_delay(&mut self.rng);
|
||||
}
|
||||
// No message, but we need to send something => Send cover traffic right away, prepare next timestamp
|
||||
None if timestamp >= self.main_loop_next_timestamp => {
|
||||
output.push(SphinxPipelinePayload::new(
|
||||
timestamp,
|
||||
LOOP_COVER_MESSAGE_PAYLOAD.to_vec(),
|
||||
self.cover_traffic_options(),
|
||||
));
|
||||
self.main_loop_next_timestamp = self.main_loop_next_timestamp.clone()
|
||||
+ Ts::generate_sending_delay(&mut self.rng);
|
||||
}
|
||||
// No message, not the time to send anything, nothing to do
|
||||
None => {}
|
||||
}
|
||||
|
||||
output
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! SURB-ACK reliability layer for [`SphinxClient`].
|
||||
//!
|
||||
//! Implements the [`Reliability`] trait by prepending a single-use reply block
|
||||
//! (SURB) acknowledgement to every outgoing payload. The SURB allows the
|
||||
//! recipient to send a compact acknowledgement back to the sender through the
|
||||
//! mix network without revealing the sender's address to intermediate nodes.
|
||||
//!
|
||||
//! [`SphinxClient`]: super::SphinxClient
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::{
|
||||
client::{
|
||||
ClientId,
|
||||
sphinx::{SphinxInputOptions, SphinxPipelinePayload},
|
||||
},
|
||||
node::NodeId,
|
||||
packet::sphinx::{GenerateDelay, SurbAck},
|
||||
topology::directory::Directory,
|
||||
};
|
||||
|
||||
use nym_lp_data::clients::traits::Reliability;
|
||||
|
||||
use rand::Rng;
|
||||
|
||||
/// Prepends a freshly-constructed SURB acknowledgement to every outgoing packet.
|
||||
///
|
||||
/// Each call to [`Reliability::reliable_encode`] builds a new [`SurbAck`] keyed
|
||||
/// to a random 64-bit packet identifier, prepends it to the payload, and returns
|
||||
/// the augmented packet. If `input` is `None` (cover-traffic slot) no packet
|
||||
/// is produced.
|
||||
pub struct SurbAcksReliability<R>
|
||||
where
|
||||
R: Rng,
|
||||
{
|
||||
address: ClientId,
|
||||
directory: Arc<Directory>,
|
||||
rng: R,
|
||||
}
|
||||
|
||||
impl<R> SurbAcksReliability<R>
|
||||
where
|
||||
R: Rng,
|
||||
{
|
||||
/// Create a new SURB-ACK reliability layer.
|
||||
///
|
||||
/// `address` is used as the SURB reply destination so that ACKs are routed
|
||||
/// back to this client. `directory` is used to sample the 3-hop SURB route.
|
||||
pub fn new(rng: R, address: ClientId, directory: Arc<Directory>) -> Self {
|
||||
Self {
|
||||
address,
|
||||
directory,
|
||||
rng,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts, R> Reliability<Ts, SphinxInputOptions, NodeId> for SurbAcksReliability<R>
|
||||
where
|
||||
R: Rng,
|
||||
Ts: GenerateDelay,
|
||||
{
|
||||
const OVERHEAD_SIZE: usize = SurbAck::len();
|
||||
|
||||
/// Prepend a SURB ACK to `input`, or return an empty vec for cover slots.
|
||||
///
|
||||
/// A fresh [`SurbAck`] is constructed for each real packet so that every
|
||||
/// in-flight packet carries a unique acknowledgement path.
|
||||
fn reliable_encode(
|
||||
&mut self,
|
||||
input: Option<SphinxPipelinePayload<Ts>>,
|
||||
_timestamp: Ts,
|
||||
) -> Vec<SphinxPipelinePayload<Ts>> {
|
||||
if let Some(packet) = input {
|
||||
let random_id = self.rng.next_u64();
|
||||
tracing::debug!("Generating SURB Ack with ID {random_id}");
|
||||
let surb_ack = SurbAck::construct::<Ts, R>(
|
||||
&mut self.rng,
|
||||
self.address,
|
||||
random_id,
|
||||
&self.directory,
|
||||
)
|
||||
.prepare_for_sending()
|
||||
.1;
|
||||
let reliable_packet = packet.data_transform(|payload| {
|
||||
surb_ack.iter().copied().chain(payload).collect::<Vec<_>>()
|
||||
});
|
||||
|
||||
vec![reliable_packet]
|
||||
} else {
|
||||
Vec::new()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,245 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Top-level simulation orchestrator.
|
||||
//!
|
||||
//! [`MixSimDriver`] owns the complete list of [`MixSimNode`]s and
|
||||
//! [`MixSimClient`]s and is the single entry point for running the simulation.
|
||||
//! It is responsible for:
|
||||
//!
|
||||
//! 1. **Bootstrapping** — building the shared [`Directory`] from pre-constructed
|
||||
//! nodes and clients, then distributing it to every participant.
|
||||
//! 2. **Ticking** — advancing every node and client through the phases of a
|
||||
//! simulation step (client tick → incoming → processing → outgoing).
|
||||
//! 3. **Driving** — either automatically (sleeping between ticks) or manually
|
||||
//! (waiting for the user to press ENTER).
|
||||
//!
|
||||
//! Nodes and clients are built externally (e.g. in [`SimpleMixDriver`]) and
|
||||
//! passed to [`MixSimDriver::new`] as boxed trait objects, so the driver only
|
||||
//! needs to know the timestamp type `Ts`.
|
||||
//!
|
||||
//! To inject packets into a running simulation, use the standalone `mix-client`
|
||||
//! binary, which sends payloads to a client's app socket.
|
||||
|
||||
use std::{
|
||||
fmt::Debug,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
use tracing::info;
|
||||
|
||||
use crate::{client::MixSimClient, node::MixSimNode};
|
||||
|
||||
mod simple;
|
||||
mod sphinx;
|
||||
|
||||
pub use simple::SimpleMixDriver;
|
||||
pub use sphinx::{DiscreteSphinxMixDriver, SphinxMixDriver};
|
||||
|
||||
/// Top-level orchestrator for the mix-network simulation.
|
||||
///
|
||||
/// Holds ordered lists of type-erased [`MixSimNode`]s and [`MixSimClient`]s.
|
||||
/// Only the timestamp type `Ts` is visible at this level; packet format, frame
|
||||
/// type, and message marker are encapsulated inside each concrete node/client.
|
||||
pub struct MixSimDriver<Ts>
|
||||
where
|
||||
Ts: Clone + PartialOrd + Debug + Send,
|
||||
{
|
||||
nodes: Vec<Box<dyn MixSimNode<Ts> + Send>>,
|
||||
clients: Vec<Box<dyn MixSimClient<Ts> + Send>>,
|
||||
}
|
||||
|
||||
impl<Ts> MixSimDriver<Ts>
|
||||
where
|
||||
Ts: Clone + PartialOrd + Debug + Send,
|
||||
{
|
||||
/// Construct the driver from pre-built nodes and clients.
|
||||
///
|
||||
/// Topology parsing and socket binding are the caller's responsibility.
|
||||
pub fn new(
|
||||
nodes: Vec<Box<dyn MixSimNode<Ts> + Send>>,
|
||||
clients: Vec<Box<dyn MixSimClient<Ts> + Send>>,
|
||||
) -> Self {
|
||||
Self { nodes, clients }
|
||||
}
|
||||
|
||||
/// Pretty-print the current state of every node at `tick`.
|
||||
pub fn display_state(&self, tick: Ts) {
|
||||
println!(
|
||||
"┌─── Tick {tick:─<3?}────────────────────────────────────────────────────────────┐"
|
||||
);
|
||||
for node in &self.nodes {
|
||||
node.display_state();
|
||||
println!("|------------------------------------------------------------------------|")
|
||||
}
|
||||
println!("└────────────────────────────────────────────────────────────────────────┘");
|
||||
}
|
||||
|
||||
/// Advance the simulation by one tick.
|
||||
///
|
||||
/// ## Phases
|
||||
///
|
||||
/// 1. **Client** - clients tick.
|
||||
/// 2. **Incoming** — every node drains its UDP socket into `packets_to_process`.
|
||||
/// 3. *(optional state display)*
|
||||
/// 4. **Processing** — every node mixes buffered packets.
|
||||
/// 5. *(optional state display)*
|
||||
/// 6. **Outgoing** — nodes forward due packets;
|
||||
pub fn tick(&mut self, timestamp: Ts, display_state: bool) {
|
||||
for client in &mut self.clients {
|
||||
client.tick(timestamp.clone());
|
||||
}
|
||||
// Phase 1 — incoming
|
||||
for node in &mut self.nodes {
|
||||
node.tick_incoming(timestamp.clone());
|
||||
}
|
||||
|
||||
if display_state {
|
||||
self.display_state(timestamp.clone());
|
||||
}
|
||||
|
||||
// Phase 2 — processing
|
||||
for node in &mut self.nodes {
|
||||
node.tick_processing(timestamp.clone());
|
||||
}
|
||||
|
||||
if display_state {
|
||||
self.display_state(timestamp.clone());
|
||||
}
|
||||
|
||||
// Phase 3 — outgoing
|
||||
for node in &mut self.nodes {
|
||||
node.tick_outgoing(timestamp.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Driving logic for the concrete `Ts = u32` timestamp flavour.
|
||||
///
|
||||
/// The timestamp is a monotonically increasing tick counter starting at zero.
|
||||
/// If a richer timestamp type is needed in the future, a new impl block should
|
||||
/// be added.
|
||||
impl MixSimDriver<u32> {
|
||||
/// Start the simulation in either manual or automatic mode.
|
||||
pub async fn run(
|
||||
self,
|
||||
manual_mode: bool,
|
||||
display_state: bool,
|
||||
start_tick: u32,
|
||||
tick_duration_ms: u64,
|
||||
) -> anyhow::Result<()> {
|
||||
if manual_mode {
|
||||
self.run_manual(start_tick, display_state)
|
||||
} else {
|
||||
self.run_automatic(start_tick, tick_duration_ms).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Run the simulation automatically, advancing one tick every
|
||||
/// `tick_duration_ms` milliseconds until Ctrl-C is received.
|
||||
pub async fn run_automatic(
|
||||
mut self,
|
||||
start_tick: u32,
|
||||
tick_duration_ms: u64,
|
||||
) -> anyhow::Result<()> {
|
||||
info!("Automatic mode: tick duration : {tick_duration_ms} ms");
|
||||
let tick_duration = Duration::from_millis(tick_duration_ms);
|
||||
let handle = tokio::spawn(async move {
|
||||
let mut current_tick = start_tick;
|
||||
loop {
|
||||
self.tick(current_tick, false);
|
||||
current_tick += 1;
|
||||
tokio::time::sleep(tick_duration).await;
|
||||
}
|
||||
});
|
||||
tokio::signal::ctrl_c().await?;
|
||||
handle.abort();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Run the simulation interactively: one tick per ENTER key press.
|
||||
pub fn run_manual(mut self, start_tick: u32, display_state: bool) -> anyhow::Result<()> {
|
||||
info!("Manual mode: press ENTER to advance a tick, Ctrl-C to quit");
|
||||
let mut current_tick = start_tick;
|
||||
let mut line = String::new();
|
||||
loop {
|
||||
line.clear();
|
||||
std::io::stdin().read_line(&mut line)?;
|
||||
info!("Tick {current_tick}");
|
||||
self.tick(current_tick, display_state);
|
||||
current_tick += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Driving logic for the concrete `Ts = Instant` timestamp flavour.
|
||||
impl MixSimDriver<Instant> {
|
||||
/// Start the simulation in either manual or automatic mode.
|
||||
pub async fn run(self, manual_mode: bool, tick_duration_ms: u64) -> anyhow::Result<()> {
|
||||
if manual_mode {
|
||||
tracing::error!("Instant-based MixSim is incompatible with manual driving mode");
|
||||
Ok(())
|
||||
} else {
|
||||
self.run_automatic(tick_duration_ms).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Run the simulation automatically, advancing one tick every
|
||||
/// `tick_duration_ms` milliseconds until Ctrl-C is received.
|
||||
pub async fn run_automatic(mut self, tick_duration_ms: u64) -> anyhow::Result<()> {
|
||||
info!("Automatic mode: tick duration : {tick_duration_ms} ms");
|
||||
let tick_duration = Duration::from_millis(tick_duration_ms);
|
||||
let handle = tokio::spawn(async move {
|
||||
loop {
|
||||
let current_tick = Instant::now();
|
||||
self.tick(current_tick, false);
|
||||
tokio::time::sleep(tick_duration).await;
|
||||
}
|
||||
});
|
||||
tokio::signal::ctrl_c().await?;
|
||||
handle.abort();
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Which simulation driver to use.
|
||||
#[derive(Clone, Debug, Default, strum::Display, strum::EnumString)]
|
||||
#[strum(serialize_all = "kebab-case")]
|
||||
pub enum SimDriver {
|
||||
/// Simple pass-through packets, discrete tick counter.
|
||||
Simple,
|
||||
/// Full Sphinx encryption, wall-clock timestamps, automatic mode only.
|
||||
Sphinx,
|
||||
/// Full Sphinx encryption, discrete tick counter, supports manual mode.
|
||||
#[default]
|
||||
DiscreteSphinx,
|
||||
}
|
||||
|
||||
impl SimDriver {
|
||||
/// Dispatch to the appropriate concrete driver and start the simulation.
|
||||
pub async fn run(
|
||||
self,
|
||||
topology: String,
|
||||
manual: bool,
|
||||
display_state: bool,
|
||||
tick_duration_ms: u64,
|
||||
) -> anyhow::Result<()> {
|
||||
match self {
|
||||
SimDriver::Simple => {
|
||||
SimpleMixDriver::new(topology)?
|
||||
.run(manual, display_state, tick_duration_ms)
|
||||
.await
|
||||
}
|
||||
SimDriver::Sphinx => {
|
||||
SphinxMixDriver::new(topology)?
|
||||
.run(manual, tick_duration_ms)
|
||||
.await
|
||||
}
|
||||
SimDriver::DiscreteSphinx => {
|
||||
DiscreteSphinxMixDriver::new(topology)?
|
||||
.run(manual, display_state, tick_duration_ms)
|
||||
.await
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! [`SimpleMixDriver`] — concrete driver using the simple (non-Sphinx) packet pipeline.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::Context;
|
||||
|
||||
use crate::{
|
||||
client::{MixSimClient, simple::SimpleClient},
|
||||
driver::MixSimDriver,
|
||||
node::{MixSimNode, simple::SimpleNode},
|
||||
topology::{Topology, directory::Directory},
|
||||
};
|
||||
|
||||
/// Concrete [`MixSimDriver`] instantiation that uses [`SimplePacket`]s and a
|
||||
/// pass-through processing pipeline.
|
||||
///
|
||||
/// Each mix node runs a [`SimpleProcessingNode`] that forwards packets
|
||||
/// unchanged to the next node in the topology; each client uses a
|
||||
/// [`SimpleClientWrappingPipeline`] with no Sphinx layering, reliability
|
||||
/// encoding, or obfuscation.
|
||||
///
|
||||
/// [`SimpleProcessingNode`]: crate::node::simple::SimpleProcessingNode
|
||||
/// [`SimpleClientWrappingPipeline`]: crate::client::simple::SimpleClientWrappingPipeline
|
||||
pub struct SimpleMixDriver(MixSimDriver<u32>);
|
||||
|
||||
impl SimpleMixDriver {
|
||||
/// Load a topology JSON file and initialise the driver with simple pipelines.
|
||||
pub fn new(topology: String) -> anyhow::Result<Self> {
|
||||
let topology_data =
|
||||
std::fs::read_to_string(&topology).context("Failed to read topology file")?;
|
||||
let topology: Topology =
|
||||
serde_json::from_str(&topology_data).context("Topology file malformed")?;
|
||||
|
||||
let directory: Arc<Directory> = Arc::new((&topology).into());
|
||||
|
||||
let mut nodes: Vec<Box<dyn MixSimNode<u32> + Send>> =
|
||||
Vec::with_capacity(topology.nodes.len());
|
||||
for top_node in topology.nodes {
|
||||
let node = SimpleNode::new(top_node, directory.clone())?;
|
||||
nodes.push(Box::new(node));
|
||||
}
|
||||
|
||||
let mut clients: Vec<Box<dyn MixSimClient<u32> + Send>> =
|
||||
Vec::with_capacity(topology.clients.len());
|
||||
for top_client in topology.clients {
|
||||
let client = SimpleClient::new(top_client, directory.clone())?;
|
||||
clients.push(Box::new(client));
|
||||
}
|
||||
|
||||
Ok(SimpleMixDriver(MixSimDriver::new(nodes, clients)))
|
||||
}
|
||||
|
||||
/// Run the simulation; delegates to [`MixSimDriver::run`].
|
||||
pub async fn run(self, manual_mode: bool, display_state: bool, tick_duration_ms: u64) -> anyhow::Result<()> {
|
||||
self.0.run(manual_mode, display_state, 0, tick_duration_ms).await
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Sphinx-based driver variants.
|
||||
//!
|
||||
//! Two flavours are provided:
|
||||
//!
|
||||
//! * [`SphinxMixDriver`] — wall-clock ([`Instant`]) timestamps; automatic mode only.
|
||||
//! * [`DiscreteSphinxMixDriver`] — discrete `u32` tick counter (1 tick = 1 ms);
|
||||
//! supports both automatic and manual stepping modes.
|
||||
|
||||
use std::{sync::Arc, time::Instant};
|
||||
|
||||
use anyhow::Context;
|
||||
use rand::rngs::OsRng;
|
||||
|
||||
use crate::{
|
||||
client::{MixSimClient, sphinx::SphinxClient},
|
||||
driver::MixSimDriver,
|
||||
node::{MixSimNode, sphinx::SphinxNode},
|
||||
topology::{Topology, directory::Directory},
|
||||
};
|
||||
|
||||
/// Concrete [`MixSimDriver`] instantiation that uses [`SphinxPacket`]s.
|
||||
pub struct SphinxMixDriver(MixSimDriver<Instant>);
|
||||
|
||||
impl SphinxMixDriver {
|
||||
/// Load a topology JSON file and initialise the driver with Sphinx pipelines.
|
||||
pub fn new(topology: String) -> anyhow::Result<Self> {
|
||||
let topology_data =
|
||||
std::fs::read_to_string(&topology).context("Failed to read topology file")?;
|
||||
let topology: Topology =
|
||||
serde_json::from_str(&topology_data).context("Topology file malformed")?;
|
||||
|
||||
let directory: Arc<Directory> = Arc::new((&topology).into());
|
||||
|
||||
let mut nodes: Vec<Box<dyn MixSimNode<Instant> + Send>> =
|
||||
Vec::with_capacity(topology.nodes.len());
|
||||
for top_node in topology.nodes {
|
||||
let node = SphinxNode::new(top_node, directory.clone())?;
|
||||
nodes.push(Box::new(node));
|
||||
}
|
||||
|
||||
let mut clients: Vec<Box<dyn MixSimClient<Instant> + Send>> =
|
||||
Vec::with_capacity(topology.clients.len());
|
||||
for top_client in topology.clients {
|
||||
let client = SphinxClient::new(top_client, directory.clone(), Instant::now(), OsRng)?;
|
||||
clients.push(Box::new(client));
|
||||
}
|
||||
|
||||
Ok(SphinxMixDriver(MixSimDriver::new(nodes, clients)))
|
||||
}
|
||||
|
||||
/// Run the simulation; delegates to [`MixSimDriver::run`].
|
||||
///
|
||||
/// `manual_mode` is ignored: [`Instant`]-based drivers cannot be stepped
|
||||
/// manually because wall-clock time cannot be advanced by keypress.
|
||||
pub async fn run(self, _manual_mode: bool, tick_duration_ms: u64) -> anyhow::Result<()> {
|
||||
self.0.run(false, tick_duration_ms).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Concrete [`MixSimDriver`] instantiation that uses full Sphinx encryption with a
|
||||
/// discrete tick counter.
|
||||
///
|
||||
/// Each tick corresponds to 1 ms of simulated time, enabling deterministic
|
||||
/// stepping and delay arithmetic without requiring wall-clock time. This is
|
||||
/// the default driver and the only Sphinx variant that supports manual mode.
|
||||
pub struct DiscreteSphinxMixDriver(MixSimDriver<u32>);
|
||||
|
||||
impl DiscreteSphinxMixDriver {
|
||||
const START_TICK: u32 = 0;
|
||||
|
||||
/// Load a topology JSON file and initialise the driver with Sphinx pipelines.
|
||||
pub fn new(topology: String) -> anyhow::Result<Self> {
|
||||
let topology_data =
|
||||
std::fs::read_to_string(&topology).context("Failed to read topology file")?;
|
||||
let topology: Topology =
|
||||
serde_json::from_str(&topology_data).context("Topology file malformed")?;
|
||||
|
||||
let directory: Arc<Directory> = Arc::new((&topology).into());
|
||||
|
||||
let mut nodes: Vec<Box<dyn MixSimNode<u32> + Send>> =
|
||||
Vec::with_capacity(topology.nodes.len());
|
||||
for top_node in topology.nodes {
|
||||
let node = SphinxNode::new(top_node, directory.clone())?;
|
||||
nodes.push(Box::new(node));
|
||||
}
|
||||
|
||||
let mut clients: Vec<Box<dyn MixSimClient<u32> + Send>> =
|
||||
Vec::with_capacity(topology.clients.len());
|
||||
for top_client in topology.clients {
|
||||
let client = SphinxClient::new(top_client, directory.clone(), Self::START_TICK, OsRng)?;
|
||||
clients.push(Box::new(client));
|
||||
}
|
||||
|
||||
Ok(DiscreteSphinxMixDriver(MixSimDriver::new(nodes, clients)))
|
||||
}
|
||||
|
||||
/// Run the simulation; delegates to [`MixSimDriver::run`].
|
||||
pub async fn run(
|
||||
self,
|
||||
manual_mode: bool,
|
||||
display_state: bool,
|
||||
tick_duration_ms: u64,
|
||||
) -> anyhow::Result<()> {
|
||||
self.0
|
||||
.run(
|
||||
manual_mode,
|
||||
display_state,
|
||||
Self::START_TICK,
|
||||
tick_duration_ms,
|
||||
)
|
||||
.await
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! # nym-mix-sim
|
||||
//!
|
||||
//! A discrete-time simulator for a Nym mixnet, intended for local testing and
|
||||
//! experimentation. The simulator models a network of mix nodes that exchange
|
||||
//! UDP packets on localhost.
|
||||
//!
|
||||
//! ## Architecture overview
|
||||
//!
|
||||
//! ```text
|
||||
//! ┌──────────────┐ JSON ┌───────────────────────────────┐
|
||||
//! │ topology.json│ ─────────────▶ │ MixSimDriver │
|
||||
//! └──────────────┘ │ ├─ Node 0 (UDP :9000) │
|
||||
//! │ ├─ Node N (UDP :900N) │
|
||||
//! │ ├─ Client 0 (UDP :9500/:9600)│
|
||||
//! │ └─ Client C (UDP :950C/:960C)│
|
||||
//! └───────────────────────────────┘
|
||||
//!
|
||||
//! Each simulation tick:
|
||||
//! 1. client tick – every client drains its app socket, queues outgoing
|
||||
//! packets, and processes inbound mix packets
|
||||
//! 2. tick_incoming – every node drains its UDP socket into an inbound buffer
|
||||
//! 3. tick_processing – every node transforms buffered packets (mix operation)
|
||||
//! 4. tick_outgoing – every node forwards processed packets to the next hop
|
||||
//! ```
|
||||
//!
|
||||
//! ## Crate layout
|
||||
//!
|
||||
//! | Module | Purpose |
|
||||
//! |--------|---------|
|
||||
//! | [`driver`] | Top-level orchestrator; owns all nodes and clients, drives simulation ticks |
|
||||
//! | [`node`] | Individual mix node: UDP socket, inbound/outbound packet buffers |
|
||||
//! | [`client`] | Simulated client: injects application payloads into the mix network |
|
||||
//! | [`packet`] | Wire format types and the [`packet::WirePacketFormat`] trait |
|
||||
//! | [`topology`] | Topology file types and the in-memory [`topology::directory::Directory`] |
|
||||
|
||||
pub mod client;
|
||||
pub mod driver;
|
||||
pub mod node;
|
||||
pub mod packet;
|
||||
pub mod topology;
|
||||
@@ -0,0 +1,130 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Binary entry point for the `nym-mix-sim` CLI tool.
|
||||
//!
|
||||
//! Provides two subcommands:
|
||||
//!
|
||||
//! * **`init-topology`** — generate a `topology.json` file describing N
|
||||
//! localhost mix nodes and C clients, with sequential UDP ports.
|
||||
//! * **`run`** — load a topology, spin up the chosen [`SimDriver`], and drive
|
||||
//! the simulation until Ctrl-C. Supports automatic tick mode (configurable
|
||||
//! interval via `--tick-duration-ms`) or manual RETURN-driven stepping
|
||||
//! (`--manual`). Use the standalone `mix-client` binary to inject packets
|
||||
//! while the simulation is running.
|
||||
|
||||
use std::net::SocketAddr;
|
||||
|
||||
use clap::{Parser, Subcommand};
|
||||
use nym_mix_sim::{
|
||||
driver::SimDriver,
|
||||
topology::{Topology, TopologyClient, TopologyNode},
|
||||
};
|
||||
use tracing::info;
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(name = "nym-mix-sim", about = "Nym mix network simulator")]
|
||||
struct Cli {
|
||||
#[command(subcommand)]
|
||||
command: Commands,
|
||||
}
|
||||
|
||||
#[derive(Subcommand)]
|
||||
enum Commands {
|
||||
/// Generate a topology.json file with a given number of nodes and one client
|
||||
InitTopology {
|
||||
/// Number of mix nodes to generate.
|
||||
///
|
||||
/// Each node receives an auto-assigned ID (0..N-1) and a sequential
|
||||
/// localhost address starting at `127.0.0.1:9000`.
|
||||
#[arg(short, long, default_value_t = 6)]
|
||||
nodes: u8,
|
||||
|
||||
/// Number of clients to generate.
|
||||
///
|
||||
/// Each client receives an auto-assigned ID (`N..N+C`) and two
|
||||
/// sequential localhost addresses: a mix-network socket starting at
|
||||
/// `127.0.0.1:9500` and an app socket starting at `127.0.0.1:9600`.
|
||||
#[arg(short, long, default_value_t = 2)]
|
||||
clients: u8,
|
||||
|
||||
/// Output file path
|
||||
#[arg(short, long, default_value = "topology.json")]
|
||||
output: String,
|
||||
},
|
||||
|
||||
/// Run the mix simulation with a given topology file
|
||||
Run {
|
||||
/// Path to the topology.json file
|
||||
#[arg(short, long, default_value = "topology.json")]
|
||||
topology: String,
|
||||
|
||||
/// Use manual (RETURN-driven) mode instead of automatic ticks.
|
||||
#[arg(short, long)]
|
||||
manual: bool,
|
||||
|
||||
/// Suppress node state display after each tick phase (manual mode only).
|
||||
#[arg(long)]
|
||||
no_display_state: bool,
|
||||
|
||||
/// Tick duration in milliseconds (automatic mode only).
|
||||
#[arg(short = 'd', long, default_value = "1")]
|
||||
tick_duration_ms: u64,
|
||||
|
||||
/// Simulation driver to use: simple | sphinx | discrete-sphinx (default).
|
||||
#[arg(long, default_value_t = SimDriver::DiscreteSphinx)]
|
||||
driver: SimDriver,
|
||||
},
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
nym_bin_common::logging::setup_tracing_logger();
|
||||
|
||||
let cli = Cli::parse();
|
||||
|
||||
match cli.command {
|
||||
Commands::InitTopology {
|
||||
nodes,
|
||||
clients,
|
||||
output,
|
||||
} => {
|
||||
info!("Generating topology with {nodes} node(s) and {clients} client(s)");
|
||||
let node_list = (0..nodes)
|
||||
.map(|id| {
|
||||
let addr = SocketAddr::from(([127, 0, 0, 1], 9000 + id as u16));
|
||||
TopologyNode::new(id, 100, addr)
|
||||
})
|
||||
.collect();
|
||||
// Client binds to the next port after all nodes.
|
||||
let client_list = (nodes..nodes + clients)
|
||||
.map(|id| {
|
||||
let mix_addr = SocketAddr::from(([127, 0, 0, 1], 9500 + id as u16));
|
||||
let app_addr = SocketAddr::from(([127, 0, 0, 1], 9600 + id as u16));
|
||||
TopologyClient::new(id, mix_addr, app_addr)
|
||||
})
|
||||
.collect();
|
||||
let topology = Topology {
|
||||
nodes: node_list,
|
||||
clients: client_list,
|
||||
};
|
||||
let json = serde_json::to_string_pretty(&topology)?;
|
||||
std::fs::write(&output, &json)?;
|
||||
info!("Topology written to {output}");
|
||||
}
|
||||
Commands::Run {
|
||||
topology,
|
||||
manual,
|
||||
no_display_state,
|
||||
tick_duration_ms,
|
||||
driver,
|
||||
} => {
|
||||
info!("Loading topology from {topology} with driver={driver}");
|
||||
driver
|
||||
.run(topology, manual, !no_display_state, tick_duration_ms)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,247 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use std::{
|
||||
fmt::Debug,
|
||||
io::ErrorKind,
|
||||
net::{SocketAddr, UdpSocket},
|
||||
sync::Arc,
|
||||
};
|
||||
|
||||
use nym_lp_data::{AddressedTimedData, TimedData};
|
||||
|
||||
use crate::{packet::WirePacketFormat, topology::directory::Directory};
|
||||
|
||||
pub mod simple;
|
||||
pub mod sphinx;
|
||||
|
||||
/// Compact identifier for a mix node in the simulation topology.
|
||||
///
|
||||
/// `u8` keeps the IDs small (max 255 nodes) and is large enough for any
|
||||
/// realistic simulated topology.
|
||||
pub type NodeId = u8;
|
||||
|
||||
/// Driver-facing interface for a mix node.
|
||||
///
|
||||
/// Erases `Pkt` and `Pn` so that [`MixSimDriver`] only needs `Ts`.
|
||||
/// Implemented by [`BaseNode<Ts, Pkt, Pn>`] for any compatible `Pkt` and
|
||||
/// `Pn`.
|
||||
///
|
||||
/// [`MixSimDriver`]: crate::driver::MixSimDriver
|
||||
pub trait MixSimNode<Ts: Clone + PartialOrd + Debug + Send>: Send {
|
||||
/// **Phase 1** — drain the UDP socket into the inbound buffer, tagging each
|
||||
/// received packet with `timestamp`.
|
||||
fn tick_incoming(&mut self, timestamp: Ts);
|
||||
|
||||
/// **Phase 2** — pass every buffered packet through the mix pipeline and
|
||||
/// move the results into the outbound queue.
|
||||
fn tick_processing(&mut self, timestamp: Ts);
|
||||
|
||||
/// **Phase 3** — forward all outbound packets whose scheduled timestamp is
|
||||
/// ≤ `timestamp` to their next-hop address.
|
||||
fn tick_outgoing(&mut self, timestamp: Ts);
|
||||
|
||||
/// Pretty-print the node's current buffer state to stdout (used in manual mode).
|
||||
fn display_state(&self);
|
||||
}
|
||||
|
||||
/// Minimal pipeline interface used by [`BaseNode`].
|
||||
///
|
||||
/// Hides the `Frame` and message-marker type parameters of
|
||||
/// [`MixnodeProcessingPipeline`] so that [`BaseNode`] only needs
|
||||
/// `<Ts, Pkt, Pipeline>` rather than five generics.
|
||||
///
|
||||
/// Implement [`MixnodeProcessingPipeline`] on your concrete type and then add
|
||||
/// a trivial delegation impl of this trait; the two-line body just calls
|
||||
/// through to [`MixnodeProcessingPipeline::process`].
|
||||
///
|
||||
/// [`MixnodeProcessingPipeline`]: nym_lp_data::mixnodes::traits::MixnodeProcessingPipeline
|
||||
/// [`MixnodeProcessingPipeline::process`]: nym_lp_data::mixnodes::traits::MixnodeProcessingPipeline::process
|
||||
pub trait ProcessingNode<Ts, Pkt>: Send {
|
||||
fn process(
|
||||
&mut self,
|
||||
input: TimedData<Ts, Pkt>,
|
||||
timestamp: Ts,
|
||||
) -> anyhow::Result<Vec<AddressedTimedData<Ts, Pkt, NodeId>>>;
|
||||
}
|
||||
|
||||
/// Full mix-node state: UDP transport, routing directory, packet buffers, and
|
||||
/// processing pipeline.
|
||||
///
|
||||
/// `Ts` is the timestamp / tick-context type. `Pkt` is the wire packet type
|
||||
/// (e.g. [`SimplePacket`] or [`SimMixPacket`]). `Pn` is any type that
|
||||
/// implements [`ProcessingNode<Ts, Pkt>`].
|
||||
///
|
||||
/// Concrete node variants (`SimpleNode`, `SphinxNode`, …) are type aliases
|
||||
/// over this struct and only need to supply a `new()` constructor that wires
|
||||
/// up the right pipeline.
|
||||
///
|
||||
/// [`SimplePacket`]: crate::packet::simple::SimplePacket
|
||||
/// [`SimMixPacket`]: crate::packet::sphinx::SimMixPacket
|
||||
pub struct BaseNode<Ts, Pkt, Pn> {
|
||||
pub(crate) id: NodeId,
|
||||
_reliability: u8, // Unused yet, can be used later for testing the reliability layer
|
||||
pub(crate) socket_address: SocketAddr,
|
||||
socket: UdpSocket,
|
||||
directory: Arc<Directory>,
|
||||
|
||||
packets_to_process: Vec<TimedData<Ts, Pkt>>,
|
||||
processed_packets: Vec<AddressedTimedData<Ts, Pkt, NodeId>>,
|
||||
|
||||
processing_node: Pn,
|
||||
}
|
||||
|
||||
impl<Ts, Pkt, Pn> BaseNode<Ts, Pkt, Pn> {
|
||||
/// Bind a non-blocking UDP socket to `socket_address` and initialise the
|
||||
/// node with the given `pipeline`.
|
||||
pub(crate) fn with_pipeline(
|
||||
id: NodeId,
|
||||
reliability: u8,
|
||||
socket_address: SocketAddr,
|
||||
directory: Arc<Directory>,
|
||||
processing_node: Pn,
|
||||
) -> anyhow::Result<Self> {
|
||||
let socket = UdpSocket::bind(socket_address)?;
|
||||
socket.set_nonblocking(true)?;
|
||||
Ok(Self {
|
||||
id,
|
||||
_reliability: reliability,
|
||||
socket_address,
|
||||
socket,
|
||||
directory,
|
||||
packets_to_process: Vec::new(),
|
||||
processed_packets: Vec::new(),
|
||||
processing_node,
|
||||
})
|
||||
}
|
||||
|
||||
/// Send `packet` to the node or client identified by `node_id`.
|
||||
///
|
||||
/// Resolves `node_id` against the shared [`Directory`], serialises via
|
||||
/// [`WirePacketFormat::to_bytes`], and dispatches with a single `sendto`.
|
||||
/// Errors are logged but not propagated.
|
||||
pub fn send_to_node(&self, node_id: NodeId, packet: Pkt)
|
||||
where
|
||||
Pkt: WirePacketFormat,
|
||||
{
|
||||
if let Some(node) = self.directory.node(node_id) {
|
||||
if let Err(e) = self.socket.send_to(&packet.to_bytes(), node.addr) {
|
||||
tracing::error!(
|
||||
"[Node {}] Failed to send data to node {node_id} : {e}",
|
||||
self.id
|
||||
);
|
||||
} else {
|
||||
tracing::debug!(
|
||||
"[Node {}] Successfully sent a packet to node {node_id}",
|
||||
self.id
|
||||
);
|
||||
}
|
||||
} else if let Some(client) = self.directory.client(node_id) {
|
||||
if let Err(e) = self.socket.send_to(&packet.to_bytes(), client) {
|
||||
tracing::error!(
|
||||
"[Node {}] Failed to send data to client {node_id} : {e}",
|
||||
self.id
|
||||
);
|
||||
} else {
|
||||
tracing::debug!(
|
||||
"[Node {}] Successfully sent a packet to client {node_id} @ {client}",
|
||||
self.id
|
||||
);
|
||||
}
|
||||
} else {
|
||||
tracing::error!(
|
||||
"[Node {}] Trying to send to non-existing node/client {node_id}",
|
||||
self.id
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Attempt to receive one UDP datagram and deserialise it as `Pkt`.
|
||||
///
|
||||
/// Returns `None` when the socket would block (no datagram waiting).
|
||||
pub fn recv_packet(&self) -> Option<anyhow::Result<Pkt>>
|
||||
where
|
||||
Pkt: WirePacketFormat,
|
||||
{
|
||||
let mut buf = [0; 1500];
|
||||
let (nb_bytes, src_address) = match self.socket.recv_from(&mut buf) {
|
||||
Ok(result) => result,
|
||||
Err(e) if e.kind() == ErrorKind::WouldBlock => return None,
|
||||
Err(e) => {
|
||||
tracing::error!("Error receiving packet : {e}");
|
||||
return None;
|
||||
}
|
||||
};
|
||||
tracing::debug!(
|
||||
"[Node {}] Received {nb_bytes} bytes from {src_address}",
|
||||
self.id
|
||||
);
|
||||
Some(Pkt::try_from_bytes(&buf[..nb_bytes]))
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts, Pkt, Pn> MixSimNode<Ts> for BaseNode<Ts, Pkt, Pn>
|
||||
where
|
||||
Ts: Clone + PartialOrd + Debug + Send,
|
||||
Pkt: WirePacketFormat + Debug + Send,
|
||||
Pn: ProcessingNode<Ts, Pkt>,
|
||||
{
|
||||
fn tick_incoming(&mut self, timestamp: Ts) {
|
||||
while let Some(maybe_packet) = self.recv_packet() {
|
||||
match maybe_packet {
|
||||
Ok(packet) => self
|
||||
.packets_to_process
|
||||
.push(TimedData::new(timestamp.clone(), packet)),
|
||||
Err(e) => tracing::error!("[Node {}] Failed to deserialize packet : {e}", self.id),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn tick_processing(&mut self, timestamp: Ts) {
|
||||
while let Some(packet) = self.packets_to_process.pop() {
|
||||
match self.processing_node.process(packet, timestamp.clone()) {
|
||||
Ok(processed_packets) => self.processed_packets.extend(processed_packets),
|
||||
Err(e) => {
|
||||
tracing::error!("[Node {}] Failed to process packet : {e}", self.id)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn tick_outgoing(&mut self, timestamp: Ts) {
|
||||
let to_send = self
|
||||
.processed_packets
|
||||
.extract_if(.., |pkt| pkt.data.timestamp <= timestamp)
|
||||
.collect::<Vec<_>>();
|
||||
for pkt in to_send {
|
||||
self.send_to_node(pkt.dst, pkt.data.data);
|
||||
}
|
||||
}
|
||||
|
||||
fn display_state(&self) {
|
||||
println!("│ Node {:2} @ {}", self.id, self.socket_address);
|
||||
if self.packets_to_process.is_empty() {
|
||||
println!("│ to_process buffer: (empty)");
|
||||
} else {
|
||||
println!(
|
||||
"│ to_process buffer: {} packet(s)",
|
||||
self.packets_to_process.len()
|
||||
);
|
||||
for (i, pkt) in self.packets_to_process.iter().enumerate() {
|
||||
println!("│ [{i}] {pkt:?}");
|
||||
}
|
||||
}
|
||||
|
||||
if self.processed_packets.is_empty() {
|
||||
println!("│ processed buffer: (empty)");
|
||||
} else {
|
||||
println!(
|
||||
"│ processed buffer: {} packet(s)",
|
||||
self.processed_packets.len()
|
||||
);
|
||||
for (i, pkt) in self.processed_packets.iter().enumerate() {
|
||||
println!("│ [{i}] {pkt:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,158 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! [`SimpleNode`] — mix node using the simple (non-Sphinx) packet pipeline.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use nym_lp_data::{
|
||||
AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload,
|
||||
common::traits::{
|
||||
Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline,
|
||||
WireWrappingPipeline,
|
||||
},
|
||||
mixnodes::traits::MixnodeProcessingPipeline,
|
||||
};
|
||||
|
||||
use crate::{
|
||||
node::{BaseNode, NodeId, ProcessingNode},
|
||||
packet::simple::{
|
||||
SimpleFrame, SimpleMessage, SimplePacket, SimpleWireUnwrapper, SimpleWireWrapper,
|
||||
},
|
||||
topology::{TopologyNode, directory::Directory},
|
||||
};
|
||||
|
||||
/// A mix-node that uses the simple (non-Sphinx) packet pipeline.
|
||||
///
|
||||
/// This is a type alias for [`BaseNode`] specialised to [`SimplePacket`] and
|
||||
/// [`SimpleProcessingNode`]. All tick logic lives in the generic
|
||||
/// [`MixSimNode`] impl on `BaseNode`.
|
||||
///
|
||||
/// [`MixSimNode`]: crate::node::MixSimNode
|
||||
pub type SimpleNode<Ts> = BaseNode<Ts, SimplePacket, SimpleProcessingNode>;
|
||||
|
||||
impl<Ts> SimpleNode<Ts> {
|
||||
/// Create a [`SimpleNode`] from a [`TopologyNode`] description by binding a
|
||||
/// non-blocking UDP socket to `node.socket_address`.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if the UDP socket cannot be bound or set non-blocking.
|
||||
pub fn new(topology_node: TopologyNode, directory: Arc<Directory>) -> anyhow::Result<Self> {
|
||||
let pipeline = SimpleProcessingNode::new(topology_node.node_id);
|
||||
BaseNode::with_pipeline(
|
||||
topology_node.node_id,
|
||||
topology_node.reliability,
|
||||
topology_node.socket_address,
|
||||
directory,
|
||||
pipeline,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> ProcessingNode<Ts, SimplePacket> for SimpleProcessingNode {
|
||||
fn process(
|
||||
&mut self,
|
||||
input: TimedData<Ts, SimplePacket>,
|
||||
timestamp: Ts,
|
||||
) -> anyhow::Result<Vec<AddressedTimedData<Ts, SimplePacket, NodeId>>> {
|
||||
MixnodeProcessingPipeline::<Ts, SimplePacket, SimpleMessage, NodeId>::process(
|
||||
self, input, timestamp,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// A simple [`MixnodeProcessingPipeline`] for [`SimplePacket`].
|
||||
///
|
||||
/// Demonstrates the full pipeline: unwraps the incoming packet through the
|
||||
/// wire layer (transport → frame → payload), applies a routing decision in
|
||||
/// [`mix`] (forwards to `self.id + 1`), then re-wraps the outgoing payload
|
||||
/// (payload → frame → transport) before sending.
|
||||
pub struct SimpleProcessingNode {
|
||||
id: NodeId,
|
||||
wrapper: SimpleWireWrapper,
|
||||
unwrapper: SimpleWireUnwrapper,
|
||||
}
|
||||
|
||||
impl SimpleProcessingNode {
|
||||
/// Construct a pipeline for the node identified by `id`.
|
||||
pub fn new(id: NodeId) -> Self {
|
||||
Self {
|
||||
id,
|
||||
wrapper: SimpleWireWrapper,
|
||||
unwrapper: SimpleWireUnwrapper,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> MixnodeProcessingPipeline<Ts, SimplePacket, SimpleMessage, NodeId>
|
||||
for SimpleProcessingNode
|
||||
{
|
||||
/// Route the payload to the next node in the chain (`self.id + 1`).
|
||||
///
|
||||
/// This is a trivial fixed routing rule used for simulation testing.
|
||||
/// Real mix nodes would perform cryptographic route unwrapping here.
|
||||
fn mix(
|
||||
&mut self,
|
||||
_: SimpleMessage,
|
||||
payload: TimedPayload<Ts>,
|
||||
_timestamp: Ts,
|
||||
) -> Vec<AddressedTimedPayload<Ts, NodeId>> {
|
||||
vec![(self.id + 1, payload).into()]
|
||||
}
|
||||
}
|
||||
|
||||
// Delegation of subtraits
|
||||
impl<Ts: Clone> Framing<Ts, NodeId> for SimpleProcessingNode {
|
||||
type Frame = SimpleFrame;
|
||||
const OVERHEAD_SIZE: usize = <SimpleWireWrapper as Framing<Ts, _>>::OVERHEAD_SIZE;
|
||||
fn to_frame(
|
||||
&self,
|
||||
payload: AddressedTimedPayload<Ts, NodeId>,
|
||||
frame_size: usize,
|
||||
) -> Vec<AddressedTimedData<Ts, SimpleFrame, NodeId>> {
|
||||
self.wrapper.to_frame(payload, frame_size)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> Transport<Ts, SimplePacket, NodeId> for SimpleProcessingNode {
|
||||
type Frame = SimpleFrame;
|
||||
const OVERHEAD_SIZE: usize = <SimpleWireWrapper as Transport<Ts, _, _>>::OVERHEAD_SIZE;
|
||||
fn to_transport_packet(
|
||||
&self,
|
||||
frame: AddressedTimedData<Ts, SimpleFrame, NodeId>,
|
||||
) -> AddressedTimedData<Ts, SimplePacket, NodeId> {
|
||||
self.wrapper.to_transport_packet(frame)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> WireWrappingPipeline<Ts, SimplePacket, NodeId> for SimpleProcessingNode {
|
||||
fn packet_size(&self) -> usize {
|
||||
<SimpleWireWrapper as WireWrappingPipeline<Ts, _, _>>::packet_size(&self.wrapper)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts> FramingUnwrap<Ts, SimpleMessage> for SimpleProcessingNode {
|
||||
type Frame = SimpleFrame;
|
||||
fn frame_to_message(
|
||||
&mut self,
|
||||
frame: TimedData<Ts, SimpleFrame>,
|
||||
) -> Option<(TimedPayload<Ts>, SimpleMessage)> {
|
||||
self.unwrapper.frame_to_message(frame)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> TransportUnwrap<Ts, SimplePacket> for SimpleProcessingNode {
|
||||
type Frame = SimpleFrame;
|
||||
fn packet_to_frame(
|
||||
&self,
|
||||
packet: SimplePacket,
|
||||
timestamp: Ts,
|
||||
) -> anyhow::Result<TimedData<Ts, SimpleFrame>> {
|
||||
self.unwrapper.packet_to_frame(packet, timestamp)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> WireUnwrappingPipeline<Ts, SimplePacket, SimpleMessage> for SimpleProcessingNode {}
|
||||
@@ -0,0 +1,173 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! [`SphinxNode`] — mix node using the full Sphinx packet pipeline.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use nym_crypto::asymmetric::x25519;
|
||||
use nym_lp_data::{
|
||||
AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload,
|
||||
common::helpers::{NoOpWireUnwrapper, NoOpWireWrapper},
|
||||
mixnodes::traits::MixnodeProcessingPipeline,
|
||||
};
|
||||
use nym_sphinx::SphinxPacket;
|
||||
|
||||
use crate::{
|
||||
node::{BaseNode, NodeId, ProcessingNode},
|
||||
packet::{
|
||||
WirePacketFormat,
|
||||
sphinx::{AddDelay, SimMixPacket, SphinxMessage, SurbAck},
|
||||
},
|
||||
topology::{TopologyNode, directory::Directory},
|
||||
};
|
||||
|
||||
/// A mix-node that uses the Sphinx packet pipeline.
|
||||
///
|
||||
/// This is a type alias for [`BaseNode`] specialised to [`SimMixPacket`] and
|
||||
/// [`SphinxProcessingNode`]. All tick logic lives in the generic
|
||||
/// [`MixSimNode`] impl on `BaseNode`.
|
||||
///
|
||||
/// [`MixSimNode`]: crate::node::MixSimNode
|
||||
pub type SphinxNode<Ts> = BaseNode<Ts, SimMixPacket, SphinxProcessingNode>;
|
||||
|
||||
impl<Ts> SphinxNode<Ts> {
|
||||
/// Create a [`SphinxNode`] from a [`TopologyNode`] description by binding a
|
||||
/// non-blocking UDP socket to `node.socket_address`.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if the UDP socket cannot be bound or set non-blocking.
|
||||
pub fn new(topology_node: TopologyNode, directory: Arc<Directory>) -> anyhow::Result<Self> {
|
||||
let pipeline =
|
||||
SphinxProcessingNode::new(topology_node.node_id, topology_node.sphinx_private_key);
|
||||
BaseNode::with_pipeline(
|
||||
topology_node.node_id,
|
||||
topology_node.reliability,
|
||||
topology_node.socket_address,
|
||||
directory,
|
||||
pipeline,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts> ProcessingNode<Ts, SimMixPacket> for SphinxProcessingNode
|
||||
where
|
||||
Ts: AddDelay + Clone,
|
||||
{
|
||||
fn process(
|
||||
&mut self,
|
||||
input: TimedData<Ts, SimMixPacket>,
|
||||
timestamp: Ts,
|
||||
) -> anyhow::Result<Vec<AddressedTimedData<Ts, SimMixPacket, NodeId>>> {
|
||||
MixnodeProcessingPipeline::<Ts, SimMixPacket, SphinxMessage, NodeId>::process(
|
||||
self, input, timestamp,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/// A [`MixnodeProcessingPipeline`] for [`SphinxPacket`].
|
||||
///
|
||||
/// Uses no-op framing and transport wrappers because a Sphinx packet is already
|
||||
/// its own self-contained wire unit — no additional framing or transport header
|
||||
/// is needed. The real work happens in [`mix`](SphinxProcessingNode::mix), which
|
||||
/// peels one onion layer with the node's private key and extracts the next-hop
|
||||
/// address and per-hop delay.
|
||||
pub struct SphinxProcessingNode {
|
||||
id: NodeId,
|
||||
sphinx_secret: x25519::PrivateKey,
|
||||
}
|
||||
|
||||
impl SphinxProcessingNode {
|
||||
/// Construct a pipeline for the node identified by `node_id`, using
|
||||
/// `sphinx_secret` to decrypt incoming Sphinx packets.
|
||||
pub fn new(node_id: NodeId, sphinx_secret: x25519::PrivateKey) -> Self {
|
||||
Self {
|
||||
id: node_id,
|
||||
sphinx_secret,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts> MixnodeProcessingPipeline<Ts, SimMixPacket, SphinxMessage, NodeId> for SphinxProcessingNode
|
||||
where
|
||||
Ts: AddDelay + Clone,
|
||||
{
|
||||
/// Peel one Sphinx layer and forward or deliver the result.
|
||||
///
|
||||
/// - **ForwardHop**: extracts the next-hop packet, address (byte 0 of the
|
||||
/// 32-byte address field encodes the [`NodeId`]), and per-hop delay; schedules
|
||||
/// the re-wrapped packet at `timestamp + delay`.
|
||||
/// - **FinalHop**: delivers the plaintext payload directly to the destination
|
||||
/// client (identified by byte 0 of the destination address).
|
||||
fn mix(
|
||||
&mut self,
|
||||
_: SphinxMessage,
|
||||
payload: TimedPayload<Ts>,
|
||||
timestamp: Ts,
|
||||
) -> Vec<AddressedTimedPayload<Ts, NodeId>> {
|
||||
// SAFETY: Given the no-op unwrapper used here, payload.data is always a
|
||||
// valid serialised SphinxPacket at this point.
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let sphinx_packet = SphinxPacket::from_bytes(&payload.data).unwrap();
|
||||
|
||||
match sphinx_packet.process(self.sphinx_secret.inner()) {
|
||||
Ok(packet) => match packet.data {
|
||||
nym_sphinx::ProcessedPacketData::ForwardHop {
|
||||
next_hop_packet,
|
||||
next_hop_address,
|
||||
delay,
|
||||
} => {
|
||||
let timed_sphinx = AddressedTimedData::new(
|
||||
timestamp.add_delay(delay),
|
||||
next_hop_packet.to_bytes(),
|
||||
next_hop_address.as_bytes()[0],
|
||||
);
|
||||
vec![timed_sphinx]
|
||||
}
|
||||
nym_sphinx::ProcessedPacketData::FinalHop {
|
||||
destination,
|
||||
identifier: _,
|
||||
payload,
|
||||
} => {
|
||||
if let Ok(plaintext) = payload
|
||||
.recover_plaintext()
|
||||
.inspect_err(|e| tracing::warn!("Impossible to recover plaintext : {e}"))
|
||||
{
|
||||
let (surb_ack_bytes, message) = SurbAck::extract_ack_and_message(plaintext);
|
||||
let mut packets_to_forward = vec![AddressedTimedData::new(
|
||||
timestamp.clone(),
|
||||
message,
|
||||
destination.as_bytes()[0],
|
||||
)];
|
||||
if !surb_ack_bytes.is_empty()
|
||||
&& let Ok((next_hop, surb_ack)) = SurbAck::try_recover_first_hop_packet(
|
||||
&surb_ack_bytes,
|
||||
)
|
||||
.inspect_err(|e| tracing::warn!("Fail to deserialize SURB Ack : {e}"))
|
||||
{
|
||||
packets_to_forward.push(AddressedTimedData::new(
|
||||
timestamp,
|
||||
surb_ack.to_bytes(),
|
||||
next_hop,
|
||||
));
|
||||
}
|
||||
packets_to_forward
|
||||
} else {
|
||||
Vec::new()
|
||||
}
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
tracing::error!("[Node {}] Failed to process a sphinx packet : {e}", self.id);
|
||||
Vec::new()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Boilerplate subtraits delegation
|
||||
impl NoOpWireWrapper for SphinxProcessingNode {}
|
||||
impl NoOpWireUnwrapper for SphinxProcessingNode {}
|
||||
@@ -0,0 +1,39 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Packet types and the generic wire-format trait used by the simulation.
|
||||
//!
|
||||
//! The central abstraction is [`WirePacketFormat`]: a trait that any packet
|
||||
//! type must implement to participate in a simulation. It covers only
|
||||
//! wire serialisation; mix logic is handled separately by
|
||||
//! [`nym_lp_data::mixnodes::traits::MixnodeProcessingPipeline`].
|
||||
//!
|
||||
|
||||
pub mod simple;
|
||||
pub mod sphinx;
|
||||
|
||||
/// Trait that every packet type must implement to participate in the simulation.
|
||||
///
|
||||
pub trait WirePacketFormat: Sized {
|
||||
/// Deserialise a packet from the raw bytes received off the wire.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Should return an error on length mismatch, invalid magic bytes, or any
|
||||
/// other malformed-datagram condition.
|
||||
fn try_from_bytes(bytes: &[u8]) -> anyhow::Result<Self>;
|
||||
|
||||
/// Serialise the packet to its on-wire byte representation, ready to be
|
||||
/// sent via UDP.
|
||||
fn to_bytes(&self) -> Vec<u8>;
|
||||
}
|
||||
|
||||
impl WirePacketFormat for Vec<u8> {
|
||||
fn try_from_bytes(bytes: &[u8]) -> anyhow::Result<Self> {
|
||||
Ok(bytes.to_vec())
|
||||
}
|
||||
|
||||
fn to_bytes(&self) -> Vec<u8> {
|
||||
self.clone()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,278 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use std::fmt::Debug;
|
||||
|
||||
use nym_common::debug::format_debug_bytes;
|
||||
use nym_lp_data::{
|
||||
AddressedTimedData, AddressedTimedPayload, TimedData, TimedPayload,
|
||||
common::traits::{
|
||||
Framing, FramingUnwrap, Transport, TransportUnwrap, WireUnwrappingPipeline,
|
||||
WireWrappingPipeline,
|
||||
},
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{node::NodeId, packet::WirePacketFormat};
|
||||
|
||||
/// A minimal, fixed-size packet used by the simulation.
|
||||
///
|
||||
/// ## Wire format
|
||||
///
|
||||
/// ```text
|
||||
/// ┌──────────────────┬──────────────────────────────────────────────────┐
|
||||
/// │ UUID (16 bytes) │ payload (48 bytes) │
|
||||
/// │ little-endian │ │
|
||||
/// └──────────────────┴──────────────────────────────────────────────────┘
|
||||
/// byte 0 16 64
|
||||
/// ```
|
||||
///
|
||||
/// The total on-wire size is always exactly [`SimplePacket::SIZE`] = 64 bytes.
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct SimplePacket {
|
||||
/// Universally unique identifier assigned at creation time (UUID v4).
|
||||
/// Used to correlate a packet across hops for debugging and tracing.
|
||||
pub id: Uuid,
|
||||
|
||||
/// Variable-length payload buffer.
|
||||
///
|
||||
/// Despite the type being `Vec<u8>`, the simulation always creates and
|
||||
/// expects exactly 48 bytes here (i.e. `SIZE - 16`). The `Vec` is used
|
||||
/// rather than a fixed array to keep serialisation simple.
|
||||
pub data: Vec<u8>,
|
||||
}
|
||||
|
||||
impl Debug for SimplePacket {
|
||||
/// Pretty-prints the packet ID followed by a hex dump of the payload via
|
||||
/// [`format_debug_bytes`].
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
writeln!(f, "SimplePacket {{")?;
|
||||
writeln!(f, " id: {:?},", self.id)?;
|
||||
writeln!(f, " data:")?;
|
||||
for line in format_debug_bytes(&self.data)?.lines() {
|
||||
writeln!(f, " {line}")?;
|
||||
}
|
||||
write!(f, "}}")
|
||||
}
|
||||
}
|
||||
|
||||
impl SimplePacket {
|
||||
/// On-wire size of a serialised [`SimplePacket`] in bytes.
|
||||
///
|
||||
/// Layout: 16 bytes UUID (LE) + 48 bytes payload = 64 bytes total.
|
||||
const SIZE: usize = 64;
|
||||
|
||||
/// Create a new [`SimplePacket`] with a freshly generated UUID v4 and the
|
||||
/// provided 48-byte payload.
|
||||
///
|
||||
/// The payload array is exactly `SIZE - 16 = 48` bytes so that the packet
|
||||
/// serialises to exactly [`SimplePacket::SIZE`] bytes.
|
||||
pub fn new(data: [u8; Self::SIZE - 16]) -> Self {
|
||||
Self {
|
||||
id: Uuid::new_v4(),
|
||||
data: data.to_vec(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Return the packet's UUID identifier.
|
||||
pub fn id(&self) -> Uuid {
|
||||
self.id
|
||||
}
|
||||
|
||||
/// Return a clone of the raw payload bytes.
|
||||
pub fn data(&self) -> Vec<u8> {
|
||||
self.data.clone()
|
||||
}
|
||||
|
||||
/// Serialise the packet to its fixed-size wire representation.
|
||||
///
|
||||
/// Layout: UUID as 16 little-endian bytes, followed by the 48-byte payload.
|
||||
/// The returned `Vec` is always exactly [`SimplePacket::SIZE`] bytes long.
|
||||
pub fn to_bytes(&self) -> Vec<u8> {
|
||||
// fixed-size serialization: 16-byte UUID followed by 48-byte payload
|
||||
let mut bytes = Vec::with_capacity(Self::SIZE);
|
||||
|
||||
bytes.extend_from_slice(&self.id.to_bytes_le()); // 16 bytes
|
||||
bytes.extend_from_slice(&self.data); // 48 bytes
|
||||
|
||||
bytes
|
||||
}
|
||||
|
||||
/// Deserialise a [`SimplePacket`] from a raw byte slice.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if `bytes.len() != SIZE` (64). Any other slice length
|
||||
/// indicates a truncated or corrupted UDP datagram.
|
||||
pub fn try_from_bytes(bytes: &[u8]) -> anyhow::Result<Self> {
|
||||
if bytes.len() != Self::SIZE {
|
||||
return Err(anyhow::anyhow!(
|
||||
"Length mismatch to deserialize a SimplePacket : Expected {}, got {}",
|
||||
Self::SIZE,
|
||||
bytes.len()
|
||||
));
|
||||
}
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let uuid = Uuid::from_bytes_le(bytes[0..16].try_into().unwrap());
|
||||
let data = bytes[16..Self::SIZE].to_vec();
|
||||
Ok(SimplePacket { id: uuid, data })
|
||||
}
|
||||
}
|
||||
|
||||
/// [`WirePacketFormat`] implementation for [`SimplePacket`].
|
||||
impl WirePacketFormat for SimplePacket {
|
||||
fn try_from_bytes(bytes: &[u8]) -> anyhow::Result<Self> {
|
||||
Self::try_from_bytes(bytes)
|
||||
}
|
||||
|
||||
fn to_bytes(&self) -> Vec<u8> {
|
||||
self.to_bytes()
|
||||
}
|
||||
}
|
||||
|
||||
/// Intermediate frame type used by [`SimpleClientPipeline`].
|
||||
///
|
||||
/// A `SimpleFrame` wraps a chunk of payload bytes with a fixed 7-byte magic
|
||||
/// header (`b"0FRAME0"`). It is produced by the [`Framing`] stage and
|
||||
/// consumed by the [`Transport`] stage, which packs it into a [`SimplePacket`].
|
||||
pub struct SimpleFrame {
|
||||
pub data: Vec<u8>,
|
||||
}
|
||||
|
||||
impl SimpleFrame {
|
||||
/// Magic header prepended to every serialised frame.
|
||||
pub const HEADER: &[u8; 7] = b"0FRAME0";
|
||||
|
||||
/// Serialise the frame: magic header followed by the payload bytes.
|
||||
pub fn to_bytes(&self) -> Vec<u8> {
|
||||
let mut bytes = Vec::new();
|
||||
|
||||
bytes.extend_from_slice(Self::HEADER);
|
||||
bytes.extend_from_slice(&self.data);
|
||||
|
||||
bytes
|
||||
}
|
||||
|
||||
/// Deserialise a [`SimpleFrame`] by stripping the leading magic header.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if `bytes` is shorter than the 7-byte header.
|
||||
pub fn try_from_bytes(bytes: &[u8]) -> anyhow::Result<Self> {
|
||||
if bytes.len() < Self::HEADER.len() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"Length mismatch to deserialize a SimpleFrame : Expected at least {}, got {}",
|
||||
Self::HEADER.len(),
|
||||
bytes.len()
|
||||
));
|
||||
}
|
||||
let data = bytes[Self::HEADER.len()..].to_vec();
|
||||
Ok(SimpleFrame { data })
|
||||
}
|
||||
}
|
||||
|
||||
/// Marker type identifying a fully-unwrapped simple payload.
|
||||
///
|
||||
/// Passed through the pipeline's [`FramingUnwrap`] stage; because the simple
|
||||
/// pipeline has only one message kind this is a zero-sized unit struct.
|
||||
///
|
||||
/// [`FramingUnwrap`]: nym_lp_data::common::traits::FramingUnwrap
|
||||
pub struct SimpleMessage;
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Building blocks
|
||||
|
||||
/// Wrapping building block: `SimpleFrame` → `SimplePacket`.
|
||||
///
|
||||
/// Implements [`Framing`], [`Transport`], and [`WireWrappingPipeline`] for the
|
||||
/// `SimpleFrame`/`SimplePacket` pair in one place. Compose this into any
|
||||
/// pipeline that needs wire-wrapping by delegating to `SimpleWireWrapper`.
|
||||
pub struct SimpleWireWrapper;
|
||||
|
||||
impl<Ts: Clone> Framing<Ts, NodeId> for SimpleWireWrapper {
|
||||
type Frame = SimpleFrame;
|
||||
const OVERHEAD_SIZE: usize = SimpleFrame::HEADER.len();
|
||||
fn to_frame(
|
||||
&self,
|
||||
payload: AddressedTimedPayload<Ts, NodeId>,
|
||||
frame_size: usize,
|
||||
) -> Vec<AddressedTimedData<Ts, SimpleFrame, NodeId>> {
|
||||
payload
|
||||
.data
|
||||
.data
|
||||
.chunks(frame_size)
|
||||
.map(|chunk| {
|
||||
AddressedTimedData::new(
|
||||
payload.data.timestamp.clone(),
|
||||
SimpleFrame {
|
||||
data: chunk.to_vec(),
|
||||
},
|
||||
payload.dst,
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// Transport wraps a [`SimpleFrame`] into a [`SimplePacket`].
|
||||
/// Overhead = 16 bytes (UUID), so effective payload = 48 bytes.
|
||||
impl<Ts: Clone> Transport<Ts, SimplePacket, NodeId> for SimpleWireWrapper {
|
||||
type Frame = SimpleFrame;
|
||||
const OVERHEAD_SIZE: usize = 16;
|
||||
fn to_transport_packet(
|
||||
&self,
|
||||
frame: AddressedTimedData<Ts, SimpleFrame, NodeId>,
|
||||
) -> AddressedTimedData<Ts, SimplePacket, NodeId> {
|
||||
// SAFETY: If the pipeline is implemented properly, frames perfectly fit in a packet
|
||||
#[allow(clippy::unwrap_used)]
|
||||
frame.data_transform(|inner| SimplePacket::new(inner.to_bytes().try_into().unwrap()))
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> WireWrappingPipeline<Ts, SimplePacket, NodeId> for SimpleWireWrapper {
|
||||
fn packet_size(&self) -> usize {
|
||||
SimplePacket::SIZE
|
||||
}
|
||||
}
|
||||
|
||||
/// Unwrapping building block: `SimplePacket` → payload.
|
||||
///
|
||||
/// Implements [`TransportUnwrap`], [`FramingUnwrap`], and
|
||||
/// [`WireUnwrappingPipeline`] for the `SimpleFrame`/`SimplePacket` pair.
|
||||
/// Compose into any pipeline that needs frame-unwrapping by delegating to
|
||||
/// `SimpleWireUnwrapper`.
|
||||
pub struct SimpleWireUnwrapper;
|
||||
|
||||
impl<Ts> FramingUnwrap<Ts, SimpleMessage> for SimpleWireUnwrapper {
|
||||
type Frame = SimpleFrame;
|
||||
fn frame_to_message(
|
||||
&mut self,
|
||||
frame: TimedData<Ts, SimpleFrame>,
|
||||
) -> Option<(TimedPayload<Ts>, SimpleMessage)> {
|
||||
Some((
|
||||
TimedPayload {
|
||||
data: frame.data.data,
|
||||
timestamp: frame.timestamp,
|
||||
},
|
||||
SimpleMessage,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> TransportUnwrap<Ts, SimplePacket> for SimpleWireUnwrapper {
|
||||
type Frame = SimpleFrame;
|
||||
fn packet_to_frame(
|
||||
&self,
|
||||
packet: SimplePacket,
|
||||
timestamp: Ts,
|
||||
) -> anyhow::Result<TimedData<Ts, SimpleFrame>> {
|
||||
// packet.data holds the framed bytes (HEADER + payload)
|
||||
Ok(TimedData::new(
|
||||
timestamp,
|
||||
SimpleFrame::try_from_bytes(&packet.data)?,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
impl<Ts: Clone> WireUnwrappingPipeline<Ts, SimplePacket, SimpleMessage> for SimpleWireUnwrapper {}
|
||||
@@ -0,0 +1,303 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use nym_common::debug::format_debug_bytes;
|
||||
use nym_sphinx::{Delay, Destination, DestinationAddressBytes, SphinxPacketBuilder};
|
||||
|
||||
use rand::Rng;
|
||||
use rand_distr::{Distribution, Exp};
|
||||
use std::{fmt::Debug, ops::Add, time::Duration};
|
||||
|
||||
use crate::{
|
||||
client::ClientId, node::NodeId, packet::WirePacketFormat, topology::directory::Directory,
|
||||
};
|
||||
|
||||
/// Newtype wrapper that provides a trimmed [`Debug`]
|
||||
/// implementation (showing only the first 32 bytes of the serialised form to
|
||||
/// avoid flooding logs).
|
||||
pub struct SimMixPacket(Vec<u8>);
|
||||
|
||||
impl Debug for SimMixPacket {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
writeln!(f, "SimMixPacket {{")?;
|
||||
writeln!(f, " data start:")?;
|
||||
if self.0.len() > 32 {
|
||||
for line in format_debug_bytes(&self.0.to_bytes()[..32])?.lines() {
|
||||
writeln!(f, " {line}")?;
|
||||
}
|
||||
} else {
|
||||
for line in format_debug_bytes(&self.0.to_bytes())?.lines() {
|
||||
writeln!(f, " {line}")?;
|
||||
}
|
||||
}
|
||||
write!(f, "}}")
|
||||
}
|
||||
}
|
||||
|
||||
impl WirePacketFormat for SimMixPacket {
|
||||
fn to_bytes(&self) -> Vec<u8> {
|
||||
self.0.clone()
|
||||
}
|
||||
|
||||
fn try_from_bytes(bytes: &[u8]) -> anyhow::Result<Self> {
|
||||
Ok(SimMixPacket(bytes.to_vec()))
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Vec<u8>> for SimMixPacket {
|
||||
fn from(value: Vec<u8>) -> Self {
|
||||
Self(value)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<SimMixPacket> for Vec<u8> {
|
||||
fn from(value: SimMixPacket) -> Self {
|
||||
value.0
|
||||
}
|
||||
}
|
||||
|
||||
/// A pre-built Sphinx packet that the recipient sends back as an acknowledgement.
|
||||
///
|
||||
/// A `SurbAck` bundles the serialised Sphinx packet together with the first-hop
|
||||
/// node ID and the expected total mix delay so that the sender can compute the
|
||||
/// latest time by which the ACK should arrive.
|
||||
#[derive(Debug)]
|
||||
pub struct SurbAck {
|
||||
surb_ack_packet: SimMixPacket,
|
||||
first_hop_id: NodeId,
|
||||
expected_total_delay: Delay,
|
||||
}
|
||||
|
||||
impl SurbAck {
|
||||
/// Magic bytes written at the start of every SURB ACK payload so that the
|
||||
/// final-hop node can identify them and route them separately.
|
||||
pub const MARKER: &[u8; 8] = b"SURB_ACK";
|
||||
const ACK_SIZE: usize = 8 + 8; // u64 ID and MARKER
|
||||
const PAYLOAD_SIZE: usize = Self::ACK_SIZE + nym_sphinx::PAYLOAD_OVERHEAD_SIZE;
|
||||
|
||||
/// Build a fresh SURB ACK addressed to `recipient` with unique `packet_id`.
|
||||
///
|
||||
/// Samples a 3-hop route from `directory`, draws per-hop Sphinx delays using
|
||||
/// `Ts::generate_mix_delay`, and constructs a Sphinx packet whose payload is
|
||||
/// `MARKER || packet_id.to_le_bytes()`.
|
||||
pub fn construct<Ts: GenerateDelay, R>(
|
||||
rng: &mut R,
|
||||
recipient: ClientId,
|
||||
packet_id: u64,
|
||||
directory: &Directory,
|
||||
) -> Self
|
||||
where
|
||||
R: Rng,
|
||||
{
|
||||
let route = directory
|
||||
.random_route(3, rng)
|
||||
.into_iter()
|
||||
.collect::<Vec<_>>();
|
||||
// SAFETY : We just sampled 3 nodes, the vec isn't empty
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let first_hop_id = route.first().unwrap().id;
|
||||
let sphinx_route = route.into_iter().map(Into::into).collect::<Vec<_>>();
|
||||
|
||||
let destination = Destination::new(
|
||||
DestinationAddressBytes::from_bytes([recipient; 32]),
|
||||
[recipient; 16],
|
||||
);
|
||||
|
||||
let delays = (0..sphinx_route.len())
|
||||
.map(|_| Delay::new_from_millis(Ts::generate_mix_delay(rng)))
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let ack_payload = Self::MARKER
|
||||
.iter()
|
||||
.copied()
|
||||
.chain(packet_id.to_le_bytes())
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let builder = SphinxPacketBuilder::new().with_payload_size(Self::PAYLOAD_SIZE);
|
||||
|
||||
// SAFETY : We're living in a simulation, if it crashes, it crashes
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let surb_ack_packet = builder
|
||||
.build_packet(ack_payload, &sphinx_route, &destination, &delays)
|
||||
.unwrap()
|
||||
.to_bytes();
|
||||
|
||||
// in our case, the last hop is a gateway that does NOT do any delays
|
||||
let expected_total_delay = delays.iter().take(delays.len() - 1).sum();
|
||||
|
||||
SurbAck {
|
||||
surb_ack_packet: surb_ack_packet.into(),
|
||||
first_hop_id,
|
||||
expected_total_delay,
|
||||
}
|
||||
}
|
||||
|
||||
/// Byte length of a serialised SURB ACK as prepended to outgoing payloads.
|
||||
///
|
||||
/// Format: `first_hop_id (1 byte) || sphinx_header || ack_payload`.
|
||||
pub const fn len() -> usize {
|
||||
Self::PAYLOAD_SIZE + nym_sphinx::HEADER_SIZE + 1 // SURB_FIRST_HOP || SURB_ACK
|
||||
}
|
||||
|
||||
/// Return the sum of per-hop delays embedded in the SURB packet header.
|
||||
///
|
||||
/// The terminal (gateway) hop is excluded because it applies no mix delay in
|
||||
/// the simulation.
|
||||
pub fn expected_total_delay(&self) -> Delay {
|
||||
self.expected_total_delay
|
||||
}
|
||||
|
||||
/// Serialise the SURB ACK into the wire format prepended to outgoing packets.
|
||||
///
|
||||
/// Returns `(total_delay, first_hop_id || sphinx_packet_bytes)`. The caller
|
||||
/// hands the byte vector to the reliability layer and the delay to the
|
||||
/// scheduler.
|
||||
pub fn prepare_for_sending(self) -> (Delay, Vec<u8>) {
|
||||
// SURB_FIRST_HOP || SURB_ACK
|
||||
let surb_bytes: Vec<_> = std::iter::once(self.first_hop_id)
|
||||
.chain(self.surb_ack_packet.to_bytes())
|
||||
.collect();
|
||||
(self.expected_total_delay, surb_bytes)
|
||||
}
|
||||
|
||||
/// Recover the first-hop node ID and the Sphinx ACK packet from the raw bytes
|
||||
/// produced by [`prepare_for_sending`].
|
||||
///
|
||||
/// This is the partial inverse of `prepare_for_sending`, performed by the
|
||||
/// gateway (final-hop node) when it dispatches the SURB back into the network.
|
||||
pub fn try_recover_first_hop_packet(b: &[u8]) -> anyhow::Result<(NodeId, SimMixPacket)> {
|
||||
let first_hop_id = b[0];
|
||||
let packet = SimMixPacket::try_from_bytes(&b[1..])?;
|
||||
|
||||
Ok((first_hop_id, packet))
|
||||
}
|
||||
|
||||
/// Split a final-hop plaintext into `(surb_ack_bytes, message_bytes)`.
|
||||
///
|
||||
/// If `extracted_data` is shorter than [`SurbAck::len`] (e.g. cover-traffic
|
||||
/// packets carry no SURB), the ACK slice is empty and the full buffer is
|
||||
/// returned as the message.
|
||||
pub fn extract_ack_and_message(mut extracted_data: Vec<u8>) -> (Vec<u8>, Vec<u8>) {
|
||||
let ack_len = SurbAck::len();
|
||||
|
||||
if extracted_data.len() < ack_len {
|
||||
// No SURB Ack in packet, in the sim this will be the case for cover traffic
|
||||
return (Vec::new(), extracted_data);
|
||||
}
|
||||
|
||||
let message = extracted_data.split_off(ack_len);
|
||||
let ack_data = extracted_data;
|
||||
(ack_data, message)
|
||||
}
|
||||
|
||||
/// Return `true` if `data` starts with the [`MARKER`](SurbAck::MARKER) bytes.
|
||||
pub fn is_surb_ack(data: &[u8]) -> bool {
|
||||
if data.len() < Self::MARKER.len() {
|
||||
return false;
|
||||
}
|
||||
|
||||
data[..Self::MARKER.len()] == *Self::MARKER
|
||||
}
|
||||
}
|
||||
|
||||
/// Marker type identifying a fully-unwrapped Sphinx payload.
|
||||
///
|
||||
/// Passed through the pipeline's [`FramingUnwrap`] stage so that
|
||||
/// [`ClientUnwrappingPipeline::process_unwrapped`] can dispatch on the message
|
||||
/// kind. In the simulation there is only one kind, so this is a zero-sized
|
||||
/// unit struct.
|
||||
///
|
||||
/// [`FramingUnwrap`]: nym_lp_data::common::traits::FramingUnwrap
|
||||
/// [`ClientUnwrappingPipeline::process_unwrapped`]: nym_lp_data::clients::traits::ClientUnwrappingPipeline::process_unwrapped
|
||||
#[derive(Default)]
|
||||
pub struct SphinxMessage;
|
||||
|
||||
/// Abstracts adding a Sphinx [`Delay`](nym_sphinx::Delay) to a timestamp type.
|
||||
pub trait AddDelay: Sized {
|
||||
fn add_delay(self, delay: nym_sphinx::Delay) -> Self;
|
||||
}
|
||||
|
||||
impl AddDelay for u32 {
|
||||
/// One tick = 1 ms.
|
||||
fn add_delay(self, delay: nym_sphinx::Delay) -> Self {
|
||||
self + (delay.to_nanos() / 1_000_000) as u32
|
||||
}
|
||||
}
|
||||
|
||||
impl AddDelay for std::time::Instant {
|
||||
fn add_delay(self, delay: nym_sphinx::Delay) -> Self {
|
||||
self + delay.to_duration()
|
||||
}
|
||||
}
|
||||
|
||||
/// Timestamp types that can generate Sphinx delays and be advanced by them.
|
||||
///
|
||||
/// Implemented for `u32` (discrete ticks, 1 tick = 1 ms) and [`Instant`]
|
||||
/// (wall-clock time).
|
||||
pub trait GenerateDelay: Sized + Add<Self::Delay, Output = Self> {
|
||||
/// The delay unit that can be added to `Self` (e.g. `u32` ticks or
|
||||
/// [`Duration`](std::time::Duration)).
|
||||
type Delay;
|
||||
|
||||
/// Draw a per-hop mix delay in milliseconds for inclusion in a Sphinx packet header.
|
||||
fn generate_mix_delay(rng: &mut impl Rng) -> u64;
|
||||
|
||||
/// Draw an inter-packet sending delay for the main Poisson loop.
|
||||
fn generate_sending_delay(rng: &mut impl Rng) -> Self::Delay;
|
||||
|
||||
/// Draw an inter-packet sending delay for the secondary cover traffic loop.
|
||||
fn generate_cover_traffic_delay(rng: &mut impl Rng) -> Self::Delay;
|
||||
}
|
||||
|
||||
impl GenerateDelay for u32 {
|
||||
type Delay = u32;
|
||||
|
||||
/// Uniform in `[0, 10]` ms.
|
||||
fn generate_mix_delay(rng: &mut impl Rng) -> u64 {
|
||||
rng.gen_range(0..=10)
|
||||
}
|
||||
|
||||
/// Exponential with mean 10 ticks (ms).
|
||||
fn generate_sending_delay(rng: &mut impl Rng) -> u32 {
|
||||
// SAFETY : hardcoded > 0 value
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let exp: Exp<f64> = Exp::new(1.0 / 10.0).unwrap();
|
||||
exp.sample(rng).round() as u32
|
||||
}
|
||||
|
||||
/// Exponential with mean 100 ticks (ms).
|
||||
fn generate_cover_traffic_delay(rng: &mut impl Rng) -> u32 {
|
||||
// SAFETY : hardcoded > 0 value
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let exp: Exp<f64> = Exp::new(1.0 / 100.0).unwrap();
|
||||
exp.sample(rng).round() as u32
|
||||
}
|
||||
}
|
||||
|
||||
impl GenerateDelay for std::time::Instant {
|
||||
type Delay = Duration;
|
||||
|
||||
/// Exponential with mean 50 ms.
|
||||
fn generate_mix_delay(rng: &mut impl Rng) -> u64 {
|
||||
// SAFETY : hardcoded > 0 value
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let exp: Exp<f64> = Exp::new(1.0 / 50.0).unwrap();
|
||||
exp.sample(rng).round() as u64
|
||||
}
|
||||
|
||||
/// Exponential with mean 20 ms.
|
||||
fn generate_sending_delay(rng: &mut impl Rng) -> Duration {
|
||||
// SAFETY : hardcoded > 0 value
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let exp: Exp<f64> = Exp::new(1.0 / 20.0).unwrap();
|
||||
Duration::from_millis(exp.sample(rng).round() as u64)
|
||||
}
|
||||
|
||||
/// Exponential with mean 200 ms.
|
||||
fn generate_cover_traffic_delay(rng: &mut impl Rng) -> Duration {
|
||||
// SAFETY : hardcoded > 0 value
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let exp: Exp<f64> = Exp::new(1.0 / 200.0).unwrap();
|
||||
Duration::from_millis(exp.sample(rng).round() as u64)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,139 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! In-memory network directory used by nodes to resolve [`NodeId`]s to socket
|
||||
//! addresses at send time.
|
||||
//!
|
||||
//! The [`Directory`] is built once during driver initialisation (after all UDP
|
||||
//! sockets have been bound) and then shared immutably across every [`Node`] via
|
||||
//! an [`Arc`]. This means routing lookups are lock-free and allocation-free after
|
||||
//! startup.
|
||||
|
||||
use std::{collections::HashMap, net::SocketAddr};
|
||||
|
||||
use nym_crypto::asymmetric::x25519;
|
||||
use nym_sphinx::{Node as SphinxNode, NodeAddressBytes};
|
||||
use rand::{prelude::SliceRandom, seq::IteratorRandom};
|
||||
|
||||
use crate::{
|
||||
client::ClientId,
|
||||
node::NodeId,
|
||||
topology::{Topology, TopologyNode},
|
||||
};
|
||||
|
||||
/// Shared, immutable routing table for the simulation.
|
||||
///
|
||||
/// Maps every [`NodeId`] that is part of the current topology to a
|
||||
/// [`DirectoryNode`] entry containing the node's configuration and reachable
|
||||
/// [`SocketAddr`].
|
||||
#[derive(Default, Debug)]
|
||||
pub struct Directory {
|
||||
/// Keyed routing map: node ID → directory entry.
|
||||
nodes: HashMap<NodeId, DirectoryNode>,
|
||||
/// Mix-network socket address for each client, keyed by [`ClientId`].
|
||||
///
|
||||
/// Used by nodes to deliver final-hop packets directly to the target client's
|
||||
/// mix socket rather than forwarding to another node.
|
||||
clients: HashMap<ClientId, SocketAddr>,
|
||||
}
|
||||
|
||||
impl Directory {
|
||||
/// Look up a node by its [`NodeId`].
|
||||
///
|
||||
/// Returns `None` when `id` is not present in the directory
|
||||
pub fn node(&self, id: NodeId) -> Option<&DirectoryNode> {
|
||||
self.nodes.get(&id)
|
||||
}
|
||||
|
||||
/// Look up a client by its [`ClientId`].
|
||||
///
|
||||
/// Returns `None` when `id` is not present in the directory
|
||||
pub fn client(&self, id: NodeId) -> Option<&SocketAddr> {
|
||||
self.clients.get(&id)
|
||||
}
|
||||
|
||||
/// Returns the node_id of every node in the network
|
||||
pub fn node_ids(&self) -> Vec<NodeId> {
|
||||
self.nodes.keys().copied().collect()
|
||||
}
|
||||
|
||||
/// Pick a random node
|
||||
pub fn random_next_hop(&self, rng: &mut impl rand::Rng) -> NodeId {
|
||||
// SAFETY: The directory always contains at least one node in a valid simulation.
|
||||
#[allow(clippy::unwrap_used)]
|
||||
*self.node_ids().choose(rng).unwrap()
|
||||
}
|
||||
|
||||
pub fn random_route(&self, length: usize, rng: &mut impl rand::Rng) -> Vec<DirectoryNode> {
|
||||
// SAFETY: The directory always contains at least one node in a valid simulation.
|
||||
#[allow(clippy::unwrap_used)]
|
||||
std::iter::repeat_with(|| *self.nodes.values().choose(rng).unwrap())
|
||||
.take(length)
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
impl From<&Topology> for Directory {
|
||||
/// Build a [`Directory`] from a full [`Topology`], extracting only the
|
||||
/// public routing information (addresses and public keys) from each entry.
|
||||
fn from(value: &Topology) -> Self {
|
||||
let mut directory = Directory::default();
|
||||
for node in &value.nodes {
|
||||
directory.nodes.insert(node.node_id, node.into());
|
||||
}
|
||||
for client in &value.clients {
|
||||
directory
|
||||
.clients
|
||||
.insert(client.client_id, client.mixnet_address);
|
||||
}
|
||||
directory
|
||||
}
|
||||
}
|
||||
|
||||
/// Public routing information for a single mix node, stored in the [`Directory`].
|
||||
#[derive(Copy, Clone, Debug)]
|
||||
pub struct DirectoryNode {
|
||||
/// Unique identifier for this node within the topology.
|
||||
///
|
||||
/// Used as the key in the [`Directory`] when resolving routing targets.
|
||||
pub id: NodeId,
|
||||
|
||||
/// UDP socket address on which this node listens for incoming packets.
|
||||
pub addr: SocketAddr,
|
||||
|
||||
/// Sphinx (X25519) public key used to encrypt packets destined for this node.
|
||||
pub sphinx_public_key: x25519::PublicKey,
|
||||
}
|
||||
|
||||
impl From<&TopologyNode> for DirectoryNode {
|
||||
/// Derive the public [`DirectoryNode`] entry from a [`TopologyNode`] by
|
||||
/// computing the corresponding X25519 public key from the private key.
|
||||
fn from(value: &TopologyNode) -> Self {
|
||||
DirectoryNode {
|
||||
id: value.node_id,
|
||||
addr: value.socket_address,
|
||||
sphinx_public_key: x25519::PublicKey::from(&value.sphinx_private_key),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<&DirectoryNode> for SphinxNode {
|
||||
/// Convert a [`DirectoryNode`] into a [`SphinxNode`] suitable for use with
|
||||
/// [`SphinxPacketBuilder`].
|
||||
///
|
||||
/// The Sphinx [`NodeAddressBytes`] are constructed by repeating the single-byte
|
||||
/// [`NodeId`] across all 32 bytes — a simulation-only convention that lets the
|
||||
/// node recover its own ID from the address after decryption.
|
||||
///
|
||||
/// [`SphinxPacketBuilder`]: nym_sphinx::SphinxPacketBuilder
|
||||
fn from(value: &DirectoryNode) -> Self {
|
||||
let address = NodeAddressBytes::from_bytes([value.id; 32]);
|
||||
SphinxNode::new(address, *value.sphinx_public_key)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<DirectoryNode> for SphinxNode {
|
||||
fn from(value: DirectoryNode) -> Self {
|
||||
(&value).into()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
//! Topology file types and the in-memory network directory.
|
||||
//!
|
||||
//! The topology is loaded from `topology.json` and contains everything needed
|
||||
//! to construct a node or client (including private config such as keys).
|
||||
//! The [`directory::Directory`] holds only the public-facing routing information
|
||||
//! visible to other participants in the network.
|
||||
|
||||
use std::net::SocketAddr;
|
||||
|
||||
use nym_crypto::asymmetric::x25519;
|
||||
use nym_crypto::asymmetric::x25519::serde_helpers::bs58_x25519_private_key;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::{client::ClientId, node::NodeId};
|
||||
|
||||
pub mod directory;
|
||||
|
||||
/// Per-node configuration stored in `topology.json`.
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct TopologyNode {
|
||||
pub node_id: NodeId,
|
||||
/// UDP address on which the node listens for incoming packets.
|
||||
pub socket_address: SocketAddr,
|
||||
/// Notional reliability percentage (0–100); reserved for future use.
|
||||
pub reliability: u8,
|
||||
/// Sphinx (X25519) private key used by this node to unwrap packets.
|
||||
#[serde(with = "bs58_x25519_private_key")]
|
||||
pub sphinx_private_key: x25519::PrivateKey,
|
||||
}
|
||||
|
||||
impl TopologyNode {
|
||||
/// Construct a [`TopologyNode`] with a freshly generated Sphinx keypair.
|
||||
///
|
||||
/// Intended for use by `init-topology` to generate a topology file for the
|
||||
/// simulation.
|
||||
pub fn new(node_id: NodeId, reliability: u8, socket_address: SocketAddr) -> Self {
|
||||
let sphinx_private_key = x25519::PrivateKey::new(&mut rand::thread_rng());
|
||||
Self {
|
||||
node_id,
|
||||
socket_address,
|
||||
reliability,
|
||||
sphinx_private_key,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-client configuration stored in `topology.json`.
|
||||
#[derive(Serialize, Deserialize, Clone, Debug)]
|
||||
pub struct TopologyClient {
|
||||
/// Unique identifier for this client within the topology.
|
||||
pub client_id: ClientId,
|
||||
/// UDP address the client uses to talk to the mix network.
|
||||
pub mixnet_address: SocketAddr,
|
||||
/// UDP address where the client listens for messages from user applications
|
||||
/// (e.g. the standalone `client` binary). Not included in the [`Directory`].
|
||||
pub app_address: SocketAddr,
|
||||
}
|
||||
|
||||
impl TopologyClient {
|
||||
/// Construct a [`TopologyClient`] with the given addresses.
|
||||
///
|
||||
/// Intended for use by `init-topology` to generate a topology file for the
|
||||
/// simulation.
|
||||
pub fn new(client_id: ClientId, mixnet_address: SocketAddr, app_address: SocketAddr) -> Self {
|
||||
Self {
|
||||
client_id,
|
||||
mixnet_address,
|
||||
app_address,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Root topology file structure, deserialised from `topology.json`.
|
||||
#[derive(Serialize, Deserialize)]
|
||||
pub struct Topology {
|
||||
pub nodes: Vec<TopologyNode>,
|
||||
pub clients: Vec<TopologyClient>,
|
||||
}
|
||||
Reference in New Issue
Block a user