diff --git a/Cargo.lock b/Cargo.lock index a737a05462..23f75f575f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6166,9 +6166,15 @@ name = "nym-sphinx-framing" version = "0.1.0" dependencies = [ "bytes", + "log", + "nym-metrics", + "nym-sphinx-acknowledgements", + "nym-sphinx-addressing", + "nym-sphinx-forwarding", "nym-sphinx-params", "nym-sphinx-types", "thiserror", + "tokio", "tokio-util", ] diff --git a/common/mixnode-common/src/packet_processor/error.rs b/common/mixnode-common/src/packet_processor/error.rs index 405e10bb91..50ae179e37 100644 --- a/common/mixnode-common/src/packet_processor/error.rs +++ b/common/mixnode-common/src/packet_processor/error.rs @@ -2,30 +2,17 @@ // SPDX-License-Identifier: Apache-2.0 use nym_sphinx_acknowledgements::surb_ack::SurbAckRecoveryError; -use nym_sphinx_addressing::nodes::NymNodeRoutingAddressError; -use nym_sphinx_types::{NymPacketError, OutfoxError, SphinxError}; +use nym_sphinx_framing::processing::PacketProcessingError; use thiserror::Error; #[derive(Error, Debug)] pub enum MixProcessingError { - #[error("failed to process received packet: {0}")] - NymPacketProcessingError(#[from] NymPacketError), - - #[error("failed to process received sphinx packet: {0}")] - SphinxProcessingError(#[from] SphinxError), - - #[error("the forward hop address was malformed: {0}")] - InvalidForwardHopAddress(#[from] NymNodeRoutingAddressError), - - #[error("the final hop did not contain a SURB-Ack")] - NoSurbAckInFinalHop, - #[error("failed to recover the expected SURB-Ack packet: {0}")] MalformedSurbAck(#[from] SurbAckRecoveryError), #[error("the received packet was set to use the very old and very much deprecated 'VPN' mode")] ReceivedOldTypeVpnPacket, - #[error("failed to process received outfox packet: {0}")] - OutfoxProcessingError(#[from] OutfoxError), + #[error("failed to process received Nym packet: {0}")] + NymPacketProcessingError(#[from] PacketProcessingError), } diff --git a/common/mixnode-common/src/packet_processor/processor.rs b/common/mixnode-common/src/packet_processor/processor.rs index 03a8f37880..7ab70cdce6 100644 --- a/common/mixnode-common/src/packet_processor/processor.rs +++ b/common/mixnode-common/src/packet_processor/processor.rs @@ -1,38 +1,9 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::packet_processor::error::MixProcessingError; -use log::*; -use nym_metrics::nanos; -use nym_sphinx_acknowledgements::surb_ack::SurbAck; -use nym_sphinx_addressing::nodes::NymNodeRoutingAddress; -use nym_sphinx_forwarding::packet::MixPacket; -use nym_sphinx_framing::packet::FramedNymPacket; -use nym_sphinx_params::{PacketSize, PacketType}; -use nym_sphinx_types::{ - Delay as SphinxDelay, DestinationAddressBytes, NodeAddressBytes, NymPacket, NymProcessedPacket, - PrivateKey, ProcessedPacket, -}; - use std::sync::Arc; -type ForwardAck = MixPacket; - -#[derive(Debug)] -pub struct ProcessedFinalHop { - pub destination: DestinationAddressBytes, - pub forward_ack: Option, - pub message: Vec, -} - -#[derive(Debug)] -pub enum MixProcessingResult { - /// Contains unwrapped data that should first get delayed before being sent to next hop. - ForwardHop(MixPacket, Option), - - /// Contains all data extracted out of the final hop packet that could be forwarded to the destination. - FinalHop(ProcessedFinalHop), -} +use nym_sphinx_types::PrivateKey; #[derive(Clone)] pub struct SphinxPacketProcessor { @@ -48,280 +19,7 @@ impl SphinxPacketProcessor { } } - /// Performs a fresh sphinx unwrapping using no cache. - fn perform_initial_packet_processing( - &self, - packet: NymPacket, - ) -> Result { - nanos!("perform_initial_packet_processing", { - packet.process(&self.sphinx_key).map_err(|err| { - debug!("Failed to unwrap NymPacket packet: {err}"); - MixProcessingError::NymPacketProcessingError(err) - }) - }) - } - - /// Takes the received framed packet and tries to unwrap it from the sphinx encryption. - fn perform_initial_unwrapping( - &self, - received: FramedNymPacket, - ) -> Result { - nanos!("perform_initial_unwrapping", { - let packet = received.into_inner(); - self.perform_initial_packet_processing(packet) - }) - } - - /// Processed received forward hop packet - tries to extract next hop address, sets delay - /// and packs all the data in a way that can be easily sent to the next hop. - fn process_forward_hop( - &self, - packet: NymPacket, - forward_address: NodeAddressBytes, - delay: SphinxDelay, - packet_type: PacketType, - ) -> Result { - let next_hop_address = NymNodeRoutingAddress::try_from(forward_address)?; - - let mix_packet = MixPacket::new(next_hop_address, packet, packet_type); - Ok(MixProcessingResult::ForwardHop(mix_packet, Some(delay))) - } - - /// Split data extracted from the final hop sphinx packet into a SURBAck and message - /// that should get delivered to a client. - fn split_hop_data_into_ack_and_message( - &self, - mut extracted_data: Vec, - packet_type: PacketType, - ) -> Result<(Vec, Vec), MixProcessingError> { - let ack_len = SurbAck::len(Some(packet_type)); - - // in theory it's impossible for this to fail since it managed to go into correct `match` - // branch at the caller - if extracted_data.len() < ack_len { - return Err(MixProcessingError::NoSurbAckInFinalHop); - } - - let message = extracted_data.split_off(ack_len); - let ack_data = extracted_data; - Ok((ack_data, message)) - } - - /// Tries to extract a SURBAck that could be sent back into the mix network and message - /// that should get delivered to a client from received Sphinx packet. - fn split_into_ack_and_message( - &self, - data: Vec, - packet_size: PacketSize, - packet_type: PacketType, - ) -> Result<(Option, Vec), MixProcessingError> { - match packet_size { - PacketSize::AckPacket | PacketSize::OutfoxAckPacket => { - trace!("received an ack packet!"); - Ok((None, data)) - } - PacketSize::RegularPacket - | PacketSize::ExtendedPacket8 - | PacketSize::ExtendedPacket16 - | PacketSize::ExtendedPacket32 - | PacketSize::OutfoxRegularPacket => { - trace!("received a normal packet!"); - let (ack_data, message) = - self.split_hop_data_into_ack_and_message(data, packet_type)?; - let (ack_first_hop, ack_packet) = - match SurbAck::try_recover_first_hop_packet(&ack_data, packet_type) { - Ok((first_hop, packet)) => (first_hop, packet), - Err(err) => { - info!("Failed to recover first hop from ack data: {err}"); - return Err(err.into()); - } - }; - let forward_ack = MixPacket::new(ack_first_hop, ack_packet, packet_type); - Ok((Some(forward_ack), message)) - } - } - } - - /// Processed received final hop packet - tries to extract SURBAck out of it (assuming the - /// packet itself is not an ACK) and splits it from the message that should get delivered - /// to the destination. - fn process_final_hop( - &self, - destination: DestinationAddressBytes, - payload: Vec, - packet_size: PacketSize, - packet_type: PacketType, - ) -> Result { - let (forward_ack, message) = - self.split_into_ack_and_message(payload, packet_size, packet_type)?; - - Ok(MixProcessingResult::FinalHop(ProcessedFinalHop { - destination, - forward_ack, - message, - })) - } - - /// Performs final processing for the unwrapped packet based on whether it was a forward hop - /// or a final hop. - fn perform_final_processing( - &self, - packet: NymProcessedPacket, - packet_size: PacketSize, - packet_type: PacketType, - ) -> Result { - match packet { - NymProcessedPacket::Sphinx(packet) => { - match packet { - ProcessedPacket::ForwardHop(packet, address, delay) => self - .process_forward_hop( - NymPacket::Sphinx(*packet), - address, - delay, - packet_type, - ), - // right now there's no use for the surb_id included in the header - probably it should get removed from the - // sphinx all together? - ProcessedPacket::FinalHop(destination, _, payload) => self.process_final_hop( - destination, - payload.recover_plaintext()?, - packet_size, - packet_type, - ), - } - } - NymProcessedPacket::Outfox(packet) => { - let next_address = *packet.next_address(); - let packet = packet.into_packet(); - if packet.is_final_hop() { - self.process_final_hop( - DestinationAddressBytes::from_bytes(next_address), - packet.recover_plaintext()?.to_vec(), - packet_size, - packet_type, - ) - } else { - let mix_packet = MixPacket::new( - NymNodeRoutingAddress::try_from_bytes(&next_address)?, - NymPacket::Outfox(packet), - PacketType::Outfox, - ); - Ok(MixProcessingResult::ForwardHop(mix_packet, None)) - } - } - } - } - - pub fn process_received( - &self, - received: FramedNymPacket, - ) -> Result { - // explicit packet size will help to correctly parse final hop - nanos!("process_received", { - let packet_size = received.packet_size(); - let packet_type = received.packet_type(); - - // unwrap the sphinx packet and if possible and appropriate, cache keys - let processed_packet = self.perform_initial_unwrapping(received)?; - - // for forward packets, extract next hop and set delay (but do NOT delay here) - // for final packets, extract SURBAck - let final_processing_result = - self.perform_final_processing(processed_packet, packet_size, packet_type); - - if final_processing_result.is_err() { - error!("{:?}", final_processing_result) - } - - final_processing_result - }) - } -} - -// TODO: what more could we realistically test here? -#[cfg(test)] -mod tests { - use super::*; - use nym_sphinx_types::crypto::keygen; - - fn fixture() -> SphinxPacketProcessor { - let local_keys = keygen(); - SphinxPacketProcessor::new(local_keys.0) - } - - #[tokio::test] - async fn splitting_hop_data_works_for_sufficiently_long_payload() { - let processor = fixture(); - - let short_data = vec![42u8]; - assert!(processor - .split_hop_data_into_ack_and_message(short_data, PacketType::Mix) - .is_err()); - - let sufficient_data = vec![42u8; SurbAck::len(Some(PacketType::Mix))]; - let (ack, data) = processor - .split_hop_data_into_ack_and_message(sufficient_data.clone(), PacketType::Mix) - .unwrap(); - assert_eq!(sufficient_data, ack); - assert!(data.is_empty()); - - let long_data = vec![42u8; SurbAck::len(Some(PacketType::Mix)) * 5]; - let (ack, data) = processor - .split_hop_data_into_ack_and_message(long_data, PacketType::Mix) - .unwrap(); - assert_eq!(ack.len(), SurbAck::len(Some(PacketType::Mix))); - assert_eq!(data.len(), SurbAck::len(Some(PacketType::Mix)) * 4) - } - - #[tokio::test] - async fn splitting_hop_data_works_for_sufficiently_long_payload_outfox() { - let processor = fixture(); - - let short_data = vec![42u8]; - assert!(processor - .split_hop_data_into_ack_and_message(short_data, PacketType::Outfox) - .is_err()); - - let sufficient_data = vec![42u8; SurbAck::len(Some(PacketType::Outfox))]; - let (ack, data) = processor - .split_hop_data_into_ack_and_message(sufficient_data.clone(), PacketType::Outfox) - .unwrap(); - assert_eq!(sufficient_data, ack); - assert!(data.is_empty()); - - let long_data = vec![42u8; SurbAck::len(Some(PacketType::Outfox)) * 5]; - let (ack, data) = processor - .split_hop_data_into_ack_and_message(long_data, PacketType::Outfox) - .unwrap(); - assert_eq!(ack.len(), SurbAck::len(Some(PacketType::Outfox))); - assert_eq!(data.len(), SurbAck::len(Some(PacketType::Outfox)) * 4) - } - - #[tokio::test] - async fn splitting_into_ack_and_message_returns_whole_data_for_ack() { - let processor = fixture(); - - let data = vec![42u8; SurbAck::len(Some(PacketType::Mix)) + 10]; - let (ack, message) = processor - .split_into_ack_and_message(data.clone(), PacketSize::AckPacket, PacketType::Mix) - .unwrap(); - assert!(ack.is_none()); - assert_eq!(data, message) - } - - #[tokio::test] - async fn splitting_into_ack_and_message_returns_whole_data_for_ack_outfox() { - let processor = fixture(); - - let data = vec![42u8; SurbAck::len(Some(PacketType::Outfox)) + 10]; - let (ack, message) = processor - .split_into_ack_and_message( - data.clone(), - PacketSize::OutfoxAckPacket, - PacketType::Outfox, - ) - .unwrap(); - assert!(ack.is_none()); - assert_eq!(data, message) + pub fn sphinx_key(&self) -> &PrivateKey { + &self.sphinx_key } } diff --git a/common/nymsphinx/framing/Cargo.toml b/common/nymsphinx/framing/Cargo.toml index 250c0dfd4f..03cff4ca9d 100644 --- a/common/nymsphinx/framing/Cargo.toml +++ b/common/nymsphinx/framing/Cargo.toml @@ -11,7 +11,14 @@ repository = { workspace = true } bytes = { workspace = true } tokio-util = { workspace = true, features = ["codec"] } thiserror = { workspace = true } +log = { workspace = true } nym-sphinx-types = { path = "../types", features = ["sphinx", "outfox"] } nym-sphinx-params = { path = "../params", features = ["sphinx", "outfox"] } +nym-sphinx-forwarding = { path = "../forwarding" } +nym-metrics = { path = "../../nym-metrics" } +nym-sphinx-addressing = { path = "../addressing" } +nym-sphinx-acknowledgements = { path = "../acknowledgements" } +[dev-dependencies] +tokio = { workspace = true, features = ["full"] } diff --git a/common/nymsphinx/framing/src/lib.rs b/common/nymsphinx/framing/src/lib.rs index 6c3fe39018..c0da04d8c2 100644 --- a/common/nymsphinx/framing/src/lib.rs +++ b/common/nymsphinx/framing/src/lib.rs @@ -3,3 +3,4 @@ pub mod codec; pub mod packet; +pub mod processing; diff --git a/common/nymsphinx/framing/src/processing.rs b/common/nymsphinx/framing/src/processing.rs new file mode 100644 index 0000000000..f9dd02dcfb --- /dev/null +++ b/common/nymsphinx/framing/src/processing.rs @@ -0,0 +1,284 @@ +use log::{debug, error, info, trace}; +use nym_sphinx_acknowledgements::surb_ack::{SurbAck, SurbAckRecoveryError}; +use nym_sphinx_addressing::nodes::{NymNodeRoutingAddress, NymNodeRoutingAddressError}; +use nym_sphinx_params::{PacketSize, PacketType}; +use nym_sphinx_types::{ + Delay as SphinxDelay, DestinationAddressBytes, NodeAddressBytes, NymPacket, NymPacketError, + NymProcessedPacket, OutfoxError, PrivateKey, ProcessedPacket, SphinxError, +}; +use thiserror::Error; + +use crate::packet::FramedNymPacket; +use nym_metrics::nanos; +use nym_sphinx_forwarding::packet::MixPacket; + +#[derive(Debug)] +pub enum MixProcessingResult { + /// Contains unwrapped data that should first get delayed before being sent to next hop. + ForwardHop(MixPacket, Option), + + /// Contains all data extracted out of the final hop packet that could be forwarded to the destination. + FinalHop(ProcessedFinalHop), +} + +type ForwardAck = MixPacket; + +#[derive(Debug)] +pub struct ProcessedFinalHop { + pub destination: DestinationAddressBytes, + pub forward_ack: Option, + pub message: Vec, +} + +#[derive(Debug, Error)] +pub enum PacketProcessingError { + #[error("failed to process received packet: {0}")] + NymPacketProcessingError(#[from] NymPacketError), + + #[error("failed to process received sphinx packet: {0}")] + SphinxProcessingError(#[from] SphinxError), + + #[error("the forward hop address was malformed: {0}")] + InvalidForwardHopAddress(#[from] NymNodeRoutingAddressError), + + #[error("the final hop did not contain a SURB-Ack")] + NoSurbAckInFinalHop, + + #[error("failed to recover the expected SURB-Ack packet: {0}")] + MalformedSurbAck(#[from] SurbAckRecoveryError), + + #[error("the received packet was set to use the very old and very much deprecated 'VPN' mode")] + ReceivedOldTypeVpnPacket, + + #[error("failed to process received outfox packet: {0}")] + OutfoxProcessingError(#[from] OutfoxError), +} + +pub fn process_framed_packet( + received: FramedNymPacket, + sphinx_key: &PrivateKey, +) -> Result { + nanos!("process_received", { + let packet_size = received.packet_size(); + let packet_type = received.packet_type(); + + // unwrap the sphinx packet and if possible and appropriate, cache keys + let processed_packet = perform_framed_unwrapping(received, sphinx_key)?; + + // for forward packets, extract next hop and set delay (but do NOT delay here) + // for final packets, extract SURBAck + let final_processing_result = + perform_final_processing(processed_packet, packet_size, packet_type); + + if final_processing_result.is_err() { + error!("{:?}", final_processing_result) + } + + final_processing_result + }) +} + +fn perform_framed_unwrapping( + received: FramedNymPacket, + sphinx_key: &PrivateKey, +) -> Result { + nanos!("perform_initial_unwrapping", { + let packet = received.into_inner(); + perform_framed_packet_processing(packet, sphinx_key) + }) +} + +fn perform_framed_packet_processing( + packet: NymPacket, + sphinx_key: &PrivateKey, +) -> Result { + nanos!("perform_initial_packet_processing", { + packet.process(sphinx_key).map_err(|err| { + debug!("Failed to unwrap NymPacket packet: {err}"); + PacketProcessingError::NymPacketProcessingError(err) + }) + }) +} + +fn perform_final_processing( + packet: NymProcessedPacket, + packet_size: PacketSize, + packet_type: PacketType, +) -> Result { + match packet { + NymProcessedPacket::Sphinx(packet) => { + match packet { + ProcessedPacket::ForwardHop(packet, address, delay) => { + process_forward_hop(NymPacket::Sphinx(*packet), address, delay, packet_type) + } + // right now there's no use for the surb_id included in the header - probably it should get removed from the + // sphinx all together? + ProcessedPacket::FinalHop(destination, _, payload) => process_final_hop( + destination, + payload.recover_plaintext()?, + packet_size, + packet_type, + ), + } + } + NymProcessedPacket::Outfox(packet) => { + let next_address = *packet.next_address(); + let packet = packet.into_packet(); + if packet.is_final_hop() { + process_final_hop( + DestinationAddressBytes::from_bytes(next_address), + packet.recover_plaintext()?.to_vec(), + packet_size, + packet_type, + ) + } else { + let mix_packet = MixPacket::new( + NymNodeRoutingAddress::try_from_bytes(&next_address)?, + NymPacket::Outfox(packet), + PacketType::Outfox, + ); + Ok(MixProcessingResult::ForwardHop(mix_packet, None)) + } + } + } +} + +fn process_final_hop( + destination: DestinationAddressBytes, + payload: Vec, + packet_size: PacketSize, + packet_type: PacketType, +) -> Result { + let (forward_ack, message) = split_into_ack_and_message(payload, packet_size, packet_type)?; + + Ok(MixProcessingResult::FinalHop(ProcessedFinalHop { + destination, + forward_ack, + message, + })) +} + +fn split_into_ack_and_message( + data: Vec, + packet_size: PacketSize, + packet_type: PacketType, +) -> Result<(Option, Vec), PacketProcessingError> { + match packet_size { + PacketSize::AckPacket | PacketSize::OutfoxAckPacket => { + trace!("received an ack packet!"); + Ok((None, data)) + } + PacketSize::RegularPacket + | PacketSize::ExtendedPacket8 + | PacketSize::ExtendedPacket16 + | PacketSize::ExtendedPacket32 + | PacketSize::OutfoxRegularPacket => { + trace!("received a normal packet!"); + let (ack_data, message) = split_hop_data_into_ack_and_message(data, packet_type)?; + let (ack_first_hop, ack_packet) = + match SurbAck::try_recover_first_hop_packet(&ack_data, packet_type) { + Ok((first_hop, packet)) => (first_hop, packet), + Err(err) => { + info!("Failed to recover first hop from ack data: {err}"); + return Err(err.into()); + } + }; + let forward_ack = MixPacket::new(ack_first_hop, ack_packet, packet_type); + Ok((Some(forward_ack), message)) + } + } +} + +fn split_hop_data_into_ack_and_message( + mut extracted_data: Vec, + packet_type: PacketType, +) -> Result<(Vec, Vec), PacketProcessingError> { + let ack_len = SurbAck::len(Some(packet_type)); + + // in theory it's impossible for this to fail since it managed to go into correct `match` + // branch at the caller + if extracted_data.len() < ack_len { + return Err(PacketProcessingError::NoSurbAckInFinalHop); + } + + let message = extracted_data.split_off(ack_len); + let ack_data = extracted_data; + Ok((ack_data, message)) +} + +fn process_forward_hop( + packet: NymPacket, + forward_address: NodeAddressBytes, + delay: SphinxDelay, + packet_type: PacketType, +) -> Result { + let next_hop_address = NymNodeRoutingAddress::try_from(forward_address)?; + + let mix_packet = MixPacket::new(next_hop_address, packet, packet_type); + Ok(MixProcessingResult::ForwardHop(mix_packet, Some(delay))) +} + +// TODO: what more could we realistically test here? +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn splitting_hop_data_works_for_sufficiently_long_payload() { + let short_data = vec![42u8]; + assert!(split_hop_data_into_ack_and_message(short_data, PacketType::Mix).is_err()); + + let sufficient_data = vec![42u8; SurbAck::len(Some(PacketType::Mix))]; + let (ack, data) = + split_hop_data_into_ack_and_message(sufficient_data.clone(), PacketType::Mix).unwrap(); + assert_eq!(sufficient_data, ack); + assert!(data.is_empty()); + + let long_data: Vec = vec![42u8; SurbAck::len(Some(PacketType::Mix)) * 5]; + let (ack, data) = split_hop_data_into_ack_and_message(long_data, PacketType::Mix).unwrap(); + assert_eq!(ack.len(), SurbAck::len(Some(PacketType::Mix))); + assert_eq!(data.len(), SurbAck::len(Some(PacketType::Mix)) * 4) + } + + #[tokio::test] + async fn splitting_hop_data_works_for_sufficiently_long_payload_outfox() { + let short_data = vec![42u8]; + assert!(split_hop_data_into_ack_and_message(short_data, PacketType::Outfox).is_err()); + + let sufficient_data = vec![42u8; SurbAck::len(Some(PacketType::Outfox))]; + let (ack, data) = + split_hop_data_into_ack_and_message(sufficient_data.clone(), PacketType::Outfox) + .unwrap(); + assert_eq!(sufficient_data, ack); + assert!(data.is_empty()); + + let long_data = vec![42u8; SurbAck::len(Some(PacketType::Outfox)) * 5]; + let (ack, data) = + split_hop_data_into_ack_and_message(long_data, PacketType::Outfox).unwrap(); + assert_eq!(ack.len(), SurbAck::len(Some(PacketType::Outfox))); + assert_eq!(data.len(), SurbAck::len(Some(PacketType::Outfox)) * 4) + } + + #[tokio::test] + async fn splitting_into_ack_and_message_returns_whole_data_for_ack() { + let data = vec![42u8; SurbAck::len(Some(PacketType::Mix)) + 10]; + let (ack, message) = + split_into_ack_and_message(data.clone(), PacketSize::AckPacket, PacketType::Mix) + .unwrap(); + assert!(ack.is_none()); + assert_eq!(data, message) + } + + #[tokio::test] + async fn splitting_into_ack_and_message_returns_whole_data_for_ack_outfox() { + let data = vec![42u8; SurbAck::len(Some(PacketType::Outfox)) + 10]; + let (ack, message) = split_into_ack_and_message( + data.clone(), + PacketSize::OutfoxAckPacket, + PacketType::Outfox, + ) + .unwrap(); + assert!(ack.is_none()); + assert_eq!(data, message) + } +} diff --git a/contracts/Cargo.lock b/contracts/Cargo.lock index 2f57ea58ef..6185826bbf 100644 --- a/contracts/Cargo.lock +++ b/contracts/Cargo.lock @@ -1935,18 +1935,18 @@ dependencies = [ [[package]] name = "thiserror" -version = "1.0.63" +version = "1.0.64" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0342370b38b6a11b6cc11d6a805569958d54cfa061a29969c3b5ce2ea405724" +checksum = "d50af8abc119fb8bb6dbabcfa89656f46f84aa0ac7688088608076ad2b459a84" dependencies = [ "thiserror-impl", ] [[package]] name = "thiserror-impl" -version = "1.0.63" +version = "1.0.64" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a4558b58466b9ad7ca0f102865eccc95938dca1a74a856f2b57b6629050da261" +checksum = "08904e7672f5eb876eaaf87e0ce17857500934f4981c4a0ab2b4aa98baac7fc3" dependencies = [ "proc-macro2", "quote", diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs index 4ee18c77d5..5eb0b0c136 100644 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs @@ -8,10 +8,10 @@ use futures::channel::mpsc::SendError; use futures::StreamExt; use nym_gateway_storage::{error::StorageError, Storage}; use nym_mixnet_client::forwarder::MixForwardingSender; -use nym_mixnode_common::packet_processor::processor::ProcessedFinalHop; use nym_sphinx::forwarding::packet::MixPacket; use nym_sphinx::framing::codec::NymCodec; use nym_sphinx::framing::packet::FramedNymPacket; +use nym_sphinx::framing::processing::ProcessedFinalHop; use nym_sphinx::DestinationAddressBytes; use nym_task::TaskClient; use std::collections::HashMap; @@ -21,6 +21,8 @@ use tokio::net::TcpStream; use tokio_util::codec::Framed; use tracing::*; +use super::packet_processing::process_packet; + // defines errors that warrant a panic if not thrown in the context of a shutdown #[derive(Debug, Error)] enum CriticalPacketProcessingError { @@ -184,14 +186,14 @@ impl ConnectionHandler { // question: can it also be per connection vs global? // - let processed_final_hop = match self.packet_processor.process_received(framed_sphinx_packet) - { - Err(err) => { - debug!("We failed to process received sphinx packet - {err}"); - return Ok(()); - } - Ok(processed_final_hop) => processed_final_hop, - }; + let processed_final_hop = + match process_packet(framed_sphinx_packet, self.packet_processor.sphinx_key()) { + Err(err) => { + debug!("We failed to process received sphinx packet - {err}"); + return Ok(()); + } + Ok(processed_final_hop) => processed_final_hop, + }; self.handle_processed_packet(processed_final_hop).await } diff --git a/gateway/src/node/mixnet_handling/receiver/packet_processing.rs b/gateway/src/node/mixnet_handling/receiver/packet_processing.rs index 7ac90bb174..4f16c2451f 100644 --- a/gateway/src/node/mixnet_handling/receiver/packet_processing.rs +++ b/gateway/src/node/mixnet_handling/receiver/packet_processing.rs @@ -3,18 +3,24 @@ use nym_crypto::asymmetric::encryption; use nym_mixnode_common::packet_processor::error::MixProcessingError; -pub use nym_mixnode_common::packet_processor::processor::MixProcessingResult; -use nym_mixnode_common::packet_processor::processor::{ProcessedFinalHop, SphinxPacketProcessor}; +use nym_mixnode_common::packet_processor::processor::SphinxPacketProcessor; use nym_sphinx::framing::packet::FramedNymPacket; +use nym_sphinx::framing::processing::{ + process_framed_packet, MixProcessingResult, PacketProcessingError, ProcessedFinalHop, +}; +use nym_sphinx::PrivateKey; use thiserror::Error; #[derive(Error, Debug)] pub enum GatewayProcessingError { #[error("failed to process received mix packet - {0}")] - PacketProcessingError(#[from] MixProcessingError), + PacketProcessing(#[from] MixProcessingError), #[error("received a forward hop mix packet")] - ForwardHopReceivedError, + ForwardHopReceived, + + #[error("failed to process received sphinx packet: {0}")] + NymPacketProcessing(#[from] PacketProcessingError), } // PacketProcessor contains all data required to correctly unwrap and store sphinx packets @@ -24,21 +30,23 @@ pub struct PacketProcessor { } impl PacketProcessor { + pub fn sphinx_key(&self) -> &PrivateKey { + self.inner_processor.sphinx_key() + } + pub(crate) fn new(encryption_key: &encryption::PrivateKey) -> Self { PacketProcessor { inner_processor: SphinxPacketProcessor::new(encryption_key.into()), } } +} - pub(crate) fn process_received( - &self, - received: FramedNymPacket, - ) -> Result { - match self.inner_processor.process_received(received)? { - MixProcessingResult::ForwardHop(..) => { - Err(GatewayProcessingError::ForwardHopReceivedError) - } - MixProcessingResult::FinalHop(processed_final) => Ok(processed_final), - } +pub(crate) fn process_packet( + received: FramedNymPacket, + sphinx_key: &nym_sphinx::PrivateKey, +) -> Result { + match process_framed_packet(received, sphinx_key)? { + MixProcessingResult::ForwardHop(..) => Err(GatewayProcessingError::ForwardHopReceived), + MixProcessingResult::FinalHop(processed_final) => Ok(processed_final), } } diff --git a/mixnode/src/node/listener/connection_handler/mod.rs b/mixnode/src/node/listener/connection_handler/mod.rs index f4144c4f67..0b76762667 100644 --- a/mixnode/src/node/listener/connection_handler/mod.rs +++ b/mixnode/src/node/listener/connection_handler/mod.rs @@ -1,9 +1,7 @@ // Copyright 2020 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -use crate::node::listener::connection_handler::packet_processing::{ - MixProcessingResult, PacketProcessor, -}; +use crate::node::listener::connection_handler::packet_processing::PacketProcessor; use crate::node::packet_delayforwarder::PacketDelayForwardSender; use crate::node::TaskClient; use futures::StreamExt; @@ -13,7 +11,9 @@ use nym_metrics::nanos; use nym_sphinx::forwarding::packet::MixPacket; use nym_sphinx::framing::codec::NymCodec; use nym_sphinx::framing::packet::FramedNymPacket; +use nym_sphinx::framing::processing::MixProcessingResult; use nym_sphinx::Delay as SphinxDelay; +use packet_processing::process_received_packet; use std::net::SocketAddr; use tokio::net::TcpStream; use tokio::time::Instant; @@ -38,6 +38,10 @@ impl ConnectionHandler { } } + pub fn packet_processor(&self) -> &PacketProcessor { + &self.packet_processor + } + fn delay_and_forward_packet(&self, mix_packet: MixPacket, delay: Option) { // determine instant at which packet should get forwarded. this way we minimise effect of // being stuck in the queue [of the channel] to get inserted into the delay queue @@ -60,7 +64,10 @@ impl ConnectionHandler { // all processing such, key caching, etc. was done. // however, if it was a forward hop, we still need to delay it nanos!("handle_received_packet", { - match self.packet_processor.process_received(framed_sphinx_packet) { + self.packet_processor + .node_stats_update_sender() + .report_received(); + match process_received_packet(framed_sphinx_packet, self.packet_processor().inner()) { Err(err) => debug!("We failed to process received sphinx packet - {err}"), Ok(res) => match res { MixProcessingResult::ForwardHop(forward_packet, delay) => { diff --git a/mixnode/src/node/listener/connection_handler/packet_processing.rs b/mixnode/src/node/listener/connection_handler/packet_processing.rs index f849632f0f..9e6742a6b8 100644 --- a/mixnode/src/node/listener/connection_handler/packet_processing.rs +++ b/mixnode/src/node/listener/connection_handler/packet_processing.rs @@ -4,13 +4,13 @@ use crate::node::node_statistics; use nym_crypto::asymmetric::encryption; use nym_mixnode_common::packet_processor::error::MixProcessingError; -pub use nym_mixnode_common::packet_processor::processor::MixProcessingResult; use nym_mixnode_common::packet_processor::processor::SphinxPacketProcessor; use nym_sphinx::framing::packet::FramedNymPacket; +use nym_sphinx::framing::processing::{process_framed_packet, MixProcessingResult}; // PacketProcessor contains all data required to correctly unwrap and forward sphinx packets #[derive(Clone)] -pub struct PacketProcessor { +pub(crate) struct PacketProcessor { /// Responsible for performing unwrapping inner_processor: SphinxPacketProcessor, @@ -29,11 +29,18 @@ impl PacketProcessor { } } - pub(crate) fn process_received( - &self, - received: FramedNymPacket, - ) -> Result { - self.node_stats_update_sender.report_received(); - self.inner_processor.process_received(received) + pub fn inner(&self) -> &SphinxPacketProcessor { + &self.inner_processor + } + + pub fn node_stats_update_sender(&self) -> &node_statistics::UpdateSender { + &self.node_stats_update_sender } } + +pub fn process_received_packet( + packet: FramedNymPacket, + inner_processor: &SphinxPacketProcessor, +) -> Result { + Ok(process_framed_packet(packet, inner_processor.sphinx_key())?) +} diff --git a/nym-wallet/Cargo.lock b/nym-wallet/Cargo.lock index 3813f1ba4c..f65e796297 100644 --- a/nym-wallet/Cargo.lock +++ b/nym-wallet/Cargo.lock @@ -173,9 +173,9 @@ checksum = "6b4930d2cb77ce62f89ee5d5289b4ac049559b1c45539271f5ed4fdc7db34545" [[package]] name = "async-trait" -version = "0.1.82" +version = "0.1.83" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a27b8a3a6e1a44fa4c8baf1f653e4172e81486d4941f2237e20dc2d0cf4ddff1" +checksum = "721cae7de5c34fbb2acd27e21e6d2cf7b886dce0c27388d46c4e6c47ea4318dd" dependencies = [ "proc-macro2", "quote", @@ -1637,9 +1637,9 @@ dependencies = [ [[package]] name = "flate2" -version = "1.0.33" +version = "1.0.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "324a1be68054ef05ad64b861cc9eaf1d623d2d8cb25b4bf2cb9cdd902b4bf253" +checksum = "a1b589b4dc103969ad3cf85c950899926ec64300a1a46d76c03a6072957036f0" dependencies = [ "crc32fast", "miniz_oxide 0.8.0", @@ -5760,18 +5760,18 @@ checksum = "8eaa81235c7058867fa8c0e7314f33dcce9c215f535d1913822a2b3f5e289f3c" [[package]] name = "thiserror" -version = "1.0.63" +version = "1.0.64" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0342370b38b6a11b6cc11d6a805569958d54cfa061a29969c3b5ce2ea405724" +checksum = "d50af8abc119fb8bb6dbabcfa89656f46f84aa0ac7688088608076ad2b459a84" dependencies = [ "thiserror-impl", ] [[package]] name = "thiserror-impl" -version = "1.0.63" +version = "1.0.64" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a4558b58466b9ad7ca0f102865eccc95938dca1a74a856f2b57b6629050da261" +checksum = "08904e7672f5eb876eaaf87e0ce17857500934f4981c4a0ab2b4aa98baac7fc3" dependencies = [ "proc-macro2", "quote",