From fb253e53a4c3bbf6ceeb9868f5039a6a9cf4177e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 11 Aug 2021 12:47:35 +0100 Subject: [PATCH] Removed all sphinx key caching from mixnodes and gateways (#713) * Removed all sphinx key caching from mixnodes and gateways * Missing change in network monitor --- .../acknowledgement_control/mod.rs | 15 - .../src/client/real_messages_control/mod.rs | 26 +- .../real_traffic_stream.rs | 27 +- clients/client-core/src/config/mod.rs | 56 +-- clients/native/src/client/config/template.rs | 5 - clients/native/src/client/mod.rs | 15 +- clients/native/src/commands/init.rs | 11 - clients/native/src/commands/mod.rs | 4 - clients/native/src/commands/run.rs | 11 - clients/socks5/src/client/config/template.rs | 5 - clients/socks5/src/client/mod.rs | 16 +- clients/socks5/src/commands/init.rs | 11 - clients/socks5/src/commands/mod.rs | 4 - clients/socks5/src/commands/run.rs | 11 - clients/webassembly/src/client/mod.rs | 4 - .../src/cached_packet_processor/cache.rs | 170 ------- .../src/cached_packet_processor/processor.rs | 475 ------------------ common/mixnode-common/src/lib.rs | 2 +- .../error.rs | 5 + .../mod.rs | 1 - .../src/packet_processor/processor.rs | 234 +++++++++ .../acknowledgements/src/surb_ack.rs | 12 +- common/nymsphinx/cover/src/lib.rs | 2 - common/nymsphinx/framing/src/packet.rs | 2 + common/nymsphinx/params/src/packet_modes.rs | 2 +- common/nymsphinx/src/preparer/mod.rs | 85 +--- common/nymsphinx/src/preparer/vpn_manager.rs | 143 ------ gateway/src/config/mod.rs | 13 - .../receiver/connection_handler.rs | 4 +- .../receiver/packet_processing.rs | 22 +- gateway/src/node/mod.rs | 6 +- mixnode/src/config/mod.rs | 13 - .../node/listener/connection_handler/mod.rs | 8 +- .../connection_handler/packet_processing.rs | 20 +- mixnode/src/node/listener/mod.rs | 2 +- mixnode/src/node/mod.rs | 7 +- validator-api/src/network_monitor/chunker.rs | 3 - 37 files changed, 301 insertions(+), 1151 deletions(-) delete mode 100644 common/mixnode-common/src/cached_packet_processor/cache.rs delete mode 100644 common/mixnode-common/src/cached_packet_processor/processor.rs rename common/mixnode-common/src/{cached_packet_processor => packet_processor}/error.rs (91%) rename common/mixnode-common/src/{cached_packet_processor => packet_processor}/mod.rs (92%) create mode 100644 common/mixnode-common/src/packet_processor/processor.rs delete mode 100644 common/nymsphinx/src/preparer/vpn_manager.rs diff --git a/clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs b/clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs index 5156a34115..a987497af1 100644 --- a/clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs @@ -13,7 +13,6 @@ use crate::client::{inbound_messages::InputMessageReceiver, topology_control::To use futures::channel::mpsc; use gateway_client::AcknowledgementReceiver; use log::*; -use nymsphinx::params::PacketMode; use nymsphinx::{ acknowledgements::AckKey, addressing::clients::Recipient, @@ -120,14 +119,6 @@ pub(super) struct Config { /// Average delay a data packet is going to get delayed at a single mixnode. average_packet_delay: Duration, - - /// Mode of all mix packets created - VPN or Mix. They indicate whether packets should get delayed - /// and keys reused. - packet_mode: PacketMode, - - /// If the mode of the client is set to VPN it specifies number of packets created with the - /// same initial secret until it gets rotated. - vpn_key_reuse_limit: Option, } impl Config { @@ -136,16 +127,12 @@ impl Config { ack_wait_multiplier: f64, average_ack_delay: Duration, average_packet_delay: Duration, - packet_mode: PacketMode, - vpn_key_reuse_limit: Option, ) -> Self { Config { ack_wait_addition, ack_wait_multiplier, average_ack_delay, average_packet_delay, - packet_mode, - vpn_key_reuse_limit, } } } @@ -186,8 +173,6 @@ where ack_recipient, config.average_packet_delay, config.average_ack_delay, - config.packet_mode, - config.vpn_key_reuse_limit, ); // will listen for any acks coming from the network diff --git a/clients/client-core/src/client/real_messages_control/mod.rs b/clients/client-core/src/client/real_messages_control/mod.rs index 78681a6101..693aa8ceb6 100644 --- a/clients/client-core/src/client/real_messages_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/mod.rs @@ -19,7 +19,6 @@ use gateway_client::AcknowledgementReceiver; use log::*; use nymsphinx::acknowledgements::AckKey; use nymsphinx::addressing::clients::Recipient; -use nymsphinx::params::PacketMode; use rand::{rngs::OsRng, CryptoRng, Rng}; use std::sync::Arc; use std::time::Duration; @@ -51,20 +50,9 @@ pub struct Config { /// Average delay an acknowledgement packet is going to get delayed at a single mixnode. average_ack_delay_duration: Duration, - - /// Mode of all mix packets created - VPN or Mix. They indicate whether packets should get delayed - /// and keys reused. - packet_mode: PacketMode, - - /// If the mode of the client is set to VPN it specifies number of packets created with the - /// same initial secret until it gets rotated. - vpn_key_reuse_limit: Option, } impl Config { - // at this point I'm not entirely sure how to deal with this warning without - // some considerable refactoring - #[allow(clippy::too_many_arguments)] pub fn new( ack_key: Arc, ack_wait_multiplier: f64, @@ -73,8 +61,6 @@ impl Config { average_message_sending_delay: Duration, average_packet_delay_duration: Duration, self_recipient: Recipient, - packet_mode: PacketMode, - vpn_key_reuse_limit: Option, ) -> Self { Config { ack_key, @@ -84,8 +70,6 @@ impl Config { average_message_sending_delay, average_packet_delay_duration, average_ack_delay_duration, - packet_mode, - vpn_key_reuse_limit, } } } @@ -126,8 +110,6 @@ impl RealMessagesController { config.ack_wait_multiplier, config.average_ack_delay_duration, config.average_packet_delay_duration, - config.packet_mode, - config.vpn_key_reuse_limit, ); let ack_control = AcknowledgementController::new( @@ -163,7 +145,7 @@ impl RealMessagesController { } } - pub(super) async fn run(&mut self, vpn_mode: bool) { + pub(super) async fn run(&mut self) { let mut out_queue_control = self.out_queue_control.take().unwrap(); let mut ack_control = self.ack_control.take().unwrap(); @@ -171,7 +153,7 @@ impl RealMessagesController { // the task to ever finish. This will of course change once we introduce // graceful shutdowns. let out_queue_control_fut = tokio::spawn(async move { - out_queue_control.run_out_queue_control(vpn_mode).await; + out_queue_control.run_out_queue_control().await; error!("The out queue controller has finished execution!"); out_queue_control }); @@ -190,9 +172,9 @@ impl RealMessagesController { // &Handle is only passed for consistency sake with other client modules, but I think // when we get to refactoring, we should apply gateway approach and make it implicit - pub fn start(mut self, handle: &Handle, vpn_mode: bool) -> JoinHandle { + pub fn start(mut self, handle: &Handle) -> JoinHandle { handle.spawn(async move { - self.run(vpn_mode).await; + self.run().await; self }) } diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index 34f8b5c9e9..47d95d0d26 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -249,15 +249,6 @@ where tokio::task::yield_now().await; } - async fn on_batch_received(&mut self, real_messages: Vec) { - let mut mix_packets = Vec::with_capacity(real_messages.len()); - for real_message in real_messages.into_iter() { - self.sent_notify(real_message.fragment_id); - mix_packets.push(real_message.mix_packet); - } - self.mix_tx.unbounded_send(mix_packets).unwrap(); - } - // Send messages at certain rate and if no real traffic is available, send cover message. async fn run_normal_out_queue(&mut self) { // we should set initial delay only when we actually start the stream @@ -271,20 +262,8 @@ where } } - // Send real message as soon as it's available and don't inject ANY cover traffic. - async fn run_vpn_out_queue(&mut self) { - while let Some(next_messages) = self.real_receiver.next().await { - self.on_batch_received(next_messages).await - } - } - - pub(crate) async fn run_out_queue_control(&mut self, vpn_mode: bool) { - if vpn_mode { - debug!("Starting out queue controller in vpn mode..."); - self.run_vpn_out_queue().await - } else { - debug!("Starting out queue controller..."); - self.run_normal_out_queue().await - } + pub(crate) async fn run_out_queue_control(&mut self) { + debug!("Starting out queue controller..."); + self.run_normal_out_queue().await } } diff --git a/clients/client-core/src/config/mod.rs b/clients/client-core/src/config/mod.rs index 38b1237d58..993ab21742 100644 --- a/clients/client-core/src/config/mod.rs +++ b/clients/client-core/src/config/mod.rs @@ -22,9 +22,6 @@ const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(50); const DEFAULT_TOPOLOGY_REFRESH_RATE: Duration = Duration::from_secs(5 * 60); // every 5min const DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT: Duration = Duration::from_millis(5_000); const DEFAULT_GATEWAY_RESPONSE_TIMEOUT: Duration = Duration::from_millis(1_500); -const DEFAULT_VPN_KEY_REUSE_LIMIT: usize = 1000; - -const ZERO_DELAY: Duration = Duration::from_nanos(0); // helper function to get default validators as a Vec pub fn default_validator_rest_endpoints() -> Vec { @@ -139,14 +136,6 @@ impl Config { self.debug.message_sending_average_delay = Duration::from_millis(4); // 250 "real" messages / s } - pub fn set_vpn_mode(&mut self, vpn_mode: bool) { - self.client.vpn_mode = vpn_mode; - } - - pub fn set_vpn_key_reuse_limit(&mut self, reuse_limit: usize) { - self.debug.vpn_key_reuse_limit = Some(reuse_limit) - } - pub fn set_custom_version(&mut self, version: &str) { self.client.version = version.to_string(); } @@ -205,19 +194,11 @@ impl Config { // Debug getters pub fn get_average_packet_delay(&self) -> Duration { - if self.client.vpn_mode { - ZERO_DELAY - } else { - self.debug.average_packet_delay - } + self.debug.average_packet_delay } pub fn get_average_ack_delay(&self) -> Duration { - if self.client.vpn_mode { - ZERO_DELAY - } else { - self.debug.average_ack_delay - } + self.debug.average_ack_delay } pub fn get_ack_wait_multiplier(&self) -> f64 { @@ -233,11 +214,7 @@ impl Config { } pub fn get_message_sending_average_delay(&self) -> Duration { - if self.client.vpn_mode { - ZERO_DELAY - } else { - self.debug.message_sending_average_delay - } + self.debug.message_sending_average_delay } pub fn get_gateway_response_timeout(&self) -> Duration { @@ -252,21 +229,6 @@ impl Config { self.debug.topology_resolution_timeout } - pub fn get_vpn_mode(&self) -> bool { - self.client.vpn_mode - } - - pub fn get_vpn_key_reuse_limit(&self) -> Option { - match self.get_vpn_mode() { - false => None, - true => Some( - self.debug - .vpn_key_reuse_limit - .unwrap_or(DEFAULT_VPN_KEY_REUSE_LIMIT), - ), - } - } - pub fn get_version(&self) -> &str { &self.client.version } @@ -303,12 +265,6 @@ pub struct Client { #[serde(default = "missing_string_value")] mixnet_contract_address: String, - /// Special mode of the system such that all messages are sent as soon as they are received - /// and no cover traffic is generated. If set all message delays are set to 0 and overwriting - /// 'Debug' values will have no effect. - #[serde(default)] - vpn_mode: bool, - /// Path to file containing private identity key. private_identity_key_file: PathBuf, @@ -356,7 +312,6 @@ impl Default for Client { id: "".to_string(), validator_rest_urls: default_validator_rest_endpoints(), mixnet_contract_address: DEFAULT_MIXNET_CONTRACT_ADDRESS.to_string(), - vpn_mode: false, private_identity_key_file: Default::default(), public_identity_key_file: Default::default(), private_encryption_key_file: Default::default(), @@ -491,10 +446,6 @@ pub struct Debug { serialize_with = "humantime_serde::serialize" )] topology_resolution_timeout: Duration, - - /// If the mode of the client is set to VPN it specifies number of packets created with the - /// same initial secret until it gets rotated. - vpn_key_reuse_limit: Option, } impl Default for Debug { @@ -509,7 +460,6 @@ impl Default for Debug { gateway_response_timeout: DEFAULT_GATEWAY_RESPONSE_TIMEOUT, topology_refresh_rate: DEFAULT_TOPOLOGY_REFRESH_RATE, topology_resolution_timeout: DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT, - vpn_key_reuse_limit: None, } } } diff --git a/clients/native/src/client/config/template.rs b/clients/native/src/client/config/template.rs index 2a5ed41040..ebe6fe70d7 100644 --- a/clients/native/src/client/config/template.rs +++ b/clients/native/src/client/config/template.rs @@ -29,11 +29,6 @@ validator_rest_urls = [ # Address of the validator contract managing the network. mixnet_contract_address = '{{ client.mixnet_contract_address }}' -# Special mode of the system such that all messages are sent as soon as they are received -# and no cover traffic is generated. If set all message delays are set to 0 and overwriting -# 'Debug' values will have no effect. -vpn_mode = {{ client.vpn_mode }} - # Path to file containing private identity key. private_identity_key_file = '{{ client.private_identity_key_file }}' diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index ab544acb59..2c2e4f79bb 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -32,7 +32,6 @@ use log::*; use nymsphinx::addressing::clients::Recipient; use nymsphinx::addressing::nodes::NodeIdentity; use nymsphinx::anonymous_replies::ReplySurb; -use nymsphinx::params::PacketMode; use nymsphinx::receiver::ReconstructedMessage; use tokio::runtime::Runtime; @@ -119,12 +118,6 @@ impl NymClient { input_receiver: InputMessageReceiver, mix_sender: BatchMixMessageSender, ) { - let packet_mode = if self.config.get_base().get_vpn_mode() { - PacketMode::Vpn - } else { - PacketMode::Mix - }; - let controller_config = real_messages_control::Config::new( self.key_manager.ack_key(), self.config.get_base().get_ack_wait_multiplier(), @@ -133,8 +126,6 @@ impl NymClient { self.config.get_base().get_message_sending_average_delay(), self.config.get_base().get_average_packet_delay(), self.as_mix_recipient(), - packet_mode, - self.config.get_base().get_vpn_key_reuse_limit(), ); info!("Starting real traffic stream..."); @@ -151,7 +142,7 @@ impl NymClient { topology_accessor, reply_key_storage, ) - .start(self.runtime.handle(), self.config.get_base().get_vpn_mode()); + .start(self.runtime.handle()); } // buffer controlling all messages fetched from provider @@ -376,9 +367,7 @@ impl NymClient { sphinx_message_sender.clone(), ); - if !self.config.get_base().get_vpn_mode() { - self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender); - } + self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender); match self.config.get_socket_type() { SocketType::WebSocket => { diff --git a/clients/native/src/commands/init.rs b/clients/native/src/commands/init.rs index 1f68add2d3..1b253d3d22 100644 --- a/clients/native/src/commands/init.rs +++ b/clients/native/src/commands/init.rs @@ -53,17 +53,6 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> { .help("Port for the socket (if applicable) to listen on in all subsequent runs") .takes_value(true) ) - .arg(Arg::with_name("vpn-mode") - .long("vpn-mode") - .help("Set the vpn mode of the client") - .long_help( - r#" - Special mode of the system such that all messages are sent as soon as they are received - and no cover traffic is generated. If set all message delays are set to 0 and overwriting - 'Debug' values will have no effect. - "# - ) - ) .arg(Arg::with_name("fastmode") .long("fastmode") .hidden(true) // this will prevent this flag from being displayed in `--help` diff --git a/clients/native/src/commands/mod.rs b/clients/native/src/commands/mod.rs index 7edab6ee2d..04a25a3f83 100644 --- a/clients/native/src/commands/mod.rs +++ b/clients/native/src/commands/mod.rs @@ -33,10 +33,6 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi config = config.with_socket(SocketType::None); } - if matches.is_present("vpn-mode") { - config.get_base_mut().set_vpn_mode(true); - } - if let Some(port) = matches.value_of("port").map(|port| port.parse::()) { if let Err(err) = port { // if port was overridden, it must be parsable diff --git a/clients/native/src/commands/run.rs b/clients/native/src/commands/run.rs index 0b315a714a..96151e7c70 100644 --- a/clients/native/src/commands/run.rs +++ b/clients/native/src/commands/run.rs @@ -38,17 +38,6 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> { .long("disable-socket") .help("Whether to not start the websocket") ) - .arg(Arg::with_name("vpn-mode") - .long("vpn-mode") - .help("Set the vpn mode of the client") - .long_help( - r#" - Special mode of the system such that all messages are sent as soon as they are received - and no cover traffic is generated. If set all message delays are set to 0 and overwriting - 'Debug' values will have no effect. - "# - ) - ) .arg(Arg::with_name("port") .short("p") .long("port") diff --git a/clients/socks5/src/client/config/template.rs b/clients/socks5/src/client/config/template.rs index 57e2f2ae15..b1412db7e0 100644 --- a/clients/socks5/src/client/config/template.rs +++ b/clients/socks5/src/client/config/template.rs @@ -29,11 +29,6 @@ validator_rest_urls = [ # Address of the validator contract managing the network. mixnet_contract_address = '{{ client.mixnet_contract_address }}' -# Special mode of the system such that all messages are sent as soon as they are received -# and no cover traffic is generated. If set all message delays are set to 0 and overwriting -# 'Debug' values will have no effect. -vpn_mode = {{ client.vpn_mode }} - # Path to file containing private identity key. private_identity_key_file = '{{ client.private_identity_key_file }}' diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index aedd6eb265..7adf77c9e1 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -32,7 +32,6 @@ use gateway_client::{ use log::*; use nymsphinx::addressing::clients::Recipient; use nymsphinx::addressing::nodes::NodeIdentity; -use nymsphinx::params::PacketMode; use tokio::runtime::Runtime; pub(crate) mod config; @@ -107,12 +106,6 @@ impl NymClient { input_receiver: InputMessageReceiver, mix_sender: BatchMixMessageSender, ) { - let packet_mode = if self.config.get_base().get_vpn_mode() { - PacketMode::Vpn - } else { - PacketMode::Mix - }; - let controller_config = client_core::client::real_messages_control::Config::new( self.key_manager.ack_key(), self.config.get_base().get_ack_wait_multiplier(), @@ -121,8 +114,6 @@ impl NymClient { self.config.get_base().get_message_sending_average_delay(), self.config.get_base().get_average_packet_delay(), self.as_mix_recipient(), - packet_mode, - self.config.get_base().get_vpn_key_reuse_limit(), ); info!("Starting real traffic stream..."); @@ -139,7 +130,7 @@ impl NymClient { topology_accessor, reply_key_storage, ) - .start(self.runtime.handle(), self.config.get_base().get_vpn_mode()); + .start(self.runtime.handle()); } // buffer controlling all messages fetched from provider @@ -329,9 +320,8 @@ impl NymClient { input_receiver, sphinx_message_sender.clone(), ); - if !self.config.get_base().get_vpn_mode() { - self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender); - } + + self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender); self.start_socks5_listener(received_buffer_request_sender, input_sender); info!("Client startup finished!"); diff --git a/clients/socks5/src/commands/init.rs b/clients/socks5/src/commands/init.rs index 7889dbcbf5..956cb01618 100644 --- a/clients/socks5/src/commands/init.rs +++ b/clients/socks5/src/commands/init.rs @@ -54,17 +54,6 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> { .help("Port for the socket to listen on in all subsequent runs") .takes_value(true) ) - .arg(Arg::with_name("vpn-mode") - .long("vpn-mode") - .help("Set the vpn mode of the client") - .long_help( - r#" - Special mode of the system such that all messages are sent as soon as they are received - and no cover traffic is generated. If set all message delays are set to 0 and overwriting - 'Debug' values will have no effect. - "# - ) - ) .arg(Arg::with_name("fastmode") .long("fastmode") .hidden(true) // this will prevent this flag from being displayed in `--help` diff --git a/clients/socks5/src/commands/mod.rs b/clients/socks5/src/commands/mod.rs index 88dbd40ed3..eefd717e2c 100644 --- a/clients/socks5/src/commands/mod.rs +++ b/clients/socks5/src/commands/mod.rs @@ -29,10 +29,6 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi config.get_base_mut().with_gateway_id(gateway_id); } - if matches.is_present("vpn-mode") { - config.get_base_mut().set_vpn_mode(true); - } - if let Some(port) = matches.value_of("port").map(|port| port.parse::()) { if let Err(err) = port { // if port was overridden, it must be parsable diff --git a/clients/socks5/src/commands/run.rs b/clients/socks5/src/commands/run.rs index fe606a1380..173a6e4856 100644 --- a/clients/socks5/src/commands/run.rs +++ b/clients/socks5/src/commands/run.rs @@ -44,17 +44,6 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> { .help("Id of the gateway we want to connect to. If overridden, it is user's responsibility to ensure prior registration happened") .takes_value(true) ) - .arg(Arg::with_name("vpn-mode") - .long("vpn-mode") - .help("Set the vpn mode of the client") - .long_help( - r#" - Special mode of the system such that all messages are sent as soon as they are received - and no cover traffic is generated. If set all message delays are set to 0 and overwriting - 'Debug' values will have no effect. - "# - ) - ) .arg(Arg::with_name("port") .short("p") .long("port") diff --git a/clients/webassembly/src/client/mod.rs b/clients/webassembly/src/client/mod.rs index 1c0f7ab0b1..e00e6f85d5 100644 --- a/clients/webassembly/src/client/mod.rs +++ b/clients/webassembly/src/client/mod.rs @@ -22,8 +22,6 @@ pub(crate) mod received_processor; const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200); const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200); const DEFAULT_GATEWAY_RESPONSE_TIMEOUT: Duration = Duration::from_millis(1_500); -const DEFAULT_PACKET_MODE: PacketMode = PacketMode::Vpn; -const DEFAULT_VPN_KEY_REUSE_LIMIT: usize = 1000; #[wasm_bindgen] pub struct NymClient { @@ -139,8 +137,6 @@ impl NymClient { client.self_recipient(), DEFAULT_AVERAGE_PACKET_DELAY, DEFAULT_AVERAGE_ACK_DELAY, - DEFAULT_PACKET_MODE, - Some(DEFAULT_VPN_KEY_REUSE_LIMIT), ); let received_processor = ReceivedMessagesProcessor::new( diff --git a/common/mixnode-common/src/cached_packet_processor/cache.rs b/common/mixnode-common/src/cached_packet_processor/cache.rs deleted file mode 100644 index c55ce834af..0000000000 --- a/common/mixnode-common/src/cached_packet_processor/cache.rs +++ /dev/null @@ -1,170 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use dashmap::mapref::one::Ref; -use dashmap::DashMap; -use futures::channel::mpsc; -use futures::StreamExt; -use log::*; -use nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue, TimerError}; -use nymsphinx_types::header::keys::RoutingKeys; -use nymsphinx_types::SharedSecret; -use std::sync::Arc; -use tokio::time::Duration; - -type CachedKeys = (Option, RoutingKeys); - -pub(super) struct KeyCache { - vpn_key_cache: Arc>, - invalidator_sender: InvalidatorActionSender, - cache_entry_ttl: Duration, -} - -impl Drop for KeyCache { - fn drop(&mut self) { - debug!("dropping key cache"); - if self - .invalidator_sender - .unbounded_send(InvalidatorAction::Stop) - .is_err() - { - debug!("invalidator has already been dropped") - } - } -} - -impl KeyCache { - pub(super) fn new(cache_entry_ttl: Duration) -> Self { - let cache = Arc::new(DashMap::new()); - let (sender, receiver) = mpsc::unbounded(); - - let mut invalidator = CacheInvalidator { - entry_ttl: cache_entry_ttl, - vpn_key_cache: Arc::clone(&cache), - expirations: NonExhaustiveDelayQueue::new(), - action_receiver: receiver, - }; - - // TODO: is it possible to avoid tokio::spawn here and make it semi-runtime agnostic? - tokio::spawn(async move { invalidator.run().await }); - - KeyCache { - vpn_key_cache: cache, - invalidator_sender: sender, - cache_entry_ttl, - } - } - - pub(super) fn insert(&self, key: SharedSecret, cached_keys: CachedKeys) -> bool { - trace!("inserting {:?} into the cache", key); - let insertion_result = self.vpn_key_cache.insert(key, cached_keys).is_some(); - if !insertion_result { - debug!("{:?} was put into the cache", key); - // this shouldn't really happen, but don't insert entry to invalidator if it was already - // in the cache - self.invalidator_sender - .unbounded_send(InvalidatorAction::Insert(key)) - .expect("Cache invalidator has crashed!"); - } - insertion_result - } - - // ElementGuard has Deref for CachedKeys so that's fine - pub(super) fn get(&self, key: &SharedSecret) -> Option> { - self.vpn_key_cache.get(key) - } - - pub(super) fn cache_entry_ttl(&self) -> Duration { - self.cache_entry_ttl - } - - #[cfg(test)] - pub(super) fn is_empty(&self) -> bool { - self.vpn_key_cache.is_empty() - } - - #[cfg(test)] - pub(super) fn len(&self) -> usize { - self.vpn_key_cache.len() - } -} - -enum InvalidatorAction { - Insert(SharedSecret), - Stop, -} - -type InvalidatorActionSender = mpsc::UnboundedSender; -type InvalidatorActionReceiver = mpsc::UnboundedReceiver; - -struct CacheInvalidator { - entry_ttl: Duration, - vpn_key_cache: Arc>, - expirations: NonExhaustiveDelayQueue, - action_receiver: InvalidatorActionReceiver, -} - -// we do not have a strong requirement of invalidating things EXACTLY after their TTL expires. -// we want them to be eventually gone in a relatively timely manner. -impl CacheInvalidator { - // two obvious ways I've seen of running this were as follows: - // - // 1) every X second, purge all expired entries - // pros: simpler to implement - // cons: will require to obtain write lock multiple times in quick succession - // - // 2) purge entry as soon as it expires - // pros: the lock situation will be spread more in time - // cons: possibly less efficient? - - fn handle_expired(&mut self, expired: Option, TimerError>>) { - let expired = expired.expect("the queue has unexpectedly terminated!"); - let expired_entry = expired.expect("Encountered timer issue within the runtime!"); - - debug!( - "{:?} has expired and will be removed", - expired_entry.get_ref() - ); - - if self - .vpn_key_cache - .remove(&expired_entry.into_inner()) - .is_none() - { - error!("Tried to remove vpn cache entry for non-existent key!") - } - } - - /// Handles received action. Return `bool` indicates whether the invalidator - /// should terminate. - fn handle_action(&mut self, action: Option) -> bool { - if action.is_none() { - return true; - } - - match action.unwrap() { - InvalidatorAction::Stop => true, - InvalidatorAction::Insert(shared_secret) => { - self.expirations.insert(shared_secret, self.entry_ttl); - false - } - } - } - - async fn run(&mut self) { - loop { - tokio::select! { - expired = self.expirations.next() => { - self.handle_expired(expired); - } - action = self.action_receiver.next() => { - if self.handle_action(action) { - info!("Stopping cache invalidator"); - return - } - } - - } - } - } -} diff --git a/common/mixnode-common/src/cached_packet_processor/processor.rs b/common/mixnode-common/src/cached_packet_processor/processor.rs deleted file mode 100644 index d8eca98e65..0000000000 --- a/common/mixnode-common/src/cached_packet_processor/processor.rs +++ /dev/null @@ -1,475 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::cached_packet_processor::cache::KeyCache; -use crate::cached_packet_processor::error::MixProcessingError; -use log::*; -use nymsphinx_acknowledgements::surb_ack::SurbAck; -use nymsphinx_addressing::nodes::NymNodeRoutingAddress; -use nymsphinx_forwarding::packet::MixPacket; -use nymsphinx_framing::packet::FramedSphinxPacket; -use nymsphinx_params::{PacketMode, PacketSize}; -use nymsphinx_types::header::keys::RoutingKeys; -use nymsphinx_types::{ - Delay as SphinxDelay, DestinationAddressBytes, NodeAddressBytes, Payload, PrivateKey, - ProcessedPacket, SharedSecret, SphinxHeader, SphinxPacket, -}; -use std::convert::TryFrom; -use std::sync::Arc; -use tokio::time::Duration; - -type ForwardAck = MixPacket; -type CachedKeys = (Option, RoutingKeys); - -pub struct ProcessedFinalHop { - pub destination: DestinationAddressBytes, - pub forward_ack: Option, - pub message: Vec, -} - -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), -} - -pub struct CachedPacketProcessor { - /// Private sphinx key of this node required to unwrap received sphinx packet. - sphinx_key: Arc, - - /// Key cache containing derived shared keys for packets using `vpn_mode`. - // Note: as discovered this is potentially unsafe as security of Lioness depends on keys never being reused. - // So perhaps it should get completely disabled for time being? - vpn_key_cache: KeyCache, -} - -impl CachedPacketProcessor { - /// Creates new instance of `CachedPacketProcessor` - pub fn new(sphinx_key: PrivateKey, cache_entry_ttl: Duration) -> Self { - CachedPacketProcessor { - sphinx_key: Arc::new(sphinx_key), - vpn_key_cache: KeyCache::new(cache_entry_ttl), - } - } - - /// Clones `self` without the `vpn_key_cache`. - pub fn clone_without_cache(&self) -> Self { - CachedPacketProcessor { - sphinx_key: self.sphinx_key.clone(), - vpn_key_cache: KeyCache::new(self.vpn_key_cache.cache_entry_ttl()), - } - } - - /// Recomputes routing keys for the given initial secret. - fn recompute_routing_keys(&self, initial_secret: &SharedSecret) -> RoutingKeys { - SphinxHeader::compute_routing_keys(initial_secret, &self.sphinx_key) - } - - /// Performs a fresh sphinx unwrapping using no cache. - fn perform_initial_sphinx_packet_processing( - &self, - packet: SphinxPacket, - ) -> Result { - packet.process(&self.sphinx_key).map_err(|err| { - debug!("Failed to unwrap Sphinx packet: {:?}", err); - MixProcessingError::SphinxProcessingError(err) - }) - } - - /// Unwraps sphinx packet using already cached keys. - fn perform_initial_sphinx_packet_processing_with_cached_keys( - &self, - packet: SphinxPacket, - keys: &CachedKeys, - ) -> Result { - packet - .process_with_derived_keys(&keys.0, &keys.1) - .map_err(|err| { - debug!("Failed to unwrap Sphinx packet: {:?}", err); - MixProcessingError::SphinxProcessingError(err) - }) - } - - /// Stores the keys corresponding to the packet that was just processed. - fn cache_keys(&self, initial_secret: SharedSecret, processed_packet: &ProcessedPacket) { - let new_shared_secret = processed_packet.shared_secret(); - let routing_keys = self.recompute_routing_keys(&initial_secret); - if self - .vpn_key_cache - .insert(initial_secret, (new_shared_secret, routing_keys)) - { - debug!("Other thread has already cached keys for this secret!") - } - } - - /// Takes the received framed packet and tries to unwrap it from the sphinx encryption. - /// For any vpn packets it will try to re-use cached keys and if none are available, - /// after first processing, the keys are going to get cached. - fn perform_initial_unwrapping( - &self, - received: FramedSphinxPacket, - ) -> Result { - let packet_mode = received.packet_mode(); - let sphinx_packet = received.into_inner(); - let initial_secret = sphinx_packet.shared_secret(); - - // try to use pre-computed keys only for the vpn-packets - if packet_mode.is_vpn() { - if let Some(cached_keys) = self.vpn_key_cache.get(&initial_secret) { - return self.perform_initial_sphinx_packet_processing_with_cached_keys( - sphinx_packet, - cached_keys.value(), - ); - } - } - - let processing_result = self.perform_initial_sphinx_packet_processing(sphinx_packet); - // quicker exit because this will be the most common case - if !packet_mode.is_vpn() { - return processing_result; - } - - if let Ok(processed_packet) = processing_result.as_ref() { - // if we managed to process packet we saw for the first time AND it's a vpn packet - // cache the keys - self.cache_keys(initial_secret, processed_packet); - } - processing_result - } - - /// Processed received forward hop packet - tries to extract next hop address, sets delay, - /// if it was not a vpn packet and packs all the data in a way that can be easily sent - /// to the next hop. - fn process_forward_hop( - &self, - packet: SphinxPacket, - forward_address: NodeAddressBytes, - delay: SphinxDelay, - packet_mode: PacketMode, - ) -> Result { - let next_hop_address = NymNodeRoutingAddress::try_from(forward_address)?; - - // if the packet is set to vpn mode, ignore whatever might have been set as delay - let delay = if packet_mode.is_vpn() { - None - } else { - Some(delay) - }; - - let mix_packet = MixPacket::new(next_hop_address, packet, packet_mode); - Ok(MixProcessingResult::ForwardHop(mix_packet, 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, - ) -> Result<(Vec, Vec), MixProcessingError> { - // 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() < SurbAck::len() { - return Err(MixProcessingError::NoSurbAckInFinalHop); - } - - let message = extracted_data.split_off(SurbAck::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_mode: PacketMode, - ) -> Result<(Option, Vec), MixProcessingError> { - match packet_size { - PacketSize::AckPacket => { - trace!("received an ack packet!"); - Ok((None, data)) - } - PacketSize::RegularPacket | PacketSize::ExtendedPacket => { - trace!("received a normal packet!"); - let (ack_data, message) = self.split_hop_data_into_ack_and_message(data)?; - let (ack_first_hop, ack_packet) = SurbAck::try_recover_first_hop_packet(&ack_data)?; - let forward_ack = MixPacket::new(ack_first_hop, ack_packet, packet_mode); - 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: Payload, - packet_size: PacketSize, - packet_mode: PacketMode, - ) -> Result { - let packet_message = payload.recover_plaintext()?; - - let (forward_ack, message) = - self.split_into_ack_and_message(packet_message, packet_size, packet_mode)?; - - 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: ProcessedPacket, - packet_size: PacketSize, - packet_mode: PacketMode, - ) -> Result { - match packet { - ProcessedPacket::ForwardHop(packet, address, delay) => { - self.process_forward_hop(packet, address, delay, packet_mode) - } - // 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, packet_size, packet_mode) - } - } - } - - pub fn process_received( - &self, - received: FramedSphinxPacket, - ) -> Result { - // explicit packet size will help to correctly parse final hop - let packet_size = received.packet_size(); - let packet_mode = received.packet_mode(); - - // 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 - self.perform_final_processing(processed_packet, packet_size, packet_mode) - } -} - -// TODO: what more could we realistically test here? -#[cfg(test)] -mod tests { - use super::*; - use nymsphinx_types::builder::SphinxPacketBuilder; - use nymsphinx_types::crypto::keygen; - use nymsphinx_types::{ - Destination, Node, PublicKey, DESTINATION_ADDRESS_LENGTH, IDENTIFIER_LENGTH, - }; - use std::convert::TryInto; - use std::net::SocketAddr; - - fn fixture() -> CachedPacketProcessor { - let local_keys = keygen(); - CachedPacketProcessor::new(local_keys.0, Duration::from_secs(30)) - } - - fn make_valid_final_sphinx_packet(size: PacketSize, public_key: PublicKey) -> SphinxPacket { - let routing_address: NymNodeRoutingAddress = - NymNodeRoutingAddress::from("127.0.0.1:1789".parse::().unwrap()); - - let node = Node::new(routing_address.try_into().unwrap(), public_key); - - let destination = Destination::new( - DestinationAddressBytes::from_bytes([3u8; DESTINATION_ADDRESS_LENGTH]), - [4u8; IDENTIFIER_LENGTH], - ); - - // required until https://github.com/nymtech/sphinx/issues/71 is fixed - let dummy_delay = SphinxDelay::new_from_nanos(42); - - SphinxPacketBuilder::new() - .with_payload_size(size.payload_size()) - .build_packet(b"foomp".to_vec(), &[node], &destination, &[dummy_delay]) - .unwrap() - } - - fn make_valid_forward_sphinx_packet(size: PacketSize, public_key: PublicKey) -> SphinxPacket { - let routing_address: NymNodeRoutingAddress = - NymNodeRoutingAddress::from("127.0.0.1:1789".parse::().unwrap()); - - let some_node_key = keygen(); - let route = [ - Node::new(routing_address.try_into().unwrap(), public_key), - Node::new(routing_address.try_into().unwrap(), some_node_key.1), - ]; - - let destination = Destination::new( - DestinationAddressBytes::from_bytes([3u8; DESTINATION_ADDRESS_LENGTH]), - [4u8; IDENTIFIER_LENGTH], - ); - - let delays = [ - SphinxDelay::new_from_nanos(42), - SphinxDelay::new_from_nanos(42), - ]; - - SphinxPacketBuilder::new() - .with_payload_size(size.payload_size()) - .build_packet(b"foomp".to_vec(), &route, &destination, &delays) - .unwrap() - } - - #[tokio::test] - async fn recomputing_routing_keys_derives_correct_set_of_keys() { - let processor = fixture(); - let (_, initial_secret) = keygen(); - assert_eq!( - processor.recompute_routing_keys(&initial_secret), - SphinxHeader::compute_routing_keys(&initial_secret, &processor.sphinx_key) - ) - } - - #[tokio::test] - async fn caching_keys_updates_local_state_for_final_hop() { - let local_keys = keygen(); - let processor = CachedPacketProcessor::new(local_keys.0, Duration::from_secs(30)); - assert!(processor.vpn_key_cache.is_empty()); - - let final_hop = make_valid_final_sphinx_packet(Default::default(), local_keys.1); - let initial_secret = final_hop.shared_secret(); - let processed = final_hop.process(&processor.sphinx_key).unwrap(); - - processor.cache_keys(initial_secret, &processed); - let cache_entry = processor.vpn_key_cache.get(&initial_secret).unwrap(); - - let (cached_secret, cached_routing_keys) = cache_entry.value(); - - assert!(cached_secret.is_none()); - let recomputed_keys = processor.recompute_routing_keys(&initial_secret); - // if one key matches then all keys must match (or there is a serious bug inside sphinx) - assert_eq!( - cached_routing_keys.stream_cipher_key, - recomputed_keys.stream_cipher_key - ); - } - - #[tokio::test] - async fn caching_keys_updates_local_state_for_forward_hop() { - let local_keys = keygen(); - let processor = CachedPacketProcessor::new(local_keys.0, Duration::from_secs(30)); - assert!(processor.vpn_key_cache.is_empty()); - - let forward_hop = make_valid_forward_sphinx_packet(Default::default(), local_keys.1); - let initial_secret = forward_hop.shared_secret(); - let processed = forward_hop.process(&processor.sphinx_key).unwrap(); - - processor.cache_keys(initial_secret, &processed); - let cache_entry = processor.vpn_key_cache.get(&initial_secret).unwrap(); - - let (cached_secret, cached_routing_keys) = cache_entry.value(); - - assert_eq!( - cached_secret.as_ref().unwrap(), - processed.shared_secret().as_ref().unwrap() - ); - let recomputed_keys = processor.recompute_routing_keys(&initial_secret); - // if one key matches then all keys must match (or there is a serious bug inside sphinx) - assert_eq!( - cached_routing_keys.stream_cipher_key, - recomputed_keys.stream_cipher_key - ); - } - - #[tokio::test] - async fn performing_initial_unwrapping_caches_keys_if_vpnmode_used_for_final_hop() { - let local_keys = keygen(); - let processor = CachedPacketProcessor::new(local_keys.0, Duration::from_secs(30)); - assert!(processor.vpn_key_cache.is_empty()); - - let final_hop = make_valid_final_sphinx_packet(Default::default(), local_keys.1); - let framed = FramedSphinxPacket::new(final_hop, PacketMode::Vpn); - - processor.perform_initial_unwrapping(framed).unwrap(); - assert_eq!(processor.vpn_key_cache.len(), 1); - } - - #[tokio::test] - async fn performing_initial_unwrapping_caches_keys_if_vpnmode_used_for_forward_hop() { - let local_keys = keygen(); - let processor = CachedPacketProcessor::new(local_keys.0, Duration::from_secs(30)); - assert!(processor.vpn_key_cache.is_empty()); - - let forward_hop = make_valid_forward_sphinx_packet(Default::default(), local_keys.1); - let framed = FramedSphinxPacket::new(forward_hop, PacketMode::Vpn); - - processor.perform_initial_unwrapping(framed).unwrap(); - assert_eq!(processor.vpn_key_cache.len(), 1); - } - - #[tokio::test] - async fn performing_initial_unwrapping_does_no_caching_for_mix_mode_for_final_hop() { - let local_keys = keygen(); - let processor = CachedPacketProcessor::new(local_keys.0, Duration::from_secs(30)); - assert!(processor.vpn_key_cache.is_empty()); - - let final_hop = make_valid_final_sphinx_packet(Default::default(), local_keys.1); - let framed = FramedSphinxPacket::new(final_hop, PacketMode::Mix); - - processor.perform_initial_unwrapping(framed).unwrap(); - assert!(processor.vpn_key_cache.is_empty()); - } - - #[tokio::test] - async fn performing_initial_unwrapping_does_no_caching_for_mix_mode_for_forward_hop() { - let local_keys = keygen(); - let processor = CachedPacketProcessor::new(local_keys.0, Duration::from_secs(30)); - assert!(processor.vpn_key_cache.is_empty()); - - let forward_hop = make_valid_forward_sphinx_packet(Default::default(), local_keys.1); - let framed = FramedSphinxPacket::new(forward_hop, PacketMode::Mix); - - processor.perform_initial_unwrapping(framed).unwrap(); - assert!(processor.vpn_key_cache.is_empty()); - } - - #[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) - .is_err()); - - let sufficient_data = vec![42u8; SurbAck::len()]; - let (ack, data) = processor - .split_hop_data_into_ack_and_message(sufficient_data.clone()) - .unwrap(); - assert_eq!(sufficient_data, ack); - assert!(data.is_empty()); - - let long_data = vec![42u8; SurbAck::len() * 5]; - let (ack, data) = processor - .split_hop_data_into_ack_and_message(long_data) - .unwrap(); - assert_eq!(ack.len(), SurbAck::len()); - assert_eq!(data.len(), SurbAck::len() * 4) - } - - #[tokio::test] - async fn splitting_into_ack_and_message_returns_whole_data_for_ack() { - let processor = fixture(); - - let data = vec![42u8; SurbAck::len() + 10]; - let (ack, message) = processor - .split_into_ack_and_message(data.clone(), PacketSize::AckPacket, Default::default()) - .unwrap(); - assert!(ack.is_none()); - assert_eq!(data, message) - } -} diff --git a/common/mixnode-common/src/lib.rs b/common/mixnode-common/src/lib.rs index 7aa13cc7e6..34d43fc458 100644 --- a/common/mixnode-common/src/lib.rs +++ b/common/mixnode-common/src/lib.rs @@ -1,5 +1,5 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -pub mod cached_packet_processor; +pub mod packet_processor; pub mod verloc; diff --git a/common/mixnode-common/src/cached_packet_processor/error.rs b/common/mixnode-common/src/packet_processor/error.rs similarity index 91% rename from common/mixnode-common/src/cached_packet_processor/error.rs rename to common/mixnode-common/src/packet_processor/error.rs index 4078f8a2ae..38be92079d 100644 --- a/common/mixnode-common/src/cached_packet_processor/error.rs +++ b/common/mixnode-common/src/packet_processor/error.rs @@ -12,6 +12,8 @@ pub enum MixProcessingError { InvalidHopAddress(NymNodeRoutingAddressError), NoSurbAckInFinalHop, MalformedSurbAck(SurbAckRecoveryError), + + ReceivedOldTypeVpnPacket, } impl From for MixProcessingError { @@ -54,6 +56,9 @@ impl Display for MixProcessingError { MixProcessingError::MalformedSurbAck(surb_ack_err) => { write!(f, "Malformed SURBAck - {:?}", surb_ack_err) } + MixProcessingError::ReceivedOldTypeVpnPacket => { + write!(f, "Received an old-type unsafe 'VPN' mode packet") + } } } } diff --git a/common/mixnode-common/src/cached_packet_processor/mod.rs b/common/mixnode-common/src/packet_processor/mod.rs similarity index 92% rename from common/mixnode-common/src/cached_packet_processor/mod.rs rename to common/mixnode-common/src/packet_processor/mod.rs index 0d85c0954e..3d2efa6b2c 100644 --- a/common/mixnode-common/src/cached_packet_processor/mod.rs +++ b/common/mixnode-common/src/packet_processor/mod.rs @@ -1,6 +1,5 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -mod cache; pub mod error; pub mod processor; diff --git a/common/mixnode-common/src/packet_processor/processor.rs b/common/mixnode-common/src/packet_processor/processor.rs new file mode 100644 index 0000000000..344172dca4 --- /dev/null +++ b/common/mixnode-common/src/packet_processor/processor.rs @@ -0,0 +1,234 @@ +// Copyright 2021 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::packet_processor::error::MixProcessingError; +use log::*; +use nymsphinx_acknowledgements::surb_ack::SurbAck; +use nymsphinx_addressing::nodes::NymNodeRoutingAddress; +use nymsphinx_forwarding::packet::MixPacket; +use nymsphinx_framing::packet::FramedSphinxPacket; +use nymsphinx_params::{PacketMode, PacketSize}; +use nymsphinx_types::{ + Delay as SphinxDelay, DestinationAddressBytes, NodeAddressBytes, Payload, PrivateKey, + ProcessedPacket, SphinxPacket, +}; +use std::convert::TryFrom; +use std::sync::Arc; + +type ForwardAck = MixPacket; + +pub struct ProcessedFinalHop { + pub destination: DestinationAddressBytes, + pub forward_ack: Option, + pub message: Vec, +} + +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), +} + +#[derive(Clone)] +pub struct SphinxPacketProcessor { + /// Private sphinx key of this node required to unwrap received sphinx packet. + sphinx_key: Arc, +} + +impl SphinxPacketProcessor { + /// Creates new instance of `CachedPacketProcessor` + pub fn new(sphinx_key: PrivateKey) -> Self { + SphinxPacketProcessor { + sphinx_key: Arc::new(sphinx_key), + } + } + + /// Performs a fresh sphinx unwrapping using no cache. + fn perform_initial_sphinx_packet_processing( + &self, + packet: SphinxPacket, + ) -> Result { + packet.process(&self.sphinx_key).map_err(|err| { + debug!("Failed to unwrap Sphinx packet: {:?}", err); + MixProcessingError::SphinxProcessingError(err) + }) + } + + /// Takes the received framed packet and tries to unwrap it from the sphinx encryption. + fn perform_initial_unwrapping( + &self, + received: FramedSphinxPacket, + ) -> Result { + let packet_mode = received.packet_mode(); + let sphinx_packet = received.into_inner(); + + if packet_mode.is_old_vpn() { + return Err(MixProcessingError::ReceivedOldTypeVpnPacket); + } + + self.perform_initial_sphinx_packet_processing(sphinx_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: SphinxPacket, + forward_address: NodeAddressBytes, + delay: SphinxDelay, + packet_mode: PacketMode, + ) -> Result { + let next_hop_address = NymNodeRoutingAddress::try_from(forward_address)?; + + let mix_packet = MixPacket::new(next_hop_address, packet, packet_mode); + 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, + ) -> Result<(Vec, Vec), MixProcessingError> { + // 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() < SurbAck::len() { + return Err(MixProcessingError::NoSurbAckInFinalHop); + } + + let message = extracted_data.split_off(SurbAck::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_mode: PacketMode, + ) -> Result<(Option, Vec), MixProcessingError> { + match packet_size { + PacketSize::AckPacket => { + trace!("received an ack packet!"); + Ok((None, data)) + } + PacketSize::RegularPacket | PacketSize::ExtendedPacket => { + trace!("received a normal packet!"); + let (ack_data, message) = self.split_hop_data_into_ack_and_message(data)?; + let (ack_first_hop, ack_packet) = SurbAck::try_recover_first_hop_packet(&ack_data)?; + let forward_ack = MixPacket::new(ack_first_hop, ack_packet, packet_mode); + 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: Payload, + packet_size: PacketSize, + packet_mode: PacketMode, + ) -> Result { + let packet_message = payload.recover_plaintext()?; + + let (forward_ack, message) = + self.split_into_ack_and_message(packet_message, packet_size, packet_mode)?; + + 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: ProcessedPacket, + packet_size: PacketSize, + packet_mode: PacketMode, + ) -> Result { + match packet { + ProcessedPacket::ForwardHop(packet, address, delay) => { + self.process_forward_hop(packet, address, delay, packet_mode) + } + // 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, packet_size, packet_mode) + } + } + } + + pub fn process_received( + &self, + received: FramedSphinxPacket, + ) -> Result { + // explicit packet size will help to correctly parse final hop + let packet_size = received.packet_size(); + let packet_mode = received.packet_mode(); + + // 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 + self.perform_final_processing(processed_packet, packet_size, packet_mode) + } +} + +// TODO: what more could we realistically test here? +#[cfg(test)] +mod tests { + use super::*; + use nymsphinx_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) + .is_err()); + + let sufficient_data = vec![42u8; SurbAck::len()]; + let (ack, data) = processor + .split_hop_data_into_ack_and_message(sufficient_data.clone()) + .unwrap(); + assert_eq!(sufficient_data, ack); + assert!(data.is_empty()); + + let long_data = vec![42u8; SurbAck::len() * 5]; + let (ack, data) = processor + .split_hop_data_into_ack_and_message(long_data) + .unwrap(); + assert_eq!(ack.len(), SurbAck::len()); + assert_eq!(data.len(), SurbAck::len() * 4) + } + + #[tokio::test] + async fn splitting_into_ack_and_message_returns_whole_data_for_ack() { + let processor = fixture(); + + let data = vec![42u8; SurbAck::len() + 10]; + let (ack, message) = processor + .split_into_ack_and_message(data.clone(), PacketSize::AckPacket, Default::default()) + .unwrap(); + assert!(ack.is_none()); + assert_eq!(data, message) + } +} diff --git a/common/nymsphinx/acknowledgements/src/surb_ack.rs b/common/nymsphinx/acknowledgements/src/surb_ack.rs index 4bac6cbd05..22bd4bc468 100644 --- a/common/nymsphinx/acknowledgements/src/surb_ack.rs +++ b/common/nymsphinx/acknowledgements/src/surb_ack.rs @@ -10,7 +10,7 @@ use nymsphinx_params::DEFAULT_NUM_MIX_HOPS; use nymsphinx_types::builder::SphinxPacketBuilder; use nymsphinx_types::{ delays::{self, Delay}, - EphemeralSecret, SphinxPacket, + SphinxPacket, }; use rand::{CryptoRng, RngCore}; use std::convert::TryFrom; @@ -38,7 +38,6 @@ impl SurbAck { marshaled_fragment_id: [u8; 5], average_delay: time::Duration, topology: &NymTopology, - initial_sphinx_secret: Option<&EphemeralSecret>, ) -> Result where R: RngCore + CryptoRng, @@ -50,13 +49,8 @@ impl SurbAck { let surb_ack_payload = prepare_identifier(rng, ack_key, marshaled_fragment_id); - let mut surb_builder = - SphinxPacketBuilder::new().with_payload_size(PacketSize::AckPacket.payload_size()); - if let Some(initial_secret) = initial_sphinx_secret { - surb_builder = surb_builder.with_initial_secret(initial_secret); - } - - let surb_ack_packet = surb_builder + let surb_ack_packet = SphinxPacketBuilder::new() + .with_payload_size(PacketSize::AckPacket.payload_size()) .build_packet(surb_ack_payload, &route, &destination, &delays) .unwrap(); diff --git a/common/nymsphinx/cover/src/lib.rs b/common/nymsphinx/cover/src/lib.rs index 5a6cce9212..6b78559ac3 100644 --- a/common/nymsphinx/cover/src/lib.rs +++ b/common/nymsphinx/cover/src/lib.rs @@ -66,7 +66,6 @@ where COVER_FRAG_ID.to_bytes(), average_ack_delay, topology, - None, )?) } @@ -137,7 +136,6 @@ where let first_hop_address = NymNodeRoutingAddress::try_from(route.first().unwrap().address).unwrap(); - // if client is running in vpn mode, he won't even be sending cover traffic Ok(MixPacket::new(first_hop_address, packet, PacketMode::Mix)) } diff --git a/common/nymsphinx/framing/src/packet.rs b/common/nymsphinx/framing/src/packet.rs index f1db17b91c..80d190e7a0 100644 --- a/common/nymsphinx/framing/src/packet.rs +++ b/common/nymsphinx/framing/src/packet.rs @@ -58,6 +58,8 @@ pub struct Header { /// /// TODO: ask @AP whether this can be sent like this - could it introduce some anonymity issues? /// (note: this will be behind some encryption, either something implemented by us or some SSL action) + // Note: currently packet_mode is deprecated but is still left as a concept behind to not break + // compatibility with existing network pub(crate) packet_mode: PacketMode, } diff --git a/common/nymsphinx/params/src/packet_modes.rs b/common/nymsphinx/params/src/packet_modes.rs index bc634647ed..0031d8084c 100644 --- a/common/nymsphinx/params/src/packet_modes.rs +++ b/common/nymsphinx/params/src/packet_modes.rs @@ -23,7 +23,7 @@ impl PacketMode { self == PacketMode::Mix } - pub fn is_vpn(self) -> bool { + pub fn is_old_vpn(self) -> bool { self == PacketMode::Vpn } } diff --git a/common/nymsphinx/src/preparer/mod.rs b/common/nymsphinx/src/preparer/mod.rs index 250f109dc2..06ead921fe 100644 --- a/common/nymsphinx/src/preparer/mod.rs +++ b/common/nymsphinx/src/preparer/mod.rs @@ -1,7 +1,6 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use self::vpn_manager::VpnManager; use crate::chunking; use crypto::asymmetric::encryption; use crypto::shared_key::new_ephemeral_shared_key; @@ -17,7 +16,7 @@ use nymsphinx_chunking::fragment::{Fragment, FragmentIdentifier}; use nymsphinx_forwarding::packet::MixPacket; use nymsphinx_params::packet_sizes::PacketSize; use nymsphinx_params::{ - PacketEncryptionAlgorithm, PacketHkdfAlgorithm, PacketMode, ReplySurbEncryptionAlgorithm, + PacketEncryptionAlgorithm, PacketHkdfAlgorithm, ReplySurbEncryptionAlgorithm, ReplySurbKeyDigestAlgorithm, DEFAULT_NUM_MIX_HOPS, }; use nymsphinx_types::builder::SphinxPacketBuilder; @@ -27,8 +26,6 @@ use std::convert::TryFrom; use std::time::Duration; use topology::{NymTopology, NymTopologyError}; -mod vpn_manager; - /// Represents fully packed and prepared [`Fragment`] that can be sent through the mix network. pub struct PreparedFragment { /// Indicates the total expected round-trip time, i.e. delay from the sending of this message @@ -77,14 +74,6 @@ pub struct MessagePreparer { /// Number of mix hops each packet ('real' message, ack, reply) is expected to take. /// Note that it does not include gateway hops. num_mix_hops: u8, - - /// Mode of all mix packets created - VPN or Mix. They indicate whether packets should get delayed - /// and keys reused. - mode: PacketMode, - - /// If the VPN mode is activated, this underlying secret will be used for multiple sphinx - /// packets created. - vpn_manager: Option, } impl MessagePreparer @@ -92,22 +81,11 @@ where R: CryptoRng + Rng, { pub fn new( - mut rng: R, + rng: R, sender_address: Recipient, average_packet_delay: Duration, average_ack_delay: Duration, - mode: PacketMode, - vpn_key_reuse_limit: Option, ) -> Self { - let vpn_manager = if mode.is_vpn() { - Some(VpnManager::new( - &mut rng, - vpn_key_reuse_limit.expect("No key reuse limit provided in vpn mode!"), - )) - } else { - None - }; - MessagePreparer { rng, packet_size: Default::default(), @@ -115,8 +93,6 @@ where average_packet_delay, average_ack_delay, num_mix_hops: DEFAULT_NUM_MIX_HOPS, - mode, - vpn_manager, } } @@ -298,20 +274,10 @@ where // create the actual sphinx packet here. With valid route and correct payload size, // there's absolutely no reason for this call to fail. - let sphinx_packet = if let Some(vpn_manager) = self.vpn_manager.as_mut() { - let initial_secret = vpn_manager.use_secret(&mut self.rng).await; - - SphinxPacketBuilder::new() - .with_payload_size(self.packet_size.payload_size()) - .with_initial_secret(&initial_secret) - .build_packet(packet_payload, &route, &destination, &delays) - .unwrap() - } else { - SphinxPacketBuilder::new() - .with_payload_size(self.packet_size.payload_size()) - .build_packet(packet_payload, &route, &destination, &delays) - .unwrap() - }; + let sphinx_packet = SphinxPacketBuilder::new() + .with_payload_size(self.packet_size.payload_size()) + .build_packet(packet_payload, &route, &destination, &delays) + .unwrap(); // from the previously constructed route extract the first hop let first_hop_address = @@ -322,7 +288,7 @@ where // well as the total delay of the ack packet. // note that the last hop of the packet is a gateway that does not do any delays total_delay: delays.iter().take(delays.len() - 1).sum::() + ack_delay, - mix_packet: MixPacket::new(first_hop_address, sphinx_packet, self.mode), + mix_packet: MixPacket::new(first_hop_address, sphinx_packet, Default::default()), }) } @@ -333,28 +299,14 @@ where topology: &NymTopology, ack_key: &AckKey, ) -> Result { - if let Some(vpn_manager) = self.vpn_manager.as_mut() { - let initial_secret = vpn_manager.use_secret(&mut self.rng).await; - SurbAck::construct( - &mut self.rng, - &self.sender_address, - ack_key, - fragment_id.to_bytes(), - self.average_ack_delay, - topology, - Some(&initial_secret), - ) - } else { - SurbAck::construct( - &mut self.rng, - &self.sender_address, - ack_key, - fragment_id.to_bytes(), - self.average_ack_delay, - topology, - None, - ) - } + SurbAck::construct( + &mut self.rng, + &self.sender_address, + ack_key, + fragment_id.to_bytes(), + self.average_ack_delay, + topology, + ) } /// Attaches an optional reply-surb and correct padding to the underlying message @@ -451,7 +403,10 @@ where .apply_surb(&packet_payload, Some(self.packet_size)) .unwrap(); - Ok((MixPacket::new(first_hop, packet, self.mode), reply_id)) + Ok(( + MixPacket::new(first_hop, packet, Default::default()), + reply_id, + )) } #[allow(dead_code)] @@ -467,8 +422,6 @@ where average_packet_delay: Default::default(), average_ack_delay: Default::default(), num_mix_hops: DEFAULT_NUM_MIX_HOPS, - mode: Default::default(), - vpn_manager: None, } } } diff --git a/common/nymsphinx/src/preparer/vpn_manager.rs b/common/nymsphinx/src/preparer/vpn_manager.rs deleted file mode 100644 index 06a6e11c28..0000000000 --- a/common/nymsphinx/src/preparer/vpn_manager.rs +++ /dev/null @@ -1,143 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use nymsphinx_types::EphemeralSecret; -use rand::{CryptoRng, Rng}; -use std::sync::atomic::{AtomicUsize, Ordering}; -#[cfg(not(target_arch = "wasm32"))] -use std::sync::Arc; -#[cfg(not(target_arch = "wasm32"))] -use tokio::sync::{RwLock, RwLockReadGuard}; - -#[cfg(not(target_arch = "wasm32"))] -pub(super) type SpinhxKeyRef<'a> = RwLockReadGuard<'a, EphemeralSecret>; - -#[cfg(target_arch = "wasm32")] -pub(super) type SpinhxKeyRef<'a> = &'a EphemeralSecret; - -#[cfg_attr(not(target_arch = "wasm32"), derive(Clone))] -pub(super) struct VpnManager { - #[cfg(not(target_arch = "wasm32"))] - inner: Arc, - - #[cfg(target_arch = "wasm32")] - inner: Inner, -} - -struct Inner { - /// Maximum number of times particular sphinx-secret can be re-used before being rotated. - secret_reuse_limit: usize, - - /// Currently used initial sphinx-secret for the packets sent. - #[cfg(not(target_arch = "wasm32"))] - current_initial_secret: RwLock, - - #[cfg(target_arch = "wasm32")] - // this is a temporary work-around for wasm (which currently does not have retransmission - // and hence will not require multi-thread access) and also we can't import tokio's RWLock - // in wasm. - current_initial_secret: EphemeralSecret, - - /// If the client is running as VPN it's expected to keep re-using the same initial secret - /// for a while so that the mixnodes could cache some secret derivation results. However, - /// we should reset it every once in a while. - packets_with_current_secret: AtomicUsize, -} - -impl VpnManager { - #[cfg(not(target_arch = "wasm32"))] - pub(super) fn new(mut rng: R, secret_reuse_limit: usize) -> Self - where - R: CryptoRng + Rng, - { - let initial_secret = EphemeralSecret::new_with_rng(&mut rng); - VpnManager { - inner: Arc::new(Inner { - secret_reuse_limit, - current_initial_secret: RwLock::new(initial_secret), - packets_with_current_secret: AtomicUsize::new(0), - }), - } - } - - #[cfg(target_arch = "wasm32")] - pub(super) fn new(mut rng: R, secret_reuse_limit: usize) -> Self - where - R: CryptoRng + Rng, - { - let initial_secret = EphemeralSecret::new_with_rng(&mut rng); - VpnManager { - inner: Inner { - secret_reuse_limit, - current_initial_secret: initial_secret, - packets_with_current_secret: AtomicUsize::new(0), - }, - } - } - - #[cfg(not(target_arch = "wasm32"))] - pub(super) async fn rotate_secret(&mut self, mut rng: R) - where - R: CryptoRng + Rng, - { - let new_secret = EphemeralSecret::new_with_rng(&mut rng); - let mut write_guard = self.inner.current_initial_secret.write().await; - - *write_guard = new_secret; - // in here we have an exclusive lock so we don't have to have restrictive ordering as no - // other thread will be able to get here - self.inner - .packets_with_current_secret - .store(0, Ordering::Relaxed) - } - - // this method is async for consistency with non-wasm version - #[cfg(target_arch = "wasm32")] - pub(super) async fn rotate_secret(&mut self, mut rng: R) - where - R: CryptoRng + Rng, - { - let new_secret = EphemeralSecret::new_with_rng(&mut rng); - self.inner.current_initial_secret = new_secret; - - // wasm is single-threaded so relaxed ordering is also fine here - self.inner - .packets_with_current_secret - .store(0, Ordering::Relaxed); - } - - #[cfg(not(target_arch = "wasm32"))] - pub(super) async fn current_secret(&self) -> SpinhxKeyRef<'_> { - self.inner.current_initial_secret.read().await - } - - #[cfg(target_arch = "wasm32")] - pub(super) async fn current_secret(&self) -> SpinhxKeyRef<'_> { - &self.inner.current_initial_secret - } - - fn increment_key_usage(&mut self) { - // TODO: is this the appropriate ordering? - self.inner - .packets_with_current_secret - .fetch_add(1, Ordering::SeqCst); - } - - fn current_key_usage(&self) -> usize { - // TODO: is this the appropriate ordering? - self.inner - .packets_with_current_secret - .load(Ordering::SeqCst) - } - - pub(super) async fn use_secret(&mut self, rng: R) -> SpinhxKeyRef<'_> - where - R: CryptoRng + Rng, - { - if self.current_key_usage() > self.inner.secret_reuse_limit { - self.rotate_secret(rng).await; - } - self.increment_key_usage(); - self.current_secret().await - } -} diff --git a/gateway/src/config/mod.rs b/gateway/src/config/mod.rs index d263d6f423..dc7461f517 100644 --- a/gateway/src/config/mod.rs +++ b/gateway/src/config/mod.rs @@ -21,7 +21,6 @@ const DEFAULT_PRESENCE_SENDING_DELAY: Duration = Duration::from_millis(10_000); const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: Duration = Duration::from_millis(10_000); const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: Duration = Duration::from_millis(300_000); const DEFAULT_INITIAL_CONNECTION_TIMEOUT: Duration = Duration::from_millis(1_500); -const DEFAULT_CACHE_ENTRY_TTL: Duration = Duration::from_millis(30_000); const DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE: usize = 128; const DEFAULT_STORED_MESSAGE_FILENAME_LENGTH: u16 = 16; @@ -279,10 +278,6 @@ impl Config { self.debug.stored_messages_filename_length } - pub fn get_cache_entry_ttl(&self) -> Duration { - self.debug.cache_entry_ttl - } - pub fn get_version(&self) -> &str { &self.gateway.version } @@ -466,13 +461,6 @@ pub struct Debug { /// if there are no real messages, dummy ones are created to always return /// `message_retrieval_limit` total messages message_retrieval_limit: u16, - - /// Duration for which a cached vpn processing result is going to get stored for. - #[serde( - deserialize_with = "deserialize_duration", - serialize_with = "humantime_serde::serialize" - )] - cache_entry_ttl: Duration, } impl Default for Debug { @@ -485,7 +473,6 @@ impl Default for Debug { maximum_connection_buffer_size: DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE, stored_messages_filename_length: DEFAULT_STORED_MESSAGE_FILENAME_LENGTH, message_retrieval_limit: DEFAULT_MESSAGE_RETRIEVAL_LIMIT, - cache_entry_ttl: DEFAULT_CACHE_ENTRY_TTL, } } } diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs index 8975abcebe..8a3731b291 100644 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs @@ -12,7 +12,7 @@ use futures::channel::oneshot; use futures::StreamExt; use log::*; use mixnet_client::forwarder::MixForwardingSender; -use mixnode_common::cached_packet_processor::processor::ProcessedFinalHop; +use mixnode_common::packet_processor::processor::ProcessedFinalHop; use nymsphinx::forwarding::packet::MixPacket; use nymsphinx::framing::codec::SphinxCodec; use nymsphinx::framing::packet::FramedSphinxPacket; @@ -64,7 +64,7 @@ impl ConnectionHandler { } ConnectionHandler { - packet_processor: self.packet_processor.clone_without_key_cache(), + packet_processor: self.packet_processor.clone(), available_socket_senders_cache: senders_cache, client_store: self.client_store.clone(), clients_handler_sender: self.clients_handler_sender.clone(), diff --git a/gateway/src/node/mixnet_handling/receiver/packet_processing.rs b/gateway/src/node/mixnet_handling/receiver/packet_processing.rs index 063c3f11c3..af9970cb47 100644 --- a/gateway/src/node/mixnet_handling/receiver/packet_processing.rs +++ b/gateway/src/node/mixnet_handling/receiver/packet_processing.rs @@ -2,13 +2,10 @@ // SPDX-License-Identifier: Apache-2.0 use crypto::asymmetric::encryption; -use mixnode_common::cached_packet_processor::error::MixProcessingError; -pub use mixnode_common::cached_packet_processor::processor::MixProcessingResult; -use mixnode_common::cached_packet_processor::processor::{ - CachedPacketProcessor, ProcessedFinalHop, -}; +use mixnode_common::packet_processor::error::MixProcessingError; +pub use mixnode_common::packet_processor::processor::MixProcessingResult; +use mixnode_common::packet_processor::processor::{ProcessedFinalHop, SphinxPacketProcessor}; use nymsphinx::framing::packet::FramedSphinxPacket; -use tokio::time::Duration; #[derive(Debug)] pub enum GatewayProcessingError { @@ -25,20 +22,15 @@ impl From for GatewayProcessingError { } // PacketProcessor contains all data required to correctly unwrap and store sphinx packets +#[derive(Clone)] pub struct PacketProcessor { - inner_processor: CachedPacketProcessor, + inner_processor: SphinxPacketProcessor, } impl PacketProcessor { - pub(crate) fn new(encryption_key: &encryption::PrivateKey, cache_entry_ttl: Duration) -> Self { + pub(crate) fn new(encryption_key: &encryption::PrivateKey) -> Self { PacketProcessor { - inner_processor: CachedPacketProcessor::new(encryption_key.into(), cache_entry_ttl), - } - } - - pub(crate) fn clone_without_key_cache(&self) -> Self { - PacketProcessor { - inner_processor: self.inner_processor.clone_without_cache(), + inner_processor: SphinxPacketProcessor::new(encryption_key.into()), } } diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 7981b12030..fc619a35a2 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -59,10 +59,8 @@ impl Gateway { ) { info!("Starting mix socket listener..."); - let packet_processor = mixnet_handling::PacketProcessor::new( - self.encryption_keys.private_key(), - self.config.get_cache_entry_ttl(), - ); + let packet_processor = + mixnet_handling::PacketProcessor::new(self.encryption_keys.private_key()); let connection_handler = ConnectionHandler::new( packet_processor, diff --git a/mixnode/src/config/mod.rs b/mixnode/src/config/mod.rs index 967535af6b..78da10801f 100644 --- a/mixnode/src/config/mod.rs +++ b/mixnode/src/config/mod.rs @@ -30,7 +30,6 @@ const DEFAULT_NODE_STATS_UPDATING_DELAY: Duration = Duration::from_millis(30_000 const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: Duration = Duration::from_millis(10_000); const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: Duration = Duration::from_millis(300_000); const DEFAULT_INITIAL_CONNECTION_TIMEOUT: Duration = Duration::from_millis(1_500); -const DEFAULT_CACHE_ENTRY_TTL: Duration = Duration::from_millis(30_000); const DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE: usize = 128; // helper function to get default validators as a Vec @@ -286,10 +285,6 @@ impl Config { self.debug.maximum_connection_buffer_size } - pub fn get_cache_entry_ttl(&self) -> Duration { - self.debug.cache_entry_ttl - } - pub fn get_version(&self) -> &str { &self.mixnode.version } @@ -526,13 +521,6 @@ pub struct Debug { /// Maximum number of packets that can be stored waiting to get sent to a particular connection. maximum_connection_buffer_size: usize, - - /// Duration for which a cached vpn processing result is going to get stored for. - #[serde( - deserialize_with = "deserialize_duration", - serialize_with = "humantime_serde::serialize" - )] - cache_entry_ttl: Duration, } impl Default for Debug { @@ -544,7 +532,6 @@ impl Default for Debug { packet_forwarding_maximum_backoff: DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF, initial_connection_timeout: DEFAULT_INITIAL_CONNECTION_TIMEOUT, maximum_connection_buffer_size: DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE, - cache_entry_ttl: DEFAULT_CACHE_ENTRY_TTL, } } } diff --git a/mixnode/src/node/listener/connection_handler/mod.rs b/mixnode/src/node/listener/connection_handler/mod.rs index ee73a7eacd..3aff6ad3a8 100644 --- a/mixnode/src/node/listener/connection_handler/mod.rs +++ b/mixnode/src/node/listener/connection_handler/mod.rs @@ -18,6 +18,7 @@ use tokio_util::codec::Framed; pub(crate) mod packet_processing; +#[derive(Clone)] pub(crate) struct ConnectionHandler { packet_processor: PacketProcessor, delay_forwarding_channel: PacketDelayForwardSender, @@ -34,13 +35,6 @@ impl ConnectionHandler { } } - pub(crate) fn clone_without_cache(&self) -> Self { - ConnectionHandler { - packet_processor: self.packet_processor.clone_without_cache(), - delay_forwarding_channel: self.delay_forwarding_channel.clone(), - } - } - 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 diff --git a/mixnode/src/node/listener/connection_handler/packet_processing.rs b/mixnode/src/node/listener/connection_handler/packet_processing.rs index 93fb655dea..ac080a249a 100644 --- a/mixnode/src/node/listener/connection_handler/packet_processing.rs +++ b/mixnode/src/node/listener/connection_handler/packet_processing.rs @@ -3,16 +3,16 @@ use crate::node::node_statistics; use crypto::asymmetric::encryption; -use mixnode_common::cached_packet_processor::error::MixProcessingError; -use mixnode_common::cached_packet_processor::processor::CachedPacketProcessor; -pub use mixnode_common::cached_packet_processor::processor::MixProcessingResult; +use mixnode_common::packet_processor::error::MixProcessingError; +pub use mixnode_common::packet_processor::processor::MixProcessingResult; +use mixnode_common::packet_processor::processor::SphinxPacketProcessor; use nymsphinx::framing::packet::FramedSphinxPacket; -use tokio::time::Duration; // PacketProcessor contains all data required to correctly unwrap and forward sphinx packets +#[derive(Clone)] pub struct PacketProcessor { /// Responsible for performing unwrapping - inner_processor: CachedPacketProcessor, + inner_processor: SphinxPacketProcessor, /// Responsible for updating metrics data node_stats_update_sender: node_statistics::UpdateSender, @@ -22,21 +22,13 @@ impl PacketProcessor { pub(crate) fn new( encryption_key: &encryption::PrivateKey, node_stats_update_sender: node_statistics::UpdateSender, - cache_entry_ttl: Duration, ) -> Self { PacketProcessor { - inner_processor: CachedPacketProcessor::new(encryption_key.into(), cache_entry_ttl), + inner_processor: SphinxPacketProcessor::new(encryption_key.into()), node_stats_update_sender, } } - pub(crate) fn clone_without_cache(&self) -> Self { - PacketProcessor { - inner_processor: self.inner_processor.clone_without_cache(), - node_stats_update_sender: self.node_stats_update_sender.clone(), - } - } - pub(crate) fn process_received( &self, received: FramedSphinxPacket, diff --git a/mixnode/src/node/listener/mod.rs b/mixnode/src/node/listener/mod.rs index 68a7c19b63..e24e951838 100644 --- a/mixnode/src/node/listener/mod.rs +++ b/mixnode/src/node/listener/mod.rs @@ -31,7 +31,7 @@ impl Listener { loop { match listener.accept().await { Ok((socket, remote_addr)) => { - let handler = connection_handler.clone_without_cache(); + let handler = connection_handler.clone(); tokio::spawn(handler.handle_connection(socket, remote_addr)); } Err(err) => warn!("Failed to accept incoming connection - {:?}", err), diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index b2a18ead9a..2a58c14b63 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -101,11 +101,8 @@ impl MixNode { ) { info!("Starting socket listener..."); - let packet_processor = PacketProcessor::new( - self.sphinx_keypair.private_key(), - node_stats_update_sender, - self.config.get_cache_entry_ttl(), - ); + let packet_processor = + PacketProcessor::new(self.sphinx_keypair.private_key(), node_stats_update_sender); let connection_handler = ConnectionHandler::new(packet_processor, delay_forwarding_channel); diff --git a/validator-api/src/network_monitor/chunker.rs b/validator-api/src/network_monitor/chunker.rs index 0b7472bd69..a5732600f0 100644 --- a/validator-api/src/network_monitor/chunker.rs +++ b/validator-api/src/network_monitor/chunker.rs @@ -2,7 +2,6 @@ // SPDX-License-Identifier: Apache-2.0 use nymsphinx::forwarding::packet::MixPacket; -use nymsphinx::params::PacketMode; use nymsphinx::{ acknowledgements::AckKey, addressing::clients::Recipient, preparer::MessagePreparer, }; @@ -27,8 +26,6 @@ impl Chunker { tested_mix_me, DEFAULT_AVERAGE_PACKET_DELAY, DEFAULT_AVERAGE_ACK_DELAY, - PacketMode::Mix, - None, ), } }