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

add trace_id to sphinx packet

Browse Source
This commit is contained in:
Floriane TUERNAL SABOTINOV
2025-10-07 16:57:02 +02:00
parent afb2467afc
commit 6ae78d9f4d
24 changed files with 126 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+5
View File
@@ -6851,6 +6851,7 @@ dependencies = [
name = "nym-sphinx"
version = "0.1.0"
dependencies = [
"nym-bin-common",
"nym-crypto",
"nym-metrics",
"nym-mixnet-contract-common",
@@ -6868,6 +6869,7 @@ dependencies = [
"rand 0.8.5",
"rand_chacha 0.3.1",
"rand_distr",
"sphinx-packet",
"thiserror 2.0.16",
"tokio",
"tracing",
@@ -6972,6 +6974,7 @@ name = "nym-sphinx-framing"
version = "0.1.0"
dependencies = [
"bytes",
"nym-bin-common",
"nym-sphinx-acknowledgements",
"nym-sphinx-addressing",
"nym-sphinx-forwarding",
@@ -7006,6 +7009,7 @@ dependencies = [
name = "nym-sphinx-types"
version = "0.2.0"
dependencies = [
"nym-bin-common",
"nym-outfox",
"sphinx-packet",
"thiserror 2.0.16",
@@ -7172,6 +7176,7 @@ dependencies = [
"hmac",
"itertools 0.14.0",
"log",
"nym-bin-common",
"nym-config",
"nym-crypto",
"nym-mixnet-contract-common",
clients/native/src/websocket/handler.rs
+2 -2
View File
@@ -184,7 +184,7 @@ impl Handler {
});
// the ack control is now responsible for chunking, etc.
let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type);
let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type, None);
if let Err(err) = self.msg_input.send(input_msg).await {
if !self.shutdown_token.is_cancelled() {
error!("Failed to send message to the input buffer: {err}");
@@ -217,7 +217,7 @@ impl Handler {
});
let input_msg =
InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type, None);
if let Err(err) = self.msg_input.send(input_msg).await {
if !self.shutdown_token.is_cancelled() {
error!("Failed to send anonymous message to the input buffer: {err}");
common/bin-common/src/opentelemetry/compact_id_generator.rs
+8 -7
View File
@@ -11,8 +11,9 @@ impl IdGenerator for Compact13BytesIdGenerator {
let mut bytes = [0u8; 16];
// Fill the first 13 bytes with random data
rng.fill_bytes(&mut bytes[0..13]);
// Set the last 3 bytes to zero
rng.fill_bytes(&mut bytes[0..12]);
// Set the last 4 bytes to zero
bytes[12] = 0;
bytes[13] = 0;
bytes[14] = 0;
bytes[15] = 0;
@@ -29,18 +30,18 @@ impl IdGenerator for Compact13BytesIdGenerator {
}
}
pub fn compress_trace_id(trace_id: &TraceId) -> [u8; 13] {
pub fn compress_trace_id(trace_id: &TraceId) -> [u8; 12] {
let bytes = trace_id.to_bytes();
let mut compressed = [0u8; 13];
compressed.copy_from_slice(&bytes[0..13]);
let mut compressed = [0u8; 12];
compressed.copy_from_slice(&bytes[0..12]);
compressed
}
pub fn decompress_trace_id(compressed: &[u8; 13]) -> TraceId {
pub fn decompress_trace_id(compressed: &[u8; 12]) -> TraceId {
let mut bytes = [0u8; 16];
bytes[0..13].copy_from_slice(compressed);
bytes[0..12].copy_from_slice(compressed);
TraceId::from_bytes(bytes)
}
common/bin-common/src/opentelemetry/mod.rs
+1 -1
View File
@@ -1,6 +1,6 @@
pub mod context;
pub mod error;
mod compact_id_generator;
pub mod compact_id_generator;
mod trace_id_format;
use tracing::{info, Level};
common/client-core/src/client/inbound_messages.rs
+16
View File
@@ -29,6 +29,8 @@ pub enum InputMessage {
data: Vec<u8>,
lane: TransmissionLane,
max_retransmissions: Option<u32>,
// add trace_id for optional tracing of individual messages in debug mode
trace_id: Option<[u8; 12]>,
},
/// Creates a message used for a duplex anonymous communication where the recipient
@@ -45,6 +47,7 @@ pub enum InputMessage {
reply_surbs: u32,
lane: TransmissionLane,
max_retransmissions: Option<u32>,
trace_id: Option<[u8; 12]>,
},
/// Attempt to use our internally received and stored `ReplySurb` to send the message back
@@ -90,12 +93,14 @@ impl InputMessage {
data: Vec<u8>,
lane: TransmissionLane,
packet_type: Option<PacketType>,
trace_id: Option<[u8; 12]>,
) -> Self {
let message = InputMessage::Regular {
recipient,
data,
lane,
max_retransmissions: None,
trace_id,
};
if let Some(packet_type) = packet_type {
InputMessage::new_wrapper(message, packet_type)
@@ -110,6 +115,7 @@ impl InputMessage {
reply_surbs: u32,
lane: TransmissionLane,
packet_type: Option<PacketType>,
trace_id: Option<[u8; 12]>,
) -> Self {
let message = InputMessage::Anonymous {
recipient,
@@ -117,6 +123,7 @@ impl InputMessage {
reply_surbs,
lane,
max_retransmissions: None,
trace_id,
};
if let Some(packet_type) = packet_type {
InputMessage::new_wrapper(message, packet_type)
@@ -185,4 +192,13 @@ impl InputMessage {
self.set_max_retransmissions(max_retransmissions);
self
}
pub fn trace_id(&self) -> Option<[u8; 12]> {
match self {
InputMessage::Regular { trace_id, .. } => *trace_id,
InputMessage::Anonymous { trace_id, .. } => *trace_id,
InputMessage::Premade { .. } | InputMessage::Reply { .. } => None,
InputMessage::MessageWrapper { message, .. } => message.trace_id(),
}
}
}
common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs
+10 -1
View File
@@ -77,10 +77,11 @@ where
lane: TransmissionLane,
packet_type: PacketType,
max_retransmissions: Option<u32>,
trace_id: Option<[u8; 12]>,
) {
if let Err(err) = self
.message_handler
.try_send_plain_message(recipient, content, lane, packet_type, max_retransmissions)
.try_send_plain_message(recipient, content, lane, packet_type, max_retransmissions, trace_id)
.await
{
warn!("failed to send a plain message - {err}")
@@ -114,12 +115,15 @@ where
#[allow(clippy::panic)]
async fn on_input_message(&mut self, msg: InputMessage) {
let trace_id = msg.trace_id();
match msg {
InputMessage::Regular {
recipient,
data,
lane,
max_retransmissions,
..
} => {
self.handle_plain_message(
recipient,
@@ -127,6 +131,7 @@ where
lane,
PacketType::Mix,
max_retransmissions,
trace_id
)
.await
}
@@ -136,6 +141,7 @@ where
reply_surbs,
lane,
max_retransmissions,
..
} => {
self.handle_repliable_message(
recipient,
@@ -166,6 +172,7 @@ where
data,
lane,
max_retransmissions,
..
} => {
self.handle_plain_message(
recipient,
@@ -173,6 +180,7 @@ where
lane,
packet_type,
max_retransmissions,
trace_id
)
.await
}
@@ -182,6 +190,7 @@ where
reply_surbs,
lane,
max_retransmissions,
..
} => {
self.handle_repliable_message(
recipient,
common/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs
+1 -1
View File
@@ -60,7 +60,7 @@ where
// TODO: Figure out retransmission packet type signaling
self.message_handler
.try_prepare_single_chunk_for_sending(packet_recipient, chunk_data, packet_type)
.try_prepare_single_chunk_for_sending(packet_recipient, chunk_data, packet_type, None)
.await
}
common/client-core/src/client/real_messages_control/message_handler.rs
+8
View File
@@ -483,6 +483,7 @@ where
lane: TransmissionLane,
packet_type: PacketType,
max_retransmissions: Option<u32>,
trace_id: Option<[u8; 12]>,
) -> Result<(), PreparationError> {
let message = NymMessage::new_plain(message);
self.try_split_and_send_non_reply_message(
@@ -491,6 +492,7 @@ where
lane,
packet_type,
max_retransmissions,
trace_id,
)
.await
}
@@ -502,6 +504,7 @@ where
lane: TransmissionLane,
packet_type: PacketType,
max_retransmissions: Option<u32>,
trace_id: Option<[u8; 12]>,
) -> Result<(), PreparationError> {
debug!("Sending non-reply message with packet type {packet_type}");
// TODO: I really dislike existence of this assertion, it implies code has to be re-organised
@@ -534,6 +537,7 @@ where
&self.config.ack_key,
&recipient,
packet_type,
trace_id
)?;
let real_message = RealMessage::new(
@@ -585,6 +589,7 @@ where
TransmissionLane::AdditionalReplySurbs,
packet_type,
max_retransmissions,
None,
)
.await?;
@@ -625,6 +630,7 @@ where
lane,
packet_type,
max_retransmissions,
None,
)
.await?;
@@ -639,6 +645,7 @@ where
recipient: Recipient,
chunk: Fragment,
packet_type: PacketType,
trace_id: Option<[u8; 12]>,
) -> Result<PreparedFragment, PreparationError> {
debug!("Sending single chunk with packet type {packet_type}");
let topology_permit = self.topology_access.get_read_permit().await;
@@ -650,6 +657,7 @@ where
&self.config.ack_key,
&recipient,
packet_type,
trace_id,
)?;
Ok(prepared_fragment)
common/client-core/src/client/statistics_control.rs
+1
View File
@@ -80,6 +80,7 @@ impl StatisticsControl {
stats_report.into(),
TransmissionLane::General,
None,
None,
);
if let Err(err) = self.report_tx.send(report_message).await {
tracing::error!("Failed to report client stats: {err:?}");
common/node-tester-utils/src/tester.rs
+1
View File
@@ -231,6 +231,7 @@ where
&address,
&address,
PacketType::Mix,
None
)?)
}
common/nymsphinx/Cargo.toml
+2
View File
@@ -8,12 +8,14 @@ license = { workspace = true }
repository = { workspace = true }
[dependencies]
sphinx-packet = { workspace = true }
tracing = { workspace = true }
rand = { workspace = true }
rand_distr = { workspace = true }
rand_chacha = { workspace = true }
thiserror = { workspace = true }
nym-bin-common = { path = "../bin-common" }
nym-sphinx-acknowledgements = { path = "acknowledgements" }
nym-sphinx-addressing = { path = "addressing" }
nym-sphinx-anonymous-replies = { path = "anonymous-replies" }
common/nymsphinx/framing/Cargo.toml
+1
View File
@@ -13,6 +13,7 @@ tokio-util = { workspace = true, features = ["codec"] }
thiserror = { workspace = true }
tracing = { workspace = true }
nym-bin-common = { path = "../../bin-common" }
nym-sphinx-types = { path = "../types", features = ["sphinx", "outfox"] }
nym-sphinx-params = { path = "../params", features = ["sphinx", "outfox"] }
nym-sphinx-forwarding = { path = "../forwarding" }
common/nymsphinx/framing/src/processing.rs
+22 -8
View File
@@ -2,6 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
use crate::packet::FramedNymPacket;
use nym_bin_common::opentelemetry::compact_id_generator::decompress_trace_id;
use nym_sphinx_acknowledgements::surb_ack::{SurbAck, SurbAckRecoveryError};
use nym_sphinx_addressing::nodes::{NymNodeRoutingAddress, NymNodeRoutingAddressError};
use nym_sphinx_forwarding::packet::MixPacket;
@@ -258,15 +259,28 @@ fn wrap_processed_sphinx_packet(
// sphinx all together?
ProcessedPacketData::FinalHop {
destination,
identifier: _,
identifier,
payload,
} => process_final_hop(
destination,
payload.recover_plaintext()?,
packet_size,
packet_type,
key_rotation,
),
} => {
// if we have a trace id in the destination, we log it for easier correlation later on
// TODO add feature for otel
{
let trace_bytes: [u8; 12] = identifier[0..12].try_into().unwrap();
if !trace_bytes.iter().all(|b| *b == 0) {
let full_trace_id = decompress_trace_id(&trace_bytes);
tracing::Span::current().record("trace_id", &full_trace_id.to_string().as_str());
info!("Processing final hop with trace_id: {full_trace_id:?}");
}
}
process_final_hop(
destination,
payload.recover_plaintext()?,
packet_size,
packet_type,
key_rotation,
)
}
}?;
Ok(MixProcessingResult {
common/nymsphinx/src/preparer/mod.rs
+22 -18
View File
@@ -11,7 +11,8 @@ use nym_sphinx_addressing::clients::Recipient;
use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
use nym_sphinx_anonymous_replies::reply_surb::ReplySurb;
use nym_sphinx_chunking::fragment::{Fragment, FragmentIdentifier};
use nym_sphinx_forwarding::packet::MixPacket;
use nym_sphinx_forwarding::packet::{self, MixPacket};
use sphinx_packet::route::Destination;
use nym_sphinx_params::packet_sizes::PacketSize;
use nym_sphinx_params::{PacketType, ReplySurbKeyDigestAlgorithm, SphinxKeyRotation};
use nym_sphinx_types::{Delay, NymPacket};
@@ -163,19 +164,6 @@ pub trait FragmentPreparer {
})
}
/// Tries to convert this [`Fragment`] into a [`SphinxPacket`] that can be sent through the Nym mix-network,
/// such that it contains required SURB-ACK and public component of the ephemeral key used to
/// derive the shared key.
/// Also all the data, apart from the said public component, is encrypted with an ephemeral shared key.
/// This method can fail if the provided network topology is invalid.
/// It returns total expected delay as well as the [`SphinxPacket`] (including first hop address)
/// to be sent through the network.
///
/// The procedure is as follows:
/// For each fragment:
/// - compute SURB_ACK
/// - generate (x, g^x)
/// - compute k = KDF(remote encryption key ^ x) this is equivalent to KDF( dh(remote, x) )
/// - compute v_b = AES-128-CTR(k, serialized_fragment)
/// - compute vk_b = g^x || v_b
/// - compute sphinx_plaintext = SURB_ACK || g^x || v_b
@@ -189,6 +177,7 @@ pub trait FragmentPreparer {
packet_sender: &Recipient,
packet_recipient: &Recipient,
packet_type: PacketType,
trace_id: Option<[u8; 12]>,
) -> Result<PreparedFragment, NymTopologyError> {
debug!("Preparing chunk for sending");
// each plain or repliable packet (i.e. not a reply) attaches an ephemeral public key so that the recipient
@@ -249,7 +238,8 @@ pub trait FragmentPreparer {
topology.random_route_to_egress(&mut rng, destination)?
};
let destination = packet_recipient.as_sphinx_destination();
let mut destination = packet_recipient.as_sphinx_destination();
add_trace_id_to_destination(&mut destination, trace_id);
// including set of delays
let delays =
@@ -274,9 +264,21 @@ pub trait FragmentPreparer {
)?,
};
// from the previously constructed route extract the first hop
let first_hop_address =
NymNodeRoutingAddress::try_from(route.first().unwrap().address).unwrap();
/// If a trace id is provided, add it to the first 12 bytes of the destination identifier.
/// The remaining 4 bytes of the identifier are left unchanged.
fn add_trace_id_to_destination(
destination: &mut Destination,
trace_id: Option<[u8; 12]>,
) {
if let Some(trace_id) = trace_id {
destination.identifier[0..12].copy_from_slice(&trace_id);
}
}
/// - compute k = KDF(remote encryption key ^ x) this is equivalent to KDF( dh(remote, x) )
/// from the previously constructed route extract the first hop
let first_hop_address =
NymNodeRoutingAddress::try_from(route.first().unwrap().address).unwrap();
Ok(PreparedFragment {
// the round-trip delay is the sum of delays of all hops on the forward route as
@@ -428,6 +430,7 @@ where
ack_key: &AckKey,
packet_recipient: &Recipient,
packet_type: PacketType,
trace_id: Option<[u8; 12]>,
) -> Result<PreparedFragment, NymTopologyError> {
let sender = self.sender_address;
@@ -439,6 +442,7 @@ where
&sender,
packet_recipient,
packet_type,
trace_id,
)
}
common/nymsphinx/types/Cargo.toml
+2
View File
@@ -10,6 +10,8 @@ repository = { workspace = true }
[dependencies]
sphinx-packet = { workspace = true, optional = true }
nym-outfox = { path = "../../../nym-outfox", optional = true }
# TODO add optional
nym-bin-common = { path = "../../../common/bin-common" }
thiserror = { workspace = true }
[features]
common/socks5-client-core/src/socks/client.rs
+4
View File
@@ -350,6 +350,7 @@ impl SocksClient {
self.config.connection_start_surbs,
TransmissionLane::ConnectionId(self.connection_id),
self.packet_type,
None
);
self.input_sender
.send(input_message)
@@ -373,6 +374,7 @@ impl SocksClient {
msg.into_bytes(),
TransmissionLane::ConnectionId(self.connection_id),
self.packet_type,
None
);
self.input_sender
.send(input_message)
@@ -439,6 +441,7 @@ impl SocksClient {
per_request_surbs,
lane,
packet_type,
None
)
} else {
InputMessage::new_regular(
@@ -446,6 +449,7 @@ impl SocksClient {
provider_message.into_bytes(),
lane,
packet_type,
None
)
}
})
common/types/Cargo.toml
+1
View File
@@ -29,6 +29,7 @@ x25519-dalek = { workspace = true, features = ["static_secrets"] }
cosmwasm-std = { workspace = true }
cosmrs = { workspace = true }
nym-bin-common = { path = "../../common/bin-common" }
nym-validator-client = { path = "../../common/client-libs/validator-client" }
nym-mixnet-contract-common = { path = "../../common/cosmwasm-smart-contracts/mixnet-contract" }
nym-vesting-contract-common = { path = "../../common/cosmwasm-smart-contracts/vesting-contract" }
gateway/src/node/internal_service_providers/authenticator/mixnet_listener.rs
+1
View File
@@ -979,6 +979,7 @@ fn create_input_message(
response_packet,
lane,
packet_type,
None,
))
} else {
tracing::error!("No nym-address or sender tag provided");
nym-authenticator-client/src/lib.rs
+2
View File
@@ -1221,12 +1221,14 @@ fn create_input_message(
surbs,
TransmissionLane::General,
None,
None
),
IncludedSurbs::ExposeSelfAddress => nym_sdk::mixnet::InputMessage::new_regular(
recipient,
data,
TransmissionLane::General,
None,
None
),
}
}
nym-ip-packet-client/src/connect.rs
+1
View File
@@ -195,5 +195,6 @@ fn create_input_message(
surbs,
TransmissionLane::General,
None,
None,
)
}
sdk/rust/nym-sdk/src/mixnet/sink.rs
+2
View File
@@ -59,6 +59,7 @@ impl MixnetMessageSinkTranslator for DefaultMixnetMessageSinkTranslator {
bytes,
self.lane,
self.packet_type,
None
)),
IncludedSurbs::Amount(surbs) => Ok(InputMessage::new_anonymous(
**recipient,
@@ -66,6 +67,7 @@ impl MixnetMessageSinkTranslator for DefaultMixnetMessageSinkTranslator {
*surbs,
self.lane,
self.packet_type,
None
)),
},
SinkDestination::Reply(tag) => Ok(InputMessage::new_reply(
sdk/rust/nym-sdk/src/mixnet/traits.rs
+11
View File
@@ -5,8 +5,11 @@ use crate::mixnet::{AnonymousSenderTag, IncludedSurbs, Recipient};
use crate::Result;
use async_trait::async_trait;
use nym_client_core::client::inbound_messages::InputMessage;
use nym_bin_common::opentelemetry::compact_id_generator::compress_trace_id;
use nym_sphinx::params::PacketType;
use nym_task::connections::TransmissionLane;
use opentelemetry::trace::TraceContextExt;
use opentelemetry::Context;
use tracing::instrument;
// defined to guarantee common interface regardless of whether you're using the full client
@@ -72,6 +75,12 @@ pub trait MixnetMessageSender {
where
M: AsRef<[u8]> + Send,
{
// In case of surb opentelemetry tracing, we extract the context and trace_id
// in the 12 bytes format
let context = Context::current();
let trace_id = context.span().span_context().trace_id();
let trace_id = compress_trace_id(&trace_id);
let lane = TransmissionLane::General;
let input_msg = match surbs {
IncludedSurbs::Amount(surbs) => InputMessage::new_anonymous(
@@ -80,12 +89,14 @@ pub trait MixnetMessageSender {
surbs,
lane,
self.packet_type(),
Some(trace_id)
),
IncludedSurbs::ExposeSelfAddress => InputMessage::new_regular(
address,
message.as_ref().to_vec(),
lane,
self.packet_type(),
Some(trace_id)
),
};
self.send(input_msg).await
service-providers/ip-packet-router/src/util/create_message.rs
+1 -1
View File
@@ -11,7 +11,7 @@ pub(crate) fn create_input_message(
let packet_type = None;
match recipient {
ConnectedClientId::NymAddress(recipient) => {
InputMessage::new_regular(**recipient, response_packet, lane, packet_type)
InputMessage::new_regular(**recipient, response_packet, lane, packet_type, None)
}
ConnectedClientId::AnonymousSenderTag(tag) => {
InputMessage::new_reply(*tag, response_packet, lane, packet_type)
service-providers/network-requester/src/reply.rs
+1
View File
@@ -192,6 +192,7 @@ impl MixnetAddress {
data: message,
lane: TransmissionLane::ConnectionId(connection_id),
max_retransmissions: None,
trace_id: None,
}),
packet_type,
},