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

wip: choosing correct key for packet processing

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2025-04-15 15:14:45 +01:00
parent 30163a0174
commit 85b51f21e5
9 changed files with 300 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nymsphinx/framing/src/codec.rs
+69 -63
View File
@@ -3,6 +3,7 @@
use crate::packet::{FramedNymPacket, Header};
use bytes::{Buf, BufMut, BytesMut};
use nym_sphinx_params::key_rotation::InvalidSphinxKeyRotation;
use nym_sphinx_params::packet_sizes::{InvalidPacketSize, PacketSize};
use nym_sphinx_params::packet_types::InvalidPacketType;
use nym_sphinx_params::packet_version::{InvalidPacketVersion, PacketVersion};
@@ -23,6 +24,9 @@ pub enum NymCodecError {
#[error("the packet version information was malformed: {0}")]
InvalidPacketVersion(#[from] InvalidPacketVersion),
#[error("the sphinx key rotation information was malformed: {0}")]
InvalidSphinxKeyRotation(#[from] InvalidSphinxKeyRotation),
#[error("received unsupported packet version {received}. max supported is {max_supported}")]
UnsupportedPacketVersion {
received: PacketVersion,
@@ -65,72 +69,74 @@ impl Decoder for NymCodec {
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
if src.is_empty() {
// can't do anything if we have no bytes, but let's reserve enough for the most
// conservative case, i.e. receiving an ack packet
src.reserve(Header::SIZE + PacketSize::AckPacket.size());
// conservative case, i.e. receiving a legacy ack packet
src.reserve(Header::INITIAL_SIZE + PacketSize::AckPacket.size());
return Ok(None);
}
// because header is so small and simple it makes no point in trying to cache
// this result. It will be just simpler to re-decode it
let header = match Header::decode(src)? {
Some(header) => header,
None => return Ok(None), // we have some data but not enough to get header back
};
let packet_size = header.packet_size.size();
let frame_len = Header::SIZE + packet_size;
if src.len() < frame_len {
// we don't have enough bytes to read the rest of frame
src.reserve(packet_size);
return Ok(None);
}
// advance buffer past the header - at this point we have enough bytes
src.advance(Header::SIZE);
let packet_bytes = src.split_to(packet_size);
let packet = if let Some(slice) = packet_bytes.get(..) {
// here it could be debatable whether stream is corrupt or not,
// but let's go with the safer approach and assume it is.
match header.packet_type {
PacketType::Outfox => NymPacket::outfox_from_bytes(slice)?,
PacketType::Mix => NymPacket::sphinx_from_bytes(slice)?,
}
} else {
return Ok(None);
};
// let packet = SphinxPacket::from_bytes(&sphinx_packet_bytes)?;
let nymsphinx_packet = FramedNymPacket { header, packet };
// As per docs:
// Before returning from the function, implementations should ensure that the buffer
// has appropriate capacity in anticipation of future calls to decode.
// Failing to do so leads to inefficiency.
// if we have enough bytes to decode the header of the next packet, we can reserve enough bytes for
// the entire next frame, if not, we assume the next frame is an ack packet and
// reserve for that.
// we also assume the next packet coming from the same client will use exactly the same versioning
// as the current packet
let mut allocate_for_next_packet = Header::SIZE + PacketSize::AckPacket.size();
if !src.is_empty() {
match Header::decode(src) {
Ok(Some(next_header)) => {
allocate_for_next_packet = Header::SIZE + next_header.packet_size.size();
}
Ok(None) => {
// we don't have enough information to know how much to reserve, fallback to the ack case
}
// the next frame will be malformed but let's leave handling the error to the next
// call to 'decode', as presumably, the current sphinx packet is still valid
Err(_) => return Ok(Some(nymsphinx_packet)),
};
}
src.reserve(allocate_for_next_packet);
Ok(Some(nymsphinx_packet))
todo!()
//
// // because header is so small and simple it makes no point in trying to cache
// // this result. It will be just simpler to re-decode it
// let header = match Header::decode(src)? {
// Some(header) => header,
// None => return Ok(None), // we have some data but not enough to get header back
// };
//
// let packet_size = header.packet_size.size();
// let frame_len = Header::SIZE + packet_size;
//
// if src.len() < frame_len {
// // we don't have enough bytes to read the rest of frame
// src.reserve(packet_size);
// return Ok(None);
// }
//
// // advance buffer past the header - at this point we have enough bytes
// src.advance(Header::SIZE);
// let packet_bytes = src.split_to(packet_size);
// let packet = if let Some(slice) = packet_bytes.get(..) {
// // here it could be debatable whether stream is corrupt or not,
// // but let's go with the safer approach and assume it is.
// match header.packet_type {
// PacketType::Outfox => NymPacket::outfox_from_bytes(slice)?,
// PacketType::Mix => NymPacket::sphinx_from_bytes(slice)?,
// }
// } else {
// return Ok(None);
// };
//
// // let packet = SphinxPacket::from_bytes(&sphinx_packet_bytes)?;
// let nymsphinx_packet = FramedNymPacket { header, packet };
//
// // As per docs:
// // Before returning from the function, implementations should ensure that the buffer
// // has appropriate capacity in anticipation of future calls to decode.
// // Failing to do so leads to inefficiency.
//
// // if we have enough bytes to decode the header of the next packet, we can reserve enough bytes for
// // the entire next frame, if not, we assume the next frame is an ack packet and
// // reserve for that.
// // we also assume the next packet coming from the same client will use exactly the same versioning
// // as the current packet
//
// let mut allocate_for_next_packet = Header::SIZE + PacketSize::AckPacket.size();
// if !src.is_empty() {
// match Header::decode(src) {
// Ok(Some(next_header)) => {
// allocate_for_next_packet = Header::SIZE + next_header.packet_size.size();
// }
// Ok(None) => {
// // we don't have enough information to know how much to reserve, fallback to the ack case
// }
//
// // the next frame will be malformed but let's leave handling the error to the next
// // call to 'decode', as presumably, the current sphinx packet is still valid
// Err(_) => return Ok(Some(nymsphinx_packet)),
// };
// }
// src.reserve(allocate_for_next_packet);
// Ok(Some(nymsphinx_packet))
}
}
common/nymsphinx/framing/src/packet.rs
+39 -17
View File
@@ -3,6 +3,7 @@
use crate::codec::NymCodecError;
use bytes::{BufMut, BytesMut};
use nym_sphinx_params::key_rotation::SphinxKeyRotation;
use nym_sphinx_params::packet_sizes::PacketSize;
use nym_sphinx_params::packet_version::{PacketVersion, CURRENT_PACKET_VERSION};
use nym_sphinx_params::PacketType;
@@ -18,7 +19,11 @@ pub struct FramedNymPacket {
}
impl FramedNymPacket {
pub fn new(packet: NymPacket, packet_type: PacketType) -> Self {
pub fn new(
packet: NymPacket,
packet_type: PacketType,
key_rotation: SphinxKeyRotation,
) -> Self {
// If this fails somebody is using the library in a super incorrect way, because they
// already managed to somehow create a sphinx packet
let packet_size = PacketSize::get_type(packet.len()).unwrap();
@@ -26,6 +31,7 @@ impl FramedNymPacket {
let header = Header {
packet_version: PacketVersion::new(),
packet_size,
key_rotation,
packet_type,
};
@@ -60,13 +66,16 @@ impl FramedNymPacket {
// Contains any metadata that might be useful for sending between mix nodes.
// TODO: in theory all those data could be put in a single `u8` by setting appropriate bits,
// but would that really be worth it?
#[derive(Debug, Default, PartialEq, Eq, Copy, Clone)]
#[derive(Debug, PartialEq, Eq, Copy, Clone)]
pub struct Header {
/// Represents the wire format version used to construct this packet.
pub(crate) packet_version: PacketVersion,
pub packet_version: PacketVersion,
/// Represents type and consequently size of the included SphinxPacket.
pub(crate) packet_size: PacketSize,
pub packet_size: PacketSize,
/// Represents information regarding which key rotation has been used for constructing this packet.
pub key_rotation: SphinxKeyRotation,
/// Represents whether this packet is sent in a `vpn_mode` meaning it should not get delayed
/// and shared keys might get reused. Mixnodes are capable of inferring this mode from the
@@ -77,35 +86,36 @@ pub struct Header {
/// (note: this will be behind some encryption, either something implemented by us or some SSL action)
// Note: currently packet_type is deprecated but is still left as a concept behind to not break
// compatibility with existing network
pub(crate) packet_type: PacketType,
pub packet_type: PacketType,
}
impl Header {
pub(crate) const SIZE: usize = 3;
pub fn outfox() -> Header {
Header {
packet_version: PacketVersion::default(),
packet_size: PacketSize::OutfoxRegularPacket,
packet_type: PacketType::Outfox,
}
}
pub(crate) const INITIAL_SIZE: usize = 3;
pub(crate) const V8_SIZE: usize = 4;
pub(crate) fn encode(&self, dst: &mut BytesMut) {
dst.reserve(Self::SIZE);
if self.packet_version.is_initial() {
dst.reserve(Self::INITIAL_SIZE);
} else {
dst.reserve(Self::V8_SIZE)
}
dst.put_u8(self.packet_version.as_u8());
dst.put_u8(self.packet_size as u8);
dst.put_u8(self.packet_type as u8);
if !self.packet_version.is_initial() {
dst.put_u8(self.key_rotation as u8)
}
// reserve bytes for the actual packet
dst.reserve(self.packet_size.size());
}
pub(crate) fn decode(src: &mut BytesMut) -> Result<Option<Self>, NymCodecError> {
if src.len() < Self::SIZE {
if src.len() < Self::INITIAL_SIZE {
// can't do anything if we don't have enough bytes - but reserve enough for the next call
src.reserve(Self::SIZE);
src.reserve(Self::INITIAL_SIZE);
return Ok(None);
}
@@ -119,10 +129,22 @@ impl Header {
});
}
// we need to be able to decode the full header
if !packet_version.is_initial() && src.len() < Self::V8_SIZE {
src.reserve(Self::V8_SIZE)
}
let key_rotation = if packet_version.is_initial() {
SphinxKeyRotation::Unknown
} else {
SphinxKeyRotation::try_from(src[3])?
};
Ok(Some(Header {
packet_version,
packet_size: PacketSize::try_from(src[1])?,
packet_type: PacketType::try_from(src[2])?,
key_rotation,
}))
}
}
common/nymsphinx/framing/src/processing.rs
+9 -3
View File
@@ -103,6 +103,9 @@ pub enum PacketProcessingError {
#[error("attempted to partially process an outfox packet")]
PartialOutfoxProcessing,
#[error("the key needed for unwrapping this packet has already expired")]
ExpiredKey,
#[error("this packet has already been processed before")]
PacketReplay,
}
@@ -119,16 +122,19 @@ impl PartiallyUnwrappedPacket {
pub fn new(
received_data: FramedNymPacket,
sphinx_key: &PrivateKey,
) -> Result<Self, PacketProcessingError> {
) -> Result<Self, (FramedNymPacket, PacketProcessingError)> {
let partial_result = match received_data.packet() {
NymPacket::Sphinx(packet) => {
let expanded_shared_secret =
packet.header.compute_expanded_shared_secret(sphinx_key);
// don't continue if the header is malformed
packet
if let Err(err) = packet
.header
.ensure_header_integrity(&expanded_shared_secret)?;
.ensure_header_integrity(&expanded_shared_secret)
{
return Err((received_data, err.into()));
}
PartialMixProcessingResult::Sphinx {
expanded_shared_secret,
common/nymsphinx/params/src/key_rotation.rs
+45
View File
@@ -0,0 +1,45 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use thiserror::Error;
#[derive(Default, Clone, Copy, Debug, PartialEq, Eq)]
#[repr(u8)]
pub enum SphinxKeyRotation {
// for legacy packets, where there's no explicit information which key has been used
#[default]
Unknown = 0,
OddRotation = 1,
EvenRotation = 2,
}
#[derive(Debug, Error)]
#[error("{received} is not a valid encoding of a sphinx key rotation")]
pub struct InvalidSphinxKeyRotation {
received: u8,
}
impl From<u32> for SphinxKeyRotation {
fn from(value: u32) -> Self {
if value % 2 == 0 {
SphinxKeyRotation::EvenRotation
} else {
SphinxKeyRotation::OddRotation
}
}
}
// if value is actually provided, it MUST be one of the two. otherwise is invalid
impl TryFrom<u8> for SphinxKeyRotation {
type Error = InvalidSphinxKeyRotation;
fn try_from(value: u8) -> Result<Self, Self::Error> {
match value {
_ if value == (Self::OddRotation as u8) => Ok(Self::OddRotation),
_ if value == (Self::EvenRotation as u8) => Ok(Self::EvenRotation),
received => Err(InvalidSphinxKeyRotation { received }),
}
}
}
common/nymsphinx/params/src/lib.rs
+3
View File
@@ -9,9 +9,12 @@ use nym_crypto::Aes256GcmSiv;
type Aes128Ctr = ctr::Ctr64BE<Aes128>;
// Re-export for ease of use
pub use key_rotation::SphinxKeyRotation;
pub use packet_sizes::PacketSize;
pub use packet_types::PacketType;
pub use packet_version::PacketVersion;
pub mod key_rotation;
pub mod packet_sizes;
pub mod packet_types;
pub mod packet_version;
common/nymsphinx/params/src/packet_version.rs
+8 -2
View File
@@ -10,13 +10,15 @@ use thiserror::Error;
// - packet_version (starting with v1.1.0)
// - packet_size indicator
// - packet_type
// - sphinx key rotation (starting with v1.11.0)
const TODO: &str = "update the ^ version number when this PR is almost ready";
// it also just so happens that the only valid values for packet_size indicator include values 1-6
// therefore if we receive byte `7` (or larger than that) we'll know we received a versioned packet,
// otherwise we should treat it as legacy
/// Increment it whenever we perform any breaking change in the wire format!
pub const INITIAL_PACKET_VERSION_NUMBER: u8 = 7;
pub const CURRENT_PACKET_VERSION_NUMBER: u8 = INITIAL_PACKET_VERSION_NUMBER;
pub const KEY_ROTATION_VERSION_NUMBER: u8 = 8;
pub const CURRENT_PACKET_VERSION_NUMBER: u8 = KEY_ROTATION_VERSION_NUMBER;
pub const CURRENT_PACKET_VERSION: PacketVersion =
PacketVersion::unchecked(CURRENT_PACKET_VERSION_NUMBER);
@@ -38,6 +40,10 @@ impl PacketVersion {
PacketVersion(CURRENT_PACKET_VERSION_NUMBER)
}
pub fn is_initial(&self) -> bool {
self.0 == INITIAL_PACKET_VERSION_NUMBER
}
const fn unchecked(version: u8) -> PacketVersion {
PacketVersion(version)
}
nym-node/src/node/key_rotation/active_keys.rs
+16 -1
View File
@@ -28,11 +28,26 @@ impl ActiveSphinxKeys {
}
}
pub(crate) fn even(&self) -> Option<impl Deref<Target = SphinxPrivateKey>> {
let primary = self.inner.primary_key.load();
if primary.is_even_rotation() {
return Some(primary);
}
self.secondary()
}
pub(crate) fn odd(&self) -> Option<impl Deref<Target = SphinxPrivateKey>> {
let primary = self.inner.primary_key.load();
if !primary.is_even_rotation() {
return Some(primary);
}
self.secondary()
}
pub(crate) fn primary(&self) -> impl Deref<Target = SphinxPrivateKey> {
self.inner.primary_key.map(|k: &SphinxPrivateKey| k).load()
}
// can't do the same trait bounds due to required borrow on the option
pub(crate) fn secondary(&self) -> Option<impl Deref<Target = SphinxPrivateKey>> {
let guard = self.inner.secondary_key.load();
if guard.is_none() {
nym-node/src/node/key_rotation/key.rs
+4
View File
@@ -31,6 +31,10 @@ impl SphinxPrivateKey {
pub(crate) fn x25519_pubkey(&self) -> x25519::PublicKey {
self.inner.public_key()
}
pub(crate) fn is_even_rotation(&self) -> bool {
self.rotation_id & 1 == 0
}
}
impl From<&SphinxPrivateKey> for SphinxPublicKey {
nym-node/src/node/mixnet/handler.rs
+107 -41
View File
@@ -10,6 +10,7 @@ use nym_sphinx_framing::processing::{
process_framed_packet, MixProcessingResult, MixProcessingResultData, PacketProcessingError,
PartiallyUnwrappedPacket, ProcessedFinalHop,
};
use nym_sphinx_params::SphinxKeyRotation;
use nym_sphinx_types::{Delay, REPLAY_TAG_SIZE};
use std::mem;
use std::net::SocketAddr;
@@ -212,45 +213,76 @@ impl ConnectionHandler {
time_threshold && count_threshold
}
fn try_partially_unwrap_packet(
&self,
packet: FramedNymPacket,
) -> Result<PartiallyUnwrappedPacket, PacketProcessingError> {
// based on the received sphinx key rotation information,
// attempt to choose appropriate key for processing the packet
match packet.header().key_rotation {
SphinxKeyRotation::Unknown => {
// we have to try both keys, start with the primary as it has higher likelihood of being correct
// if let Ok(partially_unwrapped) = PartiallyUnwrappedPacket::new()
match PartiallyUnwrappedPacket::new(packet, self.shared.sphinx_keys.primary()) {
Ok(unwrapped_packet) => Ok(unwrapped_packet),
Err((packet, err)) => {
if let Some(secondary) = self.shared.sphinx_keys.secondary() {
PartiallyUnwrappedPacket::new(packet, secondary).map_err(|(_, err)| err)
} else {
Err(err)
}
}
}
}
SphinxKeyRotation::OddRotation => {
let Some(odd_key) = self.shared.sphinx_keys.odd() else {
return Err(PacketProcessingError::ExpiredKey);
};
PartiallyUnwrappedPacket::new(packet, odd_key).map_err(|(_, err)| err)
}
SphinxKeyRotation::EvenRotation => {
let Some(even_key) = self.shared.sphinx_keys.even() else {
return Err(PacketProcessingError::ExpiredKey);
};
PartiallyUnwrappedPacket::new(packet, even_key).map_err(|(_, err)| err)
}
}
}
async fn handle_received_packet_with_replay_detection(
&mut self,
now: Instant,
packet: FramedNymPacket,
) {
todo!()
//
// // 1. derive and expand shared secret
// // also check the header integrity
// let partially_unwrapped = match PartiallyUnwrappedPacket::new(
// packet,
// self.shared.sphinx_keys.private_key().as_ref(),
// ) {
// Ok(unwrapped) => unwrapped,
// Err(err) => {
// trace!("failed to process received mix packet: {err}");
// self.shared
// .metrics
// .mixnet
// .ingress_malformed_packet(self.remote_address.ip());
// return;
// }
// };
//
// self.pending_packets.push(now, partially_unwrapped);
//
// // 2. check for packet replay
// // 2.1 first try it without locking
// if self.handle_pending_packets_batch_no_locking(now).await {
// return;
// }
//
// // 2.2 if we're within deferral threshold, just leave it queued up for another call
// if self.within_deferral_threshold(now) {
// return;
// }
//
// // 2.3. otherwise block until we obtain the lock and clear the whole batch
// self.handle_pending_packets_batch(now).await;
// 1. derive and expand shared secret
// also check the header integrity
let partially_unwrapped = match self.try_partially_unwrap_packet(packet) {
Ok(unwrapped) => unwrapped,
Err(err) => {
trace!("failed to process received mix packet: {err}");
self.shared
.metrics
.mixnet
.ingress_malformed_packet(self.remote_address.ip());
return;
}
};
self.pending_packets.push(now, partially_unwrapped);
// 2. check for packet replay
// 2.1 first try it without locking
if self.handle_pending_packets_batch_no_locking(now).await {
return;
}
// 2.2 if we're within deferral threshold, just leave it queued up for another call
if self.within_deferral_threshold(now) {
return;
}
// 2.3. otherwise block until we obtain the lock and clear the whole batch
self.handle_pending_packets_batch(now).await;
}
async fn handle_unwrapped_packet(
@@ -342,6 +374,43 @@ impl ConnectionHandler {
.await;
}
fn try_full_unwrap_packet(
&self,
packet: FramedNymPacket,
) -> Result<MixProcessingResult, PacketProcessingError> {
// based on the received sphinx key rotation information,
// attempt to choose appropriate key for processing the packet
// NOTE: due to the function signatures, outfox packets will **only** attempt primary key
// if no rotation information is available (but that's fine given outfox is not really in use,
// and by the time we need it, the rotation info should be present)
match packet.header().key_rotation {
SphinxKeyRotation::Unknown => {
process_framed_packet(packet, self.shared.sphinx_keys.primary())
}
SphinxKeyRotation::OddRotation => {
let Some(odd_key) = self.shared.sphinx_keys.odd() else {
return Err(PacketProcessingError::ExpiredKey);
};
process_framed_packet(packet, odd_key)
}
SphinxKeyRotation::EvenRotation => {
let Some(even_key) = self.shared.sphinx_keys.even() else {
return Err(PacketProcessingError::ExpiredKey);
};
process_framed_packet(packet, even_key)
}
}
}
async fn handle_received_packet_with_no_replay_detection(
&mut self,
now: Instant,
packet: FramedNymPacket,
) {
let unwrapped_packet = self.try_full_unwrap_packet(packet);
self.handle_unwrapped_packet(now, unwrapped_packet).await;
}
#[instrument(skip(self, packet), level = "debug")]
async fn handle_received_nym_packet(&mut self, packet: FramedNymPacket) {
let now = Instant::now();
@@ -352,13 +421,10 @@ impl ConnectionHandler {
self.handle_received_packet_with_replay_detection(now, packet)
.await;
} else {
todo!()
//
// // otherwise just skip that whole procedure and go straight to payload unwrapping
// // (assuming the basic framing is valid)
// let unwrapped_packet =
// process_framed_packet(packet, self.shared.sphinx_keys.private_key().as_ref());
// self.handle_unwrapped_packet(now, unwrapped_packet).await;
// otherwise just skip that whole procedure and go straight to payload unwrapping
// (assuming the basic framing is valid)
self.handle_received_packet_with_no_replay_detection(now, packet)
.await;
};
}