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

Removed all sphinx key caching from mixnodes and gateways (#713)

Browse Source 
* Removed all sphinx key caching from mixnodes and gateways

* Missing change in network monitor
This commit is contained in:
Jędrzej Stuczyński
2021-08-11 12:47:35 +01:00
committed by GitHub
parent fdd34863ba
commit fb253e53a4
37 changed files with 301 additions and 1151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs
-15
View File
@@ -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<usize>,
}
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<usize>,
) -> 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
clients/client-core/src/client/real_messages_control/mod.rs
+4 -22
View File
@@ -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<usize>,
}
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<AckKey>,
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<usize>,
) -> 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<OsRng> {
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<OsRng> {
}
}
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<OsRng> {
// 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<OsRng> {
// &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<Self> {
pub fn start(mut self, handle: &Handle) -> JoinHandle<Self> {
handle.spawn(async move {
self.run(vpn_mode).await;
self.run().await;
self
})
}
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+3 -24
View File
@@ -249,15 +249,6 @@ where
tokio::task::yield_now().await;
}
async fn on_batch_received(&mut self, real_messages: Vec<RealMessage>) {
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
}
}
clients/client-core/src/config/mod.rs
+3 -53
View File
@@ -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<String>
pub fn default_validator_rest_endpoints() -> Vec<String> {
@@ -139,14 +136,6 @@ impl<T: NymConfig> Config<T> {
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<T: NymConfig> Config<T> {
// 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<T: NymConfig> Config<T> {
}
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<T: NymConfig> Config<T> {
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<usize> {
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<T> {
#[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<T: NymConfig> Default for Client<T> {
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<usize>,
}
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,
}
}
}
clients/native/src/client/config/template.rs
-5
View File
@@ -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 }}'
clients/native/src/client/mod.rs
+2 -13
View File
@@ -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 => {
clients/native/src/commands/init.rs
-11
View File
@@ -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`
clients/native/src/commands/mod.rs
-4
View File
@@ -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::<u16>()) {
if let Err(err) = port {
// if port was overridden, it must be parsable
clients/native/src/commands/run.rs
-11
View File
@@ -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")
clients/socks5/src/client/config/template.rs
-5
View File
@@ -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 }}'
clients/socks5/src/client/mod.rs
+3 -13
View File
@@ -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!");
clients/socks5/src/commands/init.rs
-11
View File
@@ -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`
clients/socks5/src/commands/mod.rs
-4
View File
@@ -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::<u16>()) {
if let Err(err) = port {
// if port was overridden, it must be parsable
clients/socks5/src/commands/run.rs
-11
View File
@@ -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")
clients/webassembly/src/client/mod.rs
-4
View File
@@ -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(
common/mixnode-common/src/cached_packet_processor/cache.rs
-170
View File
@@ -1,170 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<SharedSecret>, RoutingKeys);
pub(super) struct KeyCache {
vpn_key_cache: Arc<DashMap<SharedSecret, CachedKeys>>,
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<Ref<SharedSecret, CachedKeys>> {
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<InvalidatorAction>;
type InvalidatorActionReceiver = mpsc::UnboundedReceiver<InvalidatorAction>;
struct CacheInvalidator {
entry_ttl: Duration,
vpn_key_cache: Arc<DashMap<SharedSecret, CachedKeys>>,
expirations: NonExhaustiveDelayQueue<SharedSecret>,
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<Result<Expired<SharedSecret>, 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<InvalidatorAction>) -> 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
}
}
}
}
}
}
common/mixnode-common/src/cached_packet_processor/processor.rs
-475
View File
@@ -1,475 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<SharedSecret>, RoutingKeys);
pub struct ProcessedFinalHop {
pub destination: DestinationAddressBytes,
pub forward_ack: Option<ForwardAck>,
pub message: Vec<u8>,
}
pub enum MixProcessingResult {
/// Contains unwrapped data that should first get delayed before being sent to next hop.
ForwardHop(MixPacket, Option<SphinxDelay>),
/// 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<PrivateKey>,
/// 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<ProcessedPacket, MixProcessingError> {
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<ProcessedPacket, MixProcessingError> {
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<ProcessedPacket, MixProcessingError> {
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<MixProcessingResult, MixProcessingError> {
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<u8>,
) -> Result<(Vec<u8>, Vec<u8>), 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<u8>,
packet_size: PacketSize,
packet_mode: PacketMode,
) -> Result<(Option<MixPacket>, Vec<u8>), 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<MixProcessingResult, MixProcessingError> {
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<MixProcessingResult, MixProcessingError> {
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<MixProcessingResult, MixProcessingError> {
// 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::<SocketAddr>().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::<SocketAddr>().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)
}
}
common/mixnode-common/src/lib.rs
+1 -1
View File
@@ -1,5 +1,5 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod cached_packet_processor;
pub mod packet_processor;
pub mod verloc;
common/mixnode-common/src/cached_packet_processor/error.rs → common/mixnode-common/src/packet_processor/error.rs
+5
View File
@@ -12,6 +12,8 @@ pub enum MixProcessingError {
InvalidHopAddress(NymNodeRoutingAddressError),
NoSurbAckInFinalHop,
MalformedSurbAck(SurbAckRecoveryError),
ReceivedOldTypeVpnPacket,
}
impl From<SphinxError> 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")
}
}
}
}
common/mixnode-common/src/cached_packet_processor/mod.rs → common/mixnode-common/src/packet_processor/mod.rs
-1
View File
@@ -1,6 +1,5 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
mod cache;
pub mod error;
pub mod processor;
common/mixnode-common/src/packet_processor/processor.rs
+234
View File
@@ -0,0 +1,234 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<ForwardAck>,
pub message: Vec<u8>,
}
pub enum MixProcessingResult {
/// Contains unwrapped data that should first get delayed before being sent to next hop.
ForwardHop(MixPacket, Option<SphinxDelay>),
/// 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<PrivateKey>,
}
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<ProcessedPacket, MixProcessingError> {
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<ProcessedPacket, MixProcessingError> {
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<MixProcessingResult, MixProcessingError> {
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<u8>,
) -> Result<(Vec<u8>, Vec<u8>), 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<u8>,
packet_size: PacketSize,
packet_mode: PacketMode,
) -> Result<(Option<MixPacket>, Vec<u8>), 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<MixProcessingResult, MixProcessingError> {
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<MixProcessingResult, MixProcessingError> {
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<MixProcessingResult, MixProcessingError> {
// 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)
}
}
common/nymsphinx/acknowledgements/src/surb_ack.rs
+3 -9
View File
@@ -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<Self, NymTopologyError>
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();
common/nymsphinx/cover/src/lib.rs
-2
View File
@@ -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))
}
common/nymsphinx/framing/src/packet.rs
+2
View File
@@ -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,
}
common/nymsphinx/params/src/packet_modes.rs
+1 -1
View File
@@ -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
}
}
common/nymsphinx/src/preparer/mod.rs
+19 -66
View File
@@ -1,7 +1,6 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<R: CryptoRng + Rng> {
/// 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<VpnManager>,
}
impl<R> MessagePreparer<R>
@@ -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<usize>,
) -> 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::<Delay>() + 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<SurbAck, NymTopologyError> {
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,
}
}
}
common/nymsphinx/src/preparer/vpn_manager.rs
-143
View File
@@ -1,143 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<Inner>,
#[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<EphemeralSecret>,
#[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<R>(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<R>(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<R>(&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<R>(&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<R>(&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
}
}
gateway/src/config/mod.rs
-13
View File
@@ -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,
}
}
}
gateway/src/node/mixnet_handling/receiver/connection_handler.rs
+2 -2
View File
@@ -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(),
gateway/src/node/mixnet_handling/receiver/packet_processing.rs
+7 -15
View File
@@ -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<MixProcessingError> 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()),
}
}
gateway/src/node/mod.rs
+2 -4
View File
@@ -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,
mixnode/src/config/mod.rs
-13
View File
@@ -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<String>
@@ -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,
}
}
}
mixnode/src/node/listener/connection_handler/mod.rs
+1 -7
View File
@@ -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<SphinxDelay>) {
// 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
mixnode/src/node/listener/connection_handler/packet_processing.rs
+6 -14
View File
@@ -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,
mixnode/src/node/listener/mod.rs
+1 -1
View File
@@ -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),
mixnode/src/node/mod.rs
+2 -5
View File
@@ -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);
validator-api/src/network_monitor/chunker.rs
-3
View File
@@ -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,
),
}
}