From f64cfb4cd12939b1e99c61fb77c732f74489ad50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Tue, 18 Apr 2023 16:18:48 +0100 Subject: [PATCH 1/5] wip --- Cargo.lock | 12 + Cargo.toml | 1 + clients/webassembly/src/client/helpers.rs | 19 +- .../src/tester/ephemeral_receiver.rs | 22 +- clients/webassembly/src/tester/helpers.rs | 6 +- clients/webassembly/src/tester/mod.rs | 9 +- common/node-tester-utils/src/error.rs | 3 + common/node-tester-utils/src/lib.rs | 2 + common/node-tester-utils/src/message.rs | 86 ++- common/node-tester-utils/src/node.rs | 82 +++ common/node-tester-utils/src/processor.rs | 95 ++++ common/node-tester-utils/src/receiver.rs | 86 +-- common/node-tester-utils/src/tester.rs | 122 ++++- common/nymsphinx/Cargo.toml | 1 + common/nymsphinx/routing/Cargo.toml | 14 + common/nymsphinx/routing/src/lib.rs | 43 ++ common/nymsphinx/src/lib.rs | 6 +- common/nymsphinx/src/preparer/mod.rs | 6 + common/topology/Cargo.toml | 1 + common/topology/src/error.rs | 36 ++ common/topology/src/lib.rs | 43 +- common/topology/src/random_route_provider.rs | 29 ++ nym-api/Cargo.toml | 1 + nym-api/src/network_monitor/chunker.rs | 1 + .../src/network_monitor/gateways_reader.rs | 27 +- .../src/network_monitor/monitor/preparer.rs | 491 ++++++++++-------- .../src/network_monitor/monitor/processor.rs | 105 ++-- .../src/network_monitor/monitor/receiver.rs | 8 +- .../monitor/summary_producer.rs | 10 +- nym-api/src/network_monitor/test_packet.rs | 264 +--------- nym-api/src/network_monitor/test_route/mod.rs | 47 +- 31 files changed, 958 insertions(+), 720 deletions(-) create mode 100644 common/node-tester-utils/src/node.rs create mode 100644 common/node-tester-utils/src/processor.rs create mode 100644 common/nymsphinx/routing/Cargo.toml create mode 100644 common/nymsphinx/routing/src/lib.rs create mode 100644 common/topology/src/error.rs create mode 100644 common/topology/src/random_route_provider.rs diff --git a/Cargo.lock b/Cargo.lock index f09f9a563a..f1608ac9cc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3047,6 +3047,7 @@ dependencies = [ "nym-inclusion-probability", "nym-mixnet-contract-common", "nym-multisig-contract-common", + "nym-node-tester-utils", "nym-pemstore", "nym-sphinx", "nym-task", @@ -3993,6 +3994,7 @@ dependencies = [ "nym-sphinx-forwarding", "nym-sphinx-framing", "nym-sphinx-params", + "nym-sphinx-routing", "nym-sphinx-types", "nym-topology", "rand 0.7.3", @@ -4102,6 +4104,15 @@ dependencies = [ "thiserror", ] +[[package]] +name = "nym-sphinx-routing" +version = "0.1.0" +dependencies = [ + "nym-sphinx-addressing", + "nym-sphinx-types", + "thiserror", +] + [[package]] name = "nym-sphinx-types" version = "0.2.0" @@ -4146,6 +4157,7 @@ dependencies = [ "nym-crypto", "nym-mixnet-contract-common", "nym-sphinx-addressing", + "nym-sphinx-routing", "nym-sphinx-types", "rand 0.7.3", "thiserror", diff --git a/Cargo.toml b/Cargo.toml index e8dd22ee65..48ec1151c1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,6 +60,7 @@ members = [ "common/nymsphinx/forwarding", "common/nymsphinx/framing", "common/nymsphinx/params", + "common/nymsphinx/routing", "common/nymsphinx/types", "common/pemstore", "common/socks5-client-core", diff --git a/clients/webassembly/src/client/helpers.rs b/clients/webassembly/src/client/helpers.rs index 5e62d88939..c3f456a891 100644 --- a/clients/webassembly/src/client/helpers.rs +++ b/clients/webassembly/src/client/helpers.rs @@ -13,7 +13,7 @@ use std::sync::Arc; use wasm_bindgen::prelude::*; use wasm_bindgen::JsValue; use wasm_bindgen_futures::future_to_promise; -use wasm_utils::{console_log, js_error, simple_js_error}; +use wasm_utils::{console_log, simple_js_error}; #[wasm_bindgen] pub struct NymClientTestRequest { @@ -142,20 +142,9 @@ impl WasmTopologyExt for Arc { return Err(WasmClientError::NonExistentMixnode { mixnode_identity }.into()); }; - let mut test_msgs = Vec::with_capacity(num_test_packets as usize); - for i in 1..=num_test_packets { - let msg = NodeTestMessage::new_mix( - mix, - i, - num_test_packets, - WasmTestMessageExt::new(test_id), - ); - let serialized = match msg.as_bytes() { - Ok(bytes) => bytes, - Err(err) => return Err(js_error!("failed to serialize test message: {err}")), - }; - test_msgs.push(serialized); - } + let ext = WasmTestMessageExt::new(test_id); + let test_msgs = NodeTestMessage::mix_plaintexts(mix, num_test_packets, ext) + .map_err(WasmClientError::from)?; let mut updated = current_topology.clone(); updated.set_mixes_in_layer(mix.layer.into(), vec![mix.to_owned()]); diff --git a/clients/webassembly/src/tester/ephemeral_receiver.rs b/clients/webassembly/src/tester/ephemeral_receiver.rs index 5bbba9aa19..1403176b36 100644 --- a/clients/webassembly/src/tester/ephemeral_receiver.rs +++ b/clients/webassembly/src/tester/ephemeral_receiver.rs @@ -1,8 +1,7 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::tester::helpers::NodeTestResult; -use crate::tester::NodeTestMessage; +use crate::tester::helpers::{NodeTestResult, WasmTestMessageExt}; use futures::StreamExt; use nym_node_tester_utils::receiver::{Received, ReceivedReceiver}; use nym_sphinx::chunking::fragment::FragmentIdentifier; @@ -21,7 +20,7 @@ pub(crate) struct EphemeralTestReceiver<'a> { duplicate_acks: u32, timeout_duration: Duration, - receiver_permit: AsyncMutexGuard<'a, ReceivedReceiver>, + receiver_permit: AsyncMutexGuard<'a, ReceivedReceiver>, } impl<'a> EphemeralTestReceiver<'a> { @@ -38,7 +37,7 @@ impl<'a> EphemeralTestReceiver<'a> { pub(crate) fn new( sent_packets: u32, expected_acks: HashSet, - receiver_permit: AsyncMutexGuard<'a, ReceivedReceiver>, + receiver_permit: AsyncMutexGuard<'a, ReceivedReceiver>, timeout: Duration, ) -> Self { EphemeralTestReceiver { @@ -53,23 +52,18 @@ impl<'a> EphemeralTestReceiver<'a> { } } - fn on_next_received_packet(&mut self, packet: Option) -> bool { + fn on_next_received_packet(&mut self, packet: Option>) -> bool { let Some(received_packet) = packet else { // can't do anything more... console_error!("packet receiver has stopped processing results!"); return true }; match received_packet { - Received::Message(msg) => match NodeTestMessage::try_recover(msg) { - Ok(test_msg) => { - if !self.received_valid_messages.insert(test_msg.msg_id) { - self.duplicate_packets += 1; - } + Received::Message(msg) => { + if !self.received_valid_messages.insert(msg.msg_id) { + self.duplicate_packets += 1; } - Err(err) => { - console_warn!("failed to recover test message from received packet: {err}") - } - }, + } Received::Ack(frag_id) => { if self.expected_acks.contains(&frag_id) { if !self.received_valid_acks.insert(frag_id) { diff --git a/clients/webassembly/src/tester/helpers.rs b/clients/webassembly/src/tester/helpers.rs index fea5367f80..59269b6921 100644 --- a/clients/webassembly/src/tester/helpers.rs +++ b/clients/webassembly/src/tester/helpers.rs @@ -14,10 +14,10 @@ use wasm_bindgen::prelude::*; use wasm_utils::{console_log, console_warn}; #[derive(Clone)] -pub(super) struct ReceivedReceiverWrapper(Arc>); +pub(super) struct ReceivedReceiverWrapper(Arc>>); impl ReceivedReceiverWrapper { - pub(super) fn new(inner: ReceivedReceiver) -> Self { + pub(super) fn new(inner: ReceivedReceiver) -> Self { ReceivedReceiverWrapper(Arc::new(AsyncMutex::new(inner))) } @@ -36,7 +36,7 @@ impl ReceivedReceiverWrapper { } } - pub(super) async fn lock(&self) -> AsyncMutexGuard<'_, ReceivedReceiver> { + pub(super) async fn lock(&self) -> AsyncMutexGuard<'_, ReceivedReceiver> { self.0.lock().await } } diff --git a/clients/webassembly/src/tester/mod.rs b/clients/webassembly/src/tester/mod.rs index 8295d2799f..3fdb67297d 100644 --- a/clients/webassembly/src/tester/mod.rs +++ b/clients/webassembly/src/tester/mod.rs @@ -162,7 +162,7 @@ impl NymNodeTesterBuilder { let tester = NodeTester::new( rng, self.base_topology, - address(&self.key_manager, gateway_identity), + Some(address(&self.key_manager, gateway_identity)), PacketSize::default(), Duration::from_millis(5), Duration::from_millis(5), @@ -266,7 +266,12 @@ impl NymNodeTester { let test_ext = WasmTestMessageExt::new(test_nonce); let mut tester_permit = self.tester.lock().expect("mutex got poisoned"); tester_permit - .existing_identity_mixnode_test_packets(mixnode_identity, test_ext, num_test_packets) + .existing_identity_mixnode_test_packets( + mixnode_identity, + test_ext, + num_test_packets, + None, + ) .map_err(Into::into) } diff --git a/common/node-tester-utils/src/error.rs b/common/node-tester-utils/src/error.rs index 264d2c128c..31bd77f98f 100644 --- a/common/node-tester-utils/src/error.rs +++ b/common/node-tester-utils/src/error.rs @@ -46,4 +46,7 @@ pub enum NetworkTestingError { #[error("received a packet that could not be reconstructed into a full message with a single fragment")] NonReconstructablePacket, + + #[error("the recipient of the test packet was never specified")] + UnknownPacketRecipient, } diff --git a/common/node-tester-utils/src/lib.rs b/common/node-tester-utils/src/lib.rs index 87f584cfd4..a5e304c584 100644 --- a/common/node-tester-utils/src/lib.rs +++ b/common/node-tester-utils/src/lib.rs @@ -3,6 +3,8 @@ pub mod error; pub mod message; +pub mod node; +pub mod processor; pub mod receiver; pub mod tester; diff --git a/common/node-tester-utils/src/message.rs b/common/node-tester-utils/src/message.rs index 25f2973b93..c4b63d753c 100644 --- a/common/node-tester-utils/src/message.rs +++ b/common/node-tester-utils/src/message.rs @@ -2,27 +2,19 @@ // SPDX-License-Identifier: Apache-2.0 use crate::error::NetworkTestingError; -use crate::MixId; +use crate::node::TestableNode; use nym_sphinx::message::NymMessage; use nym_topology::{gateway, mix}; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use std::hash::{Hash, Hasher}; -#[derive(Serialize, Deserialize, Hash, Clone, Copy)] -pub enum NodeType { - Mixnode(MixId), - Gateway, -} - #[derive(Serialize, Deserialize, Hash, Clone, Copy)] pub struct Empty; #[derive(Serialize, Deserialize, Clone)] pub struct TestMessage { - pub encoded_node_identity: String, - pub node_owner: String, - pub node_type: NodeType, + pub tested_node: TestableNode, pub msg_id: u32, pub total_msgs: u32, @@ -34,26 +26,72 @@ pub struct TestMessage { } impl TestMessage { - pub fn new_mix(node: &mix::Node, msg_id: u32, total_msgs: u32, ext: T) -> Self { + pub fn new>(node: N, msg_id: u32, total_msgs: u32, ext: T) -> Self { TestMessage { - encoded_node_identity: node.identity_key.to_base58_string(), - node_owner: node.owner.clone(), - node_type: NodeType::Mixnode(node.mix_id), + tested_node: node.into(), msg_id, total_msgs, ext, } } + pub fn new_mix(node: &mix::Node, msg_id: u32, total_msgs: u32, ext: T) -> Self { + Self::new(node, msg_id, total_msgs, ext) + } + pub fn new_gateway(node: &gateway::Node, msg_id: u32, total_msgs: u32, ext: T) -> Self { - TestMessage { - encoded_node_identity: node.identity_key.to_base58_string(), - node_owner: node.owner.clone(), - node_type: NodeType::Gateway, - msg_id, - total_msgs, - ext, + Self::new(node, msg_id, total_msgs, ext) + } + + pub fn new_serialized( + node: N, + msg_id: u32, + total_msgs: u32, + ext: T, + ) -> Result, NetworkTestingError> + where + N: Into, + T: Serialize, + { + Self::new(node, msg_id, total_msgs, ext).as_bytes() + } + + pub fn new_plaintexts( + node: &N, + total_msgs: u32, + ext: T, + ) -> Result>, NetworkTestingError> + where + for<'a> &'a N: Into, + T: Serialize + Clone, + { + let mut msgs = Vec::with_capacity(total_msgs as usize); + for msg_id in 1..=total_msgs { + msgs.push(Self::new(node, msg_id, total_msgs, ext.clone()).as_bytes()?) } + Ok(msgs) + } + + pub fn mix_plaintexts( + node: &mix::Node, + total_msgs: u32, + ext: T, + ) -> Result>, NetworkTestingError> + where + T: Serialize + Clone, + { + Self::new_plaintexts(node, total_msgs, ext) + } + + pub fn gateway_plaintexts( + node: &gateway::Node, + total_msgs: u32, + ext: T, + ) -> Result>, NetworkTestingError> + where + T: Serialize + Clone, + { + Self::new_plaintexts(node, total_msgs, ext) } pub fn as_json_string(&self) -> Result @@ -91,9 +129,9 @@ impl TestMessage { impl Hash for TestMessage { fn hash(&self, state: &mut H) { - self.encoded_node_identity.hash(state); - self.node_owner.hash(state); - self.node_type.hash(state); + self.tested_node.hash(state); + self.msg_id.hash(state); + self.total_msgs.hash(state); self.ext.hash(state) } } diff --git a/common/node-tester-utils/src/node.rs b/common/node-tester-utils/src/node.rs new file mode 100644 index 0000000000..70d328cfba --- /dev/null +++ b/common/node-tester-utils/src/node.rs @@ -0,0 +1,82 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::MixId; +use nym_topology::{gateway, mix}; +use serde::{Deserialize, Serialize}; +use std::fmt::{Display, Formatter}; + +#[derive(Serialize, Deserialize, Debug, Clone, Hash)] +pub struct TestableNode { + pub encoded_identity: String, + pub owner: String, + + #[serde(rename = "type")] + pub typ: NodeType, +} + +impl TestableNode { + pub fn new(encoded_identity: String, owner: String, typ: NodeType) -> Self { + TestableNode { + encoded_identity, + owner, + typ, + } + } + + pub fn new_mixnode(encoded_identity: String, owner: String, mix_id: MixId) -> Self { + TestableNode::new(encoded_identity, owner, NodeType::Mixnode { mix_id }) + } + + pub fn new_gateway(encoded_identity: String, owner: String) -> Self { + TestableNode::new(encoded_identity, owner, NodeType::Gateway) + } +} + +impl<'a> From<&'a mix::Node> for TestableNode { + fn from(value: &'a mix::Node) -> Self { + TestableNode { + encoded_identity: value.identity_key.to_base58_string(), + owner: value.owner.clone(), + typ: NodeType::Mixnode { + mix_id: value.mix_id, + }, + } + } +} + +impl<'a> From<&'a gateway::Node> for TestableNode { + fn from(value: &'a gateway::Node) -> Self { + TestableNode { + encoded_identity: value.identity_key.to_base58_string(), + owner: value.owner.clone(), + typ: NodeType::Gateway, + } + } +} + +impl Display for TestableNode { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{} {} owned by {}", + self.typ, self.encoded_identity, self.owner + ) + } +} + +#[derive(Serialize, Deserialize, Hash, Clone, Copy, Debug)] +#[serde(rename_all = "snake_case")] +pub enum NodeType { + Mixnode { mix_id: MixId }, + Gateway, +} + +impl Display for NodeType { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + match self { + NodeType::Mixnode { mix_id } => write!(f, "mixnode (mix_id {mix_id})"), + NodeType::Gateway => write!(f, "gateway"), + } + } +} diff --git a/common/node-tester-utils/src/processor.rs b/common/node-tester-utils/src/processor.rs new file mode 100644 index 0000000000..809af452d5 --- /dev/null +++ b/common/node-tester-utils/src/processor.rs @@ -0,0 +1,95 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::error::NetworkTestingError; +use crate::TestMessage; +use nym_crypto::asymmetric::encryption; +use nym_sphinx::acknowledgements::identifier::recover_identifier; +use nym_sphinx::acknowledgements::AckKey; +use nym_sphinx::chunking::fragment::FragmentIdentifier; +use nym_sphinx::receiver::{MessageReceiver, SphinxMessageReceiver}; +use serde::de::DeserializeOwned; +use std::marker::PhantomData; +use std::sync::Arc; + +// simple enum containing aggregated processed results +pub enum Received { + Message(TestMessage), + Ack(FragmentIdentifier), +} + +impl From> for Received { + fn from(value: TestMessage) -> Self { + Received::Message(value) + } +} + +impl From for Received { + fn from(value: FragmentIdentifier) -> Self { + Received::Ack(value) + } +} + +pub struct TestPacketProcessor { + local_encryption_keypair: Arc, + ack_key: Arc, + + /// Structure responsible for decrypting and recovering plaintext message from received ciphertexts. + message_receiver: R, + + _ext_phantom: PhantomData, +} + +impl TestPacketProcessor { + pub fn new_sphinx_processor( + local_encryption_keypair: Arc, + ack_key: Arc, + ) -> Self { + TestPacketProcessor { + local_encryption_keypair, + ack_key, + message_receiver: SphinxMessageReceiver::new(), + _ext_phantom: PhantomData, + } + } +} + +impl TestPacketProcessor +where + R: MessageReceiver, +{ + pub fn process_mixnet_message( + &mut self, + mut raw_message: Vec, + ) -> Result, NetworkTestingError> + where + T: DeserializeOwned, + { + let plaintext = self + .message_receiver + .recover_plaintext_from_regular_packet( + self.local_encryption_keypair.private_key(), + &mut raw_message, + )?; + let fragment = self.message_receiver.recover_fragment(plaintext)?; + + // test messages must consist of a single fragment + let (serialized, _) = self + .message_receiver + .insert_new_fragment(fragment)? + .ok_or(NetworkTestingError::NonReconstructablePacket)?; + + TestMessage::try_recover(serialized) + } + + pub fn process_ack( + &mut self, + raw_ack: Vec, + ) -> Result { + let serialized_ack = recover_identifier(&self.ack_key, &raw_ack) + .ok_or(NetworkTestingError::UnrecoverableAck)?; + + FragmentIdentifier::try_from_bytes(serialized_ack) + .map_err(|source| NetworkTestingError::MalformedAckIdentifier { source }) + } +} diff --git a/common/node-tester-utils/src/receiver.rs b/common/node-tester-utils/src/receiver.rs index c2c479d8d5..007b76f988 100644 --- a/common/node-tester-utils/src/receiver.rs +++ b/common/node-tester-utils/src/receiver.rs @@ -2,69 +2,45 @@ // SPDX-License-Identifier: Apache-2.0 use crate::error::NetworkTestingError; +use crate::processor::{Received, TestPacketProcessor}; use crate::{log_err, log_info, log_warn}; use futures::channel::mpsc; use futures::StreamExt; use nym_crypto::asymmetric::encryption; -use nym_sphinx::message::NymMessage; +use nym_sphinx::acknowledgements::AckKey; use nym_sphinx::receiver::{MessageReceiver, SphinxMessageReceiver}; -use nym_sphinx::{ - acknowledgements::{identifier::recover_identifier, AckKey}, - chunking::fragment::FragmentIdentifier, -}; use nym_task::TaskClient; +use serde::de::DeserializeOwned; use std::sync::Arc; -pub type ReceivedSender = mpsc::UnboundedSender; -pub type ReceivedReceiver = mpsc::UnboundedReceiver; - -// simple enum containing aggregated processed results -pub enum Received { - Message(NymMessage), - Ack(FragmentIdentifier), -} - -impl From for Received { - fn from(value: NymMessage) -> Self { - Received::Message(value) - } -} - -impl From for Received { - fn from(value: FragmentIdentifier) -> Self { - Received::Ack(value) - } -} +pub type ReceivedSender = mpsc::UnboundedSender>; +pub type ReceivedReceiver = mpsc::UnboundedReceiver>; // the 'Simple' bit comes from the fact that it expects all received messages to consist of a single `Fragment` -pub struct SimpleMessageReceiver { - local_encryption_keypair: Arc, - - ack_key: Arc, - - /// Structure responsible for decrypting and recovering plaintext message from received ciphertexts. - message_receiver: R, +pub struct SimpleMessageReceiver { + message_processor: TestPacketProcessor, mixnet_message_receiver: mpsc::UnboundedReceiver>>, acks_receiver: mpsc::UnboundedReceiver>>, - received_sender: ReceivedSender, + received_sender: ReceivedSender, shutdown: TaskClient, } -impl SimpleMessageReceiver { +impl SimpleMessageReceiver { pub fn new_sphinx_receiver( local_encryption_keypair: Arc, ack_key: Arc, mixnet_message_receiver: mpsc::UnboundedReceiver>>, acks_receiver: mpsc::UnboundedReceiver>>, - received_sender: ReceivedSender, + received_sender: ReceivedSender, shutdown: TaskClient, ) -> Self { SimpleMessageReceiver { - local_encryption_keypair, - ack_key, - message_receiver: SphinxMessageReceiver::new(), + message_processor: TestPacketProcessor::new_sphinx_processor( + local_encryption_keypair, + ack_key, + ), mixnet_message_receiver, acks_receiver, received_sender, @@ -73,43 +49,33 @@ impl SimpleMessageReceiver { } } -impl SimpleMessageReceiver { - fn forward_received>(&self, received: T) { +impl SimpleMessageReceiver { + fn forward_received>>(&self, received: U) { // TODO: remove the unwrap once/if we do graceful shutdowns here self.received_sender .unbounded_send(received.into()) .expect("ReceivedReceiver has stopped receiving"); } - fn on_mixnet_message(&mut self, mut raw_message: Vec) -> Result<(), NetworkTestingError> { - let plaintext = self - .message_receiver - .recover_plaintext_from_regular_packet( - self.local_encryption_keypair.private_key(), - &mut raw_message, - )?; - let fragment = self.message_receiver.recover_fragment(plaintext)?; - let (recovered, _) = self - .message_receiver - .insert_new_fragment(fragment)? - .ok_or(NetworkTestingError::NonReconstructablePacket)?; // by definition of this receiver, the message must consist of a single fragment - + fn on_mixnet_message(&mut self, raw_message: Vec) -> Result<(), NetworkTestingError> + where + T: DeserializeOwned, + { + let recovered = self.message_processor.process_mixnet_message(raw_message)?; self.forward_received(recovered); Ok(()) } fn on_ack(&mut self, raw_ack: Vec) -> Result<(), NetworkTestingError> { - let serialized_ack = recover_identifier(&self.ack_key, &raw_ack) - .ok_or(NetworkTestingError::UnrecoverableAck)?; - - let frag_id = FragmentIdentifier::try_from_bytes(serialized_ack) - .map_err(|source| NetworkTestingError::MalformedAckIdentifier { source })?; - + let frag_id = self.message_processor.process_ack(raw_ack)?; self.forward_received(frag_id); Ok(()) } - pub async fn run(&mut self) { + pub async fn run(&mut self) + where + T: DeserializeOwned, + { while !self.shutdown.is_shutdown() { tokio::select! { biased; diff --git a/common/node-tester-utils/src/tester.rs b/common/node-tester-utils/src/tester.rs index bb6a376c83..052f919ffe 100644 --- a/common/node-tester-utils/src/tester.rs +++ b/common/node-tester-utils/src/tester.rs @@ -21,7 +21,10 @@ pub struct NodeTester { base_topology: NymTopology, - recipient: Recipient, + /// Generally test packets are designed to be sent from ourselves to ourselves, + /// However, one might want to customise this behaviour. + /// In that case an explicit `Recipient` has to be provided when constructing test packets. + self_address: Option, packet_size: PacketSize, @@ -47,7 +50,7 @@ where pub fn new( rng: R, base_topology: NymTopology, - recipient: Recipient, + self_address: Option, packet_size: PacketSize, average_packet_delay: Duration, average_ack_delay: Duration, @@ -56,7 +59,7 @@ where Self { rng, base_topology, - recipient, + self_address, packet_size, average_packet_delay, average_ack_delay, @@ -89,7 +92,7 @@ where mix: &mix::Node, test_packets: u32, ) -> Result, NetworkTestingError> { - self.mixnode_test_packets(mix, Empty, test_packets) + self.mixnode_test_packets(mix, Empty, test_packets, None) } pub fn mixnode_test_packets( @@ -97,6 +100,7 @@ where mix: &mix::Node, msg_ext: T, test_packets: u32, + custom_recipient: Option, ) -> Result, NetworkTestingError> where T: Serialize + Clone, @@ -104,9 +108,35 @@ where let ephemeral_topology = self.testable_mix_topology(mix); let mut packets = Vec::with_capacity(test_packets as usize); - for i in 1..=test_packets { - let msg = TestMessage::new_mix(mix, i, test_packets, msg_ext.clone()); - packets.push(self.create_test_packet(&msg, &ephemeral_topology)?); + for plaintext in TestMessage::mix_plaintexts(mix, test_packets, msg_ext)? { + packets.push(self.wrap_plaintext_data( + plaintext, + &ephemeral_topology, + custom_recipient, + )?); + } + + Ok(packets) + } + + pub fn mixnodes_test_packets( + &mut self, + nodes: &[mix::Node], + msg_ext: T, + test_packets: u32, + custom_recipient: Option, + ) -> Result, NetworkTestingError> + where + T: Serialize + Clone, + { + let mut packets = Vec::new(); + for node in nodes { + packets.append(&mut self.mixnode_test_packets( + node, + msg_ext.clone(), + test_packets, + custom_recipient, + )?) } Ok(packets) @@ -117,6 +147,7 @@ where mix_id: MixId, msg_ext: T, test_packets: u32, + custom_recipient: Option, ) -> Result, NetworkTestingError> where T: Serialize + Clone, @@ -125,7 +156,7 @@ where return Err(NetworkTestingError::NonExistentMixnode {mix_id}) }; - self.mixnode_test_packets(&node.clone(), msg_ext, test_packets) + self.mixnode_test_packets(&node.clone(), msg_ext, test_packets, custom_recipient) } pub fn existing_identity_mixnode_test_packets( @@ -133,6 +164,7 @@ where encoded_mix_identity: String, msg_ext: T, test_packets: u32, + custom_recipient: Option, ) -> Result, NetworkTestingError> where T: Serialize + Clone, @@ -141,19 +173,57 @@ where return Err(NetworkTestingError::NonExistentMixnodeIdentity { mix_identity: encoded_mix_identity }) }; - self.mixnode_test_packets(&node.clone(), msg_ext, test_packets) + self.mixnode_test_packets(&node.clone(), msg_ext, test_packets, custom_recipient) } - pub fn create_test_packet( + pub fn gateway_test_packets( &mut self, - message: &TestMessage, - topology: &NymTopology, - ) -> Result + gateway: &gateway::Node, + msg_ext: T, + test_packets: u32, + custom_recipient: Option, + ) -> Result, NetworkTestingError> where - T: Serialize, + T: Serialize + Clone, { - let serialized = message.as_bytes()?; - let message = NymMessage::new_plain(serialized); + let ephemeral_topology = self.testable_gateway_topology(gateway); + + let mut packets = Vec::with_capacity(test_packets as usize); + for plaintext in TestMessage::gateway_plaintexts(gateway, test_packets, msg_ext)? { + packets.push(self.wrap_plaintext_data( + plaintext, + &ephemeral_topology, + custom_recipient, + )?); + } + + Ok(packets) + } + + pub fn existing_gateway_test_packets( + &mut self, + encoded_gateway_identity: String, + msg_ext: T, + test_packets: u32, + custom_recipient: Option, + ) -> Result, NetworkTestingError> + where + T: Serialize + Clone, + { + let Some(node) = self.base_topology.find_gateway(&encoded_gateway_identity) else { + return Err(NetworkTestingError::NonExistentGateway { gateway_identity: encoded_gateway_identity }) + }; + + self.gateway_test_packets(&node.clone(), msg_ext, test_packets, custom_recipient) + } + + pub fn wrap_plaintext_data( + &mut self, + plaintext: Vec, + topology: &NymTopology, + custom_recipient: Option, + ) -> Result { + let message = NymMessage::new_plain(plaintext); let mut fragments = self.pad_and_split_message(message, self.packet_size); @@ -165,13 +235,29 @@ where // we would have returned the error when checking for its length let fragment = fragments.pop().unwrap(); - // the packet is designed to be sent from ourselves to ourselves - let address = self.recipient; + // either `self_address` or `custom_recipient` has to be specified. + let address = custom_recipient.unwrap_or( + self.self_address + .ok_or(NetworkTestingError::UnknownPacketRecipient)?, + ); // TODO: can we avoid this arc clone? let ack_key = Arc::clone(&self.ack_key); Ok(self.prepare_chunk_for_sending(fragment, topology, &ack_key, &address, &address)?) } + + pub fn create_test_packet( + &mut self, + message: &TestMessage, + topology: &NymTopology, + custom_recipient: Option, + ) -> Result + where + T: Serialize, + { + let serialized = message.as_bytes()?; + self.wrap_plaintext_data(serialized, topology, custom_recipient) + } } impl FragmentPreparer for NodeTester { diff --git a/common/nymsphinx/Cargo.toml b/common/nymsphinx/Cargo.toml index 766a5e154a..5b6c1cee16 100644 --- a/common/nymsphinx/Cargo.toml +++ b/common/nymsphinx/Cargo.toml @@ -20,6 +20,7 @@ nym-sphinx-chunking = { path = "chunking" } nym-sphinx-cover = { path = "cover" } nym-sphinx-forwarding = { path = "forwarding" } nym-sphinx-params = { path = "params" } +nym-sphinx-routing = { path = "routing" } nym-sphinx-types = { path = "types" } # those dependencies are due to intriducing preparer and receiver. Perpaphs that indicates they should be moved diff --git a/common/nymsphinx/routing/Cargo.toml b/common/nymsphinx/routing/Cargo.toml new file mode 100644 index 0000000000..c0f7cf7628 --- /dev/null +++ b/common/nymsphinx/routing/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "nym-sphinx-routing" +version = "0.1.0" +description = "Sphinx packet routing as Nym mix packets" +edition = { workspace = true } +authors = { workspace = true } +license = { workspace = true } +repository = { workspace = true } + +[dependencies] +thiserror = { workspace = true } + +nym-sphinx-addressing = { path = "../addressing" } +nym-sphinx-types = { path = "../types" } \ No newline at end of file diff --git a/common/nymsphinx/routing/src/lib.rs b/common/nymsphinx/routing/src/lib.rs new file mode 100644 index 0000000000..a7067f252c --- /dev/null +++ b/common/nymsphinx/routing/src/lib.rs @@ -0,0 +1,43 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_sphinx_addressing::clients::Recipient; +use nym_sphinx_types::Node; +use thiserror::Error; + +pub trait SphinxRouteMaker { + type Error; + + fn sphinx_route(&mut self, hops: u8, destination: &Recipient) + -> Result, Self::Error>; +} + +#[derive(Debug, Error, Clone, Copy)] +#[error("the route vector contains {available} nodes while {requested} hops are required")] +pub struct InvalidNumberOfHops { + available: usize, + requested: u8, +} + +// if one wants to provide a hardcoded route, they can +impl SphinxRouteMaker for Vec { + type Error = InvalidNumberOfHops; + + fn sphinx_route( + &mut self, + hops: u8, + _destination: &Recipient, + ) -> Result, InvalidNumberOfHops> { + // it's the responsibility of the caller to ensure the hardcoded route has correct number of hops + // and that it's final hop include the recipient's gateway. + + if self.len() != hops as usize { + Err(InvalidNumberOfHops { + available: self.len(), + requested: hops, + }) + } else { + Ok(self.clone()) + } + } +} diff --git a/common/nymsphinx/src/lib.rs b/common/nymsphinx/src/lib.rs index a9e2d376e4..b3c6bac16a 100644 --- a/common/nymsphinx/src/lib.rs +++ b/common/nymsphinx/src/lib.rs @@ -13,10 +13,12 @@ pub use nym_sphinx_anonymous_replies as anonymous_replies; pub use nym_sphinx_chunking as chunking; pub use nym_sphinx_cover as cover; pub use nym_sphinx_forwarding as forwarding; +pub use nym_sphinx_params as params; +pub use nym_sphinx_routing as routing; +pub use nym_sphinx_types::*; + #[cfg(not(target_arch = "wasm32"))] pub use nym_sphinx_framing as framing; -pub use nym_sphinx_params as params; -pub use nym_sphinx_types::*; // TEMP UNTIL FURTHER REFACTORING pub use preparer::payload::NymsphinxPayloadBuilder; diff --git a/common/nymsphinx/src/preparer/mod.rs b/common/nymsphinx/src/preparer/mod.rs index 514ec2d901..e06a9f44b9 100644 --- a/common/nymsphinx/src/preparer/mod.rs +++ b/common/nymsphinx/src/preparer/mod.rs @@ -38,6 +38,12 @@ pub struct PreparedFragment { pub fragment_identifier: FragmentIdentifier, } +impl From for MixPacket { + fn from(value: PreparedFragment) -> Self { + value.mix_packet + } +} + // this is extracted into a trait with default implementation to remove duplicate code // (which we REALLY want to avoid with crypto) pub trait FragmentPreparer { diff --git a/common/topology/Cargo.toml b/common/topology/Cargo.toml index fbb5985ad5..1d41a93ee4 100644 --- a/common/topology/Cargo.toml +++ b/common/topology/Cargo.toml @@ -23,6 +23,7 @@ nym-crypto = { path = "../crypto" } nym-mixnet-contract-common = { path = "../cosmwasm-smart-contracts/mixnet-contract" } nym-sphinx-addressing = { path = "../nymsphinx/addressing" } nym-sphinx-types = { path = "../nymsphinx/types" } +nym-sphinx-routing = { path = "../nymsphinx/routing" } nym-bin-common = { path = "../bin-common" } [features] diff --git a/common/topology/src/error.rs b/common/topology/src/error.rs new file mode 100644 index 0000000000..170bb1889c --- /dev/null +++ b/common/topology/src/error.rs @@ -0,0 +1,36 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::MixLayer; +use thiserror::Error; + +#[derive(Debug, Clone, Error)] +pub enum NymTopologyError { + #[error("The provided network topology is empty - there are no mixnodes and no gateways on it - the network request(s) probably failed")] + EmptyNetworkTopology, + + #[error("The provided network topology has no gateways available")] + NoGatewaysAvailable, + + #[error("The provided network topology has no mixnodes available")] + NoMixnodesAvailable, + + #[error("Gateway with identity key {identity_key} doesn't exist")] + NonExistentGatewayError { identity_key: String }, + + #[error("Wanted to create a mix route with {requested} hops, while only {available} layers are available")] + InvalidNumberOfHopsError { available: usize, requested: usize }, + + #[error("No mixnodes available on layer {layer}")] + EmptyMixLayer { layer: MixLayer }, + + #[error("Uneven layer distribution. Layer {layer} has {nodes} on it, while we expected a value between {lower_bound} and {upper_bound} as we have {total_nodes} nodes in total. Full breakdown: {layer_distribution:?}")] + UnevenLayerDistribution { + layer: MixLayer, + nodes: usize, + lower_bound: usize, + upper_bound: usize, + total_nodes: usize, + layer_distribution: Vec<(MixLayer, usize)>, + }, +} diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 0ca7922909..ec3b985cc1 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -14,45 +14,17 @@ use std::fmt::{self, Display, Formatter}; use std::io; use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; use std::str::FromStr; -use thiserror::Error; +pub mod error; pub mod filter; pub mod gateway; pub mod mix; +pub mod random_route_provider; #[cfg(feature = "provider-trait")] pub mod provider_trait; -#[derive(Debug, Clone, Error)] -pub enum NymTopologyError { - #[error("The provided network topology is empty - there are no mixnodes and no gateways on it - the network request(s) probably failed")] - EmptyNetworkTopology, - - #[error("The provided network topology has no gateways available")] - NoGatewaysAvailable, - - #[error("The provided network topology has no mixnodes available")] - NoMixnodesAvailable, - - #[error("Gateway with identity key {identity_key} doesn't exist")] - NonExistentGatewayError { identity_key: String }, - - #[error("Wanted to create a mix route with {requested} hops, while only {available} layers are available")] - InvalidNumberOfHopsError { available: usize, requested: usize }, - - #[error("No mixnodes available on layer {layer}")] - EmptyMixLayer { layer: MixLayer }, - - #[error("Uneven layer distribution. Layer {layer} has {nodes} on it, while we expected a value between {lower_bound} and {upper_bound} as we have {total_nodes} nodes in total. Full breakdown: {layer_distribution:?}")] - UnevenLayerDistribution { - layer: MixLayer, - nodes: usize, - lower_bound: usize, - upper_bound: usize, - total_nodes: usize, - layer_distribution: Vec<(MixLayer, usize)>, - }, -} +pub use error::NymTopologyError; #[derive(Debug, Clone)] pub enum NetworkAddress { @@ -134,6 +106,12 @@ impl NymTopology { None } + pub fn find_gateway(&self, gateway_identity: IdentityKeyRef) -> Option<&gateway::Node> { + self.gateways + .iter() + .find(|&gateway| gateway.identity_key.to_base58_string() == gateway_identity) + } + pub fn mixes(&self) -> &BTreeMap> { &self.mixes } @@ -182,8 +160,7 @@ impl NymTopology { num_mix_hops: u8, ) -> Result, NymTopologyError> where - // I don't think there's a need for this RNG to be crypto-secure - R: Rng + ?Sized, + R: Rng + CryptoRng + ?Sized, { use rand::seq::SliceRandom; diff --git a/common/topology/src/random_route_provider.rs b/common/topology/src/random_route_provider.rs new file mode 100644 index 0000000000..65d9452161 --- /dev/null +++ b/common/topology/src/random_route_provider.rs @@ -0,0 +1,29 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::{NymTopology, NymTopologyError}; +use nym_sphinx_addressing::clients::Recipient; +use nym_sphinx_routing::SphinxRouteMaker; +use nym_sphinx_types::Node; +use rand::{CryptoRng, Rng}; + +pub struct NymTopologyRouteProvider { + rng: R, + inner: NymTopology, +} + +impl SphinxRouteMaker for NymTopologyRouteProvider +where + R: Rng + CryptoRng, +{ + type Error = NymTopologyError; + + fn sphinx_route( + &mut self, + hops: u8, + destination: &Recipient, + ) -> Result, NymTopologyError> { + self.inner + .random_route_to_gateway(&mut self.rng, hops, destination.gateway()) + } +} diff --git a/nym-api/Cargo.toml b/nym-api/Cargo.toml index daa7da44b3..85cbd470be 100644 --- a/nym-api/Cargo.toml +++ b/nym-api/Cargo.toml @@ -92,6 +92,7 @@ nym-validator-client = { path = "../common/client-libs/validator-client", featur "nyxd-client", ] } nym-bin-common = { path = "../common/bin-common" } +nym-node-tester-utils = { path = "../common/node-tester-utils" } [features] no-reward = [] diff --git a/nym-api/src/network_monitor/chunker.rs b/nym-api/src/network_monitor/chunker.rs index efce5e92d7..ef4f88e472 100644 --- a/nym-api/src/network_monitor/chunker.rs +++ b/nym-api/src/network_monitor/chunker.rs @@ -15,6 +15,7 @@ const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200); const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200); #[derive(Clone)] +#[deprecated] pub(crate) struct Chunker { rng: OsRng, packet_size: PacketSize, diff --git a/nym-api/src/network_monitor/gateways_reader.rs b/nym-api/src/network_monitor/gateways_reader.rs index ab521dd14e..38956bade1 100644 --- a/nym-api/src/network_monitor/gateways_reader.rs +++ b/nym-api/src/network_monitor/gateways_reader.rs @@ -1,18 +1,22 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use futures::Stream; use nym_crypto::asymmetric::identity; use nym_gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver}; +use nym_mixnet_contract_common::IdentityKey; use std::pin::Pin; use std::task::{Context, Poll}; use tokio_stream::StreamMap; -pub(crate) type GatewayMessages = Vec>; +pub(crate) enum GatewayMessages { + Data(Vec>), + Acks(Vec>), +} pub(crate) struct GatewaysReader { - ack_map: StreamMap, - stream_map: StreamMap, + ack_map: StreamMap, + stream_map: StreamMap, } impl GatewaysReader { @@ -41,8 +45,7 @@ impl GatewaysReader { } impl Stream for GatewaysReader { - // just return whatever is returned by our main `stream_map` - type Item = as Stream>::Item; + type Item = (IdentityKey, GatewayMessages); fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { // exhaust the ack map if possible @@ -51,10 +54,18 @@ impl Stream for GatewaysReader { // this should have never happened! return Poll::Ready(None); } - Poll::Ready(Some(_item)) => (), + Poll::Ready(Some(ack_item)) => { + // wake immediately in case there's an associated data message + cx.waker().wake(); + return Poll::Ready(Some((ack_item.0, GatewayMessages::Acks(ack_item.1)))); + } Poll::Pending => (), } - Pin::new(&mut self.stream_map).poll_next(cx) + Pin::new(&mut self.stream_map) + .poll_next(cx) + .map(|maybe_data_item| { + maybe_data_item.map(|data_item| (data_item.0, GatewayMessages::Data(data_item.1))) + }) } } diff --git a/nym-api/src/network_monitor/monitor/preparer.rs b/nym-api/src/network_monitor/monitor/preparer.rs index a0df027e9a..7183c9f4c4 100644 --- a/nym-api/src/network_monitor/monitor/preparer.rs +++ b/nym-api/src/network_monitor/monitor/preparer.rs @@ -3,112 +3,70 @@ use crate::network_monitor::chunker::Chunker; use crate::network_monitor::monitor::sender::GatewayPackets; -use crate::network_monitor::test_packet::{NodeType, TestPacket}; +use crate::network_monitor::test_packet::{NodeTestMessage, NymApiTestMessageExt}; use crate::network_monitor::test_route::TestRoute; use crate::nym_contract_cache::cache::NymContractCache; use log::info; use nym_crypto::asymmetric::{encryption, identity}; -use nym_mixnet_contract_common::{Addr, GatewayBond, Layer, MixId, MixNodeBond}; +use nym_mixnet_contract_common::{Addr, GatewayBond, IdentityKey, Layer, MixId, MixNodeBond}; +use nym_node_tester_utils::node::{NodeType, TestableNode}; +use nym_node_tester_utils::NodeTester; +use nym_sphinx::acknowledgements::AckKey; use nym_sphinx::addressing::clients::Recipient; use nym_sphinx::forwarding::packet::MixPacket; +use nym_sphinx::params::PacketSize; use nym_topology::{gateway, mix, NymTopology}; -use rand::seq::SliceRandom; -use rand::{thread_rng, Rng}; +use rand_07::{rngs::ThreadRng, seq::SliceRandom, thread_rng, Rng}; use std::collections::{HashMap, HashSet}; use std::convert::TryInto; use std::fmt::{self, Display, Formatter}; +use std::sync::Arc; use std::time::Duration; -// declared type aliases for easier code reasoning -type Version = String; -type Id = String; -type Owner = Addr; +const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200); +const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200); #[derive(Clone)] -#[allow(dead_code)] pub(crate) enum InvalidNode { - Outdated(Id, Owner, NodeType, Version), - Malformed(Id, Owner, NodeType), + Outdated { node: TestableNode, version: String }, + Malformed { node: TestableNode }, } impl Display for InvalidNode { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { match self { - InvalidNode::Outdated(id, owner, _, version) => { - write!( - f, - "Node {} (v{}) owned by {} is outdated", - id, version, owner - ) + InvalidNode::Outdated { node, version } => { + write!(f, "{node} is outdated. It runs on v{version}") } - InvalidNode::Malformed(id, owner, _) => { - write!(f, "Node {} owned by {} is malformed", id, owner) + InvalidNode::Malformed { node } => { + write!(f, "{node} is malformed") } } } } -impl InvalidNode { - pub(crate) fn mix_id(&self) -> Option { - match self { - InvalidNode::Outdated(_, _, node_type, _) => node_type.mix_id(), - InvalidNode::Malformed(_, _, node_type) => node_type.mix_id(), - } - } - - pub(crate) fn identity(&self) -> String { - match self { - InvalidNode::Outdated(id, _, _, _) => id.clone(), - InvalidNode::Malformed(id, _, _) => id.clone(), - } - } - - pub(crate) fn owner(&self) -> String { - match self { - InvalidNode::Outdated(_, owner, _, _) => owner.into(), - InvalidNode::Malformed(_, owner, _) => owner.into(), - } - } -} - -#[derive(Eq, PartialEq, Debug, Hash, Clone)] -pub(crate) struct TestedNode { - pub(crate) identity: String, - pub(crate) owner: String, - pub(crate) node_type: NodeType, -} - -impl TestedNode { - pub(crate) fn mix_id(&self) -> Option { - self.node_type.mix_id() - } -} - -impl<'a> From<&'a mix::Node> for TestedNode { - fn from(node: &'a mix::Node) -> Self { - TestedNode { - identity: node.identity_key.to_base58_string(), - owner: node.owner.clone(), - node_type: NodeType::Mixnode(node.mix_id), - } - } -} - -impl<'a> From<&'a gateway::Node> for TestedNode { - fn from(node: &'a gateway::Node) -> Self { - TestedNode { - identity: node.identity_key.to_base58_string(), - owner: node.owner.clone(), - node_type: NodeType::Gateway, - } - } -} - -impl Display for TestedNode { - fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - write!(f, "{} (owned by {})", self.identity, self.owner) - } -} +// impl InvalidNode { +// pub(crate) fn mix_id(&self) -> Option { +// match self { +// InvalidNode::Outdated(_, _, node_type, _) => node_type.mix_id(), +// InvalidNode::Malformed(_, _, node_type) => node_type.mix_id(), +// } +// } +// +// pub(crate) fn identity(&self) -> String { +// match self { +// InvalidNode::Outdated(id, _, _, _) => id.clone(), +// InvalidNode::Malformed(id, _, _) => id.clone(), +// } +// } +// +// pub(crate) fn owner(&self) -> String { +// match self { +// InvalidNode::Outdated(_, owner, _, _) => owner.into(), +// InvalidNode::Malformed(_, owner, _) => owner.into(), +// } +// } +// } pub(crate) struct PreparedPackets { /// All packets that are going to get sent during the test as well as the gateways through @@ -116,10 +74,10 @@ pub(crate) struct PreparedPackets { pub(super) packets: Vec, /// Vector containing list of public keys and owners of all nodes mixnodes being tested. - pub(super) tested_mixnodes: Vec, + pub(super) tested_mixnodes: Vec, /// Vector containing list of public keys and owners of all gateways being tested. - pub(super) tested_gateways: Vec, + pub(super) tested_gateways: Vec, /// All mixnodes that failed to get parsed correctly or were not version compatible. /// They will be marked to the validator as being down for the test. @@ -133,7 +91,6 @@ pub(crate) struct PreparedPackets { #[derive(Clone)] pub(crate) struct PacketPreparer { system_version: String, - chunker: Option, validator_cache: NymContractCache, /// Number of test packets sent to each node @@ -155,39 +112,55 @@ impl PacketPreparer { self_public_identity: identity::PublicKey, self_public_encryption: encryption::PublicKey, ) -> Self { - PacketPreparer { - system_version: system_version.to_owned(), - chunker: None, - validator_cache, - per_node_test_packets, - self_public_identity, - self_public_encryption, - } + todo!() + // PacketPreparer { + // system_version: system_version.to_owned(), + // chunker: None, + // validator_cache, + // per_node_test_packets, + // self_public_identity, + // self_public_encryption, + // } } - fn wrap_test_packet( - &mut self, - packet: &TestPacket, - topology: &NymTopology, - packet_recipient: Recipient, - ) -> MixPacket { - // this should be done only once. We can't really do it at construction time - // as there's no sane Default for Recipient - if self.chunker.is_none() { - self.chunker = Some(Chunker::new(packet_recipient)); - } - let mut mix_packets = self.chunker.as_mut().unwrap().prepare_packets_from( - packet.to_bytes(), - topology, - packet_recipient, - ); - assert_eq!( - mix_packets.len(), - 1, - "Our test packets data is longer than a single sphinx packet!" - ); + fn ephemeral_tester( + &self, + test_route: &TestRoute, + self_address: Option, + ) -> NodeTester { + let mut rng = thread_rng(); - mix_packets.pop().unwrap() + // TODO: we should make this key more long-term + let ack_key = Arc::new(AckKey::new(&mut rng)); + + NodeTester::new( + rng, + // the topology here contains 3 mixnodes and 1 gateway so its cheap to clone it + test_route.topology().clone(), + self_address, + PacketSize::RegularPacket, + DEFAULT_AVERAGE_PACKET_DELAY, + DEFAULT_AVERAGE_ACK_DELAY, + ack_key, + ) + } + + // when we're testing mixnodes, the recipient is going to stay constant, so we can specify it ahead of time + fn ephemeral_mix_tester(&self, test_route: &TestRoute) -> NodeTester { + let self_address = self.create_packet_sender(test_route.gateway()); + self.ephemeral_tester(test_route, Some(self_address)) + } + + fn ephemeral_gateway_tester(&self, test_route: &TestRoute) -> NodeTester { + self.ephemeral_tester(test_route, None) + } + + async fn topology_wait_backoff(&self, initialisation_backoff: Duration) { + info!( + "Minimal topology is still not online. Going to check again in {:?}", + initialisation_backoff + ); + tokio::time::sleep(initialisation_backoff).await; } pub(crate) async fn wait_for_validator_cache_initial_values(&self, minimum_full_routes: usize) { @@ -202,39 +175,30 @@ impl PacketPreparer { let mixnodes = self.validator_cache.mixnodes_basic().await; if gateways.len() < minimum_full_routes { - info!( - "Minimal topology is still not online. Going to check again in {:?}", - initialisation_backoff - ); - tokio::time::sleep(initialisation_backoff).await; + self.topology_wait_backoff(initialisation_backoff).await; continue; } - let mut layered_mixes = HashMap::new(); + let mut layer1_count = 0; + let mut layer2_count = 0; + let mut layer3_count = 0; + for mix in mixnodes { - let layer = mix.layer; - let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new); - mixes.push(mix) + match mix.layer { + Layer::One => layer1_count += 1, + Layer::Two => layer2_count += 1, + Layer::Three => layer3_count += 1, + } } - // we remove the entries as this gives us the ownership and thus we can unwrap to default value - // which makes the code slightly nicer without having to deal with options - let layer1 = layered_mixes.remove(&Layer::One).unwrap_or_default(); - let layer2 = layered_mixes.remove(&Layer::Two).unwrap_or_default(); - let layer3 = layered_mixes.remove(&Layer::Three).unwrap_or_default(); - - if layer1.len() >= minimum_full_routes - && layer2.len() >= minimum_full_routes - && layer3.len() >= minimum_full_routes + if layer1_count >= minimum_full_routes + && layer2_count >= minimum_full_routes + && layer3_count >= minimum_full_routes { break; } - info!( - "Minimal topology is still not online. Going to check again in {:?}", - initialisation_backoff - ); - tokio::time::sleep(initialisation_backoff).await; + self.topology_wait_backoff(initialisation_backoff).await; } } @@ -305,55 +269,43 @@ impl PacketPreparer { if most_available == 0 { error!("Cannot construct test routes. No nodes or gateways available"); - None - } else { - trace!("Generating test routes..."); - let mut routes = Vec::new(); - for i in 0..most_available { - let node_1 = match self.try_parse_mix_bond(rand_l1[i]) { - Ok(node) => node, - Err(id) => { - blacklist.insert(id); - continue; - } - }; - - let node_2 = match self.try_parse_mix_bond(rand_l2[i]) { - Ok(node) => node, - Err(id) => { - blacklist.insert(id); - continue; - } - }; - - let node_3 = match self.try_parse_mix_bond(rand_l3[i]) { - Ok(node) => node, - Err(id) => { - blacklist.insert(id); - continue; - } - }; - - let gateway = match self.try_parse_gateway_bond(rand_gateways[i]) { - Ok(node) => node, - Err(id) => { - blacklist.insert(id); - continue; - } - }; - - routes.push(TestRoute::new( - rng.gen(), - &self.system_version, - node_1, - node_2, - node_3, - gateway, - )) - } - info!("{:?}", routes); - Some(routes) + return None; } + + trace!("Generating test routes..."); + let mut routes = Vec::new(); + for i in 0..most_available { + let Ok(node_1) = self.try_parse_mix_bond(rand_l1[i]) else { + blacklist.insert(rand_l1[i].identity().to_owned()); + continue + }; + + let Ok(node_2) = self.try_parse_mix_bond(rand_l2[i]) else { + blacklist.insert(rand_l2[i].identity().to_owned()); + continue + }; + + let Ok(node_3) = self.try_parse_mix_bond(rand_l3[i]) else { + blacklist.insert(rand_l3[i].identity().to_owned()); + continue + }; + + let Ok(gateway) = self.try_parse_gateway_bond(rand_gateways[i]) else { + blacklist.insert(rand_gateways[i].identity().to_owned()); + continue + }; + + routes.push(TestRoute::new( + rng.gen(), + &self.system_version, + node_1, + node_2, + node_3, + gateway, + )) + } + info!("{:?}", routes); + Some(routes) } fn create_packet_sender(&self, gateway: &gateway::Node) -> Recipient { @@ -369,13 +321,19 @@ impl PacketPreparer { route: &TestRoute, num: usize, ) -> GatewayPackets { - let mut mix_packets = Vec::with_capacity(num); - let test_packet = route.self_test_packet(); - let recipient = self.create_packet_sender(route.gateway()); - for _ in 0..num { - let mix_packet = self.wrap_test_packet(&test_packet, route.topology(), recipient); - mix_packets.push(mix_packet) - } + let mut tester = self.ephemeral_mix_tester(route); + let topology = route.topology(); + let plaintexts = route.self_test_messages(num); + + // the unwrap here is fine as: + // 1. the topology is definitely valid (otherwise we wouldn't be here) + // 2. the recipient is specified (by calling **mix**_tester) + // 3. the test message is not too long, i.e. when serialized it will fit in a single sphinx packet + let mix_packets = plaintexts + .into_iter() + .map(|p| tester.wrap_plaintext_data(p, topology, None).unwrap()) + .map(MixPacket::from) + .collect(); GatewayPackets::new( route.gateway_clients_address(), @@ -384,6 +342,26 @@ impl PacketPreparer { ) } + // pub(crate) fn prepare_test_route_viability_packets( + // &mut self, + // route: &TestRoute, + // num: usize, + // ) -> GatewayPackets { + // let mut mix_packets = Vec::with_capacity(num); + // let test_packet = route.self_test_packet(); + // let recipient = self.create_packet_sender(route.gateway()); + // for _ in 0..num { + // let mix_packet = self.wrap_test_packet(&test_packet, route.topology(), recipient); + // mix_packets.push(mix_packet) + // } + // + // GatewayPackets::new( + // route.gateway_clients_address(), + // route.gateway_identity(), + // mix_packets, + // ) + // } + fn filter_outdated_and_malformed_mixnodes( &self, nodes: Vec, @@ -394,11 +372,13 @@ impl PacketPreparer { if let Ok(parsed_node) = (&mixnode).try_into() { parsed_nodes.push(parsed_node) } else { - invalid_nodes.push(InvalidNode::Malformed( - mixnode.mix_node.identity_key, - mixnode.owner, - NodeType::Mixnode(mixnode.mix_id), - )); + invalid_nodes.push(InvalidNode::Malformed { + node: TestableNode::new_mixnode( + mixnode.identity().to_owned(), + mixnode.owner.clone().into_string(), + mixnode.mix_id, + ), + }); } } (parsed_nodes, invalid_nodes) @@ -414,11 +394,12 @@ impl PacketPreparer { if let Ok(parsed_node) = (&gateway).try_into() { parsed_nodes.push(parsed_node) } else { - invalid_nodes.push(InvalidNode::Malformed( - gateway.gateway.identity_key, - gateway.owner, - NodeType::Gateway, - )); + invalid_nodes.push(InvalidNode::Malformed { + node: TestableNode::new_gateway( + gateway.identity().to_owned(), + gateway.owner.clone().into_string(), + ), + }); } } (parsed_nodes, invalid_nodes) @@ -449,42 +430,54 @@ impl PacketPreparer { // for each test route... for test_route in test_routes { - let recipient = self.create_packet_sender(test_route.gateway()); - let gateway_identity = test_route.gateway_identity(); + let route_ext = test_route.test_message_ext(test_nonce); let gateway_address = test_route.gateway_clients_address(); + let gateway_identity = test_route.gateway_identity(); - // it's actually going to be a tiny bit more due to gateway testing, but it's a good enough approximation - let mut mix_packets = Vec::with_capacity(mixnodes.len() * self.per_node_test_packets); + let recipient = self.create_packet_sender(test_route.gateway()); + let mut mix_tester = self.ephemeral_mix_tester(test_route); - // and for each mixnode... - for mixnode in &mixnodes { - let test_packet = TestPacket::from_mixnode(mixnode, test_route.id(), test_nonce); - let topology = test_route.substitute_mix(mixnode); - // produce n mix packets - for _ in 0..self.per_node_test_packets { - let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient); - mix_packets.push(mix_packet); - } - } + // generate test packets for mixnodes + // + // the unwrap here is fine as: + // 1. the topology is definitely valid (otherwise we wouldn't be here) + // 2. the recipient is specified (by calling **mix**_tester) + // 3. the test message is not too long, i.e. when serialized it will fit in a single sphinx packet + let mixnode_test_packets = mix_tester + .mixnodes_test_packets( + &mixnodes, + route_ext, + self.per_node_test_packets as u32, + None, + ) + .unwrap(); + let mix_packets = mixnode_test_packets.into_iter().map(Into::into).collect(); let gateway_packets = all_gateway_packets .entry(gateway_identity.to_bytes()) .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity)); gateway_packets.push_packets(mix_packets); - // and for each gateway... + // and generate test packets for gateways (note the variable recipient) for gateway in &gateways { - let mut gateway_mix_packets = Vec::new(); - let test_packet = TestPacket::from_gateway(gateway, test_route.id(), test_nonce); + let recipient = self.create_packet_sender(gateway); let gateway_identity = gateway.identity_key; let gateway_address = gateway.clients_address(); - let recipient = self.create_packet_sender(gateway); - let topology = test_route.substitute_gateway(gateway); - // produce n mix packets - for _ in 0..self.per_node_test_packets { - let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient); - gateway_mix_packets.push(mix_packet); - } + + // the unwrap here is fine as: + // 1. the topology is definitely valid (otherwise we wouldn't be here) + // 2. the recipient is specified + // 3. the test message is not too long, i.e. when serialized it will fit in a single sphinx packet + let gateway_test_packets = mix_tester + .gateway_test_packets( + gateway, + route_ext, + self.per_node_test_packets as u32, + Some(recipient), + ) + .unwrap(); + let gateway_mix_packets = + gateway_test_packets.into_iter().map(Into::into).collect(); // and push it into existing struct (if it's a "core" gateway being tested against another route) // or create a new one @@ -493,6 +486,54 @@ impl PacketPreparer { .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity)); gateway_packets.push_packets(gateway_mix_packets); } + + // + // + // + + // // let gateway_identity = test_route.gateway_identity(); + // let gateway_address = test_route.gateway_clients_address(); + // + // // it's actually going to be a tiny bit more due to gateway testing, but it's a good enough approximation + // let mut mix_packets = Vec::with_capacity(mixnodes.len() * self.per_node_test_packets); + // + // // and for each mixnode... + // for mixnode in &mixnodes { + // let test_packet = TestPacket::from_mixnode(mixnode, test_route.id(), test_nonce); + // let topology = test_route.substitute_mix(mixnode); + // // produce n mix packets + // for _ in 0..self.per_node_test_packets { + // let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient); + // mix_packets.push(mix_packet); + // } + // } + // + // let gateway_packets = all_gateway_packets + // .entry(gateway_identity.to_bytes()) + // .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity)); + // gateway_packets.push_packets(mix_packets); + // + // // and for each gateway... + // for gateway in &gateways { + // let mut gateway_mix_packets = Vec::new(); + // let test_packet = TestPacket::from_gateway(gateway, test_route.id(), test_nonce); + // let gateway_identity = gateway.identity_key; + // let gateway_address = gateway.clients_address(); + // let recipient = self.create_packet_sender(gateway); + // let topology = test_route.substitute_gateway(gateway); + // // produce n mix packets + // for _ in 0..self.per_node_test_packets { + // let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient); + // gateway_mix_packets.push(mix_packet); + // } + // + // // and push it into existing struct (if it's a "core" gateway being tested against another route) + // // or create a new one + // let gateway_packets = all_gateway_packets + // .entry(gateway_identity.to_bytes()) + // .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity)); + // gateway_packets.push_packets(gateway_mix_packets); + // } } // convert our hashmap back into a vec diff --git a/nym-api/src/network_monitor/monitor/processor.rs b/nym-api/src/network_monitor/monitor/processor.rs index 4ea4c2f880..5900543253 100644 --- a/nym-api/src/network_monitor/monitor/processor.rs +++ b/nym-api/src/network_monitor/monitor/processor.rs @@ -1,14 +1,16 @@ -// Copyright 2021-2022 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use crate::network_monitor::gateways_reader::GatewayMessages; -use crate::network_monitor::test_packet::{TestPacket, TestPacketError}; +use crate::network_monitor::test_packet::NymApiTestMessageExt; use crate::network_monitor::ROUTE_TESTING_TEST_NONCE; use futures::channel::mpsc; use futures::lock::{Mutex, MutexGuard}; use futures::{SinkExt, StreamExt}; use log::warn; use nym_crypto::asymmetric::encryption; +use nym_node_tester_utils::error::NetworkTestingError; +use nym_node_tester_utils::processor::TestPacketProcessor; use nym_sphinx::receiver::{MessageReceiver, MessageRecoveryError}; use std::mem; use std::sync::Arc; @@ -20,15 +22,15 @@ pub(crate) type ReceivedProcessorReceiver = mpsc::UnboundedReceiver { /// Channel for receiving packets/messages from the gateway clients packets_receiver: ReceivedProcessorReceiver, - // TODO: right now it's identical for each gateway we send through, but should it? - /// Encryption key of the clients sending through the gateways. - client_encryption_keypair: Arc, - - /// Structure responsible for decrypting and recovering plaintext message from received ciphertexts. - message_receiver: R, - - /// Vector containing all received (and decrypted) packets in the current test run. - received_packets: Vec, + test_processor: TestPacketProcessor, + // TODO: rethinking + // /// Vector containing all received (and decrypted) packets in the current test run. + // received_packets: Vec, } impl ReceivedProcessorInner { + fn recover_test_message(&mut self, raw_message: Vec) { + // + } + + fn on_received_data(&mut self, raw_message: Vec) -> Result<(), ProcessingError> { + let test_msg = self.test_processor.process_mixnet_message(raw_message)?; + + if test_msg.ext.test_nonce != self.test_nonce.unwrap() { + return Err(ProcessingError::NonMatchingNonce { + received: test_msg.ext.test_nonce, + expected: self.test_nonce.unwrap(), + }); + } + + todo!() + } + + fn on_received_ack(&mut self, raw_ack: Vec) -> Result<(), ProcessingError> { + let frag_id = self.test_processor.process_ack(raw_ack)?; + // TODO: hook it up at some point + trace!("received a test ack with id {frag_id}. However, we're not goingn to do anything about it (just yet)"); + + Ok(()) + } + + fn on_received(&mut self, messages: GatewayMessages) { + match messages { + GatewayMessages::Data(data_msgs) => { + for raw in data_msgs { + if let Err(err) = self.on_received_data(raw) { + warn!(target: "Monitor", "failed to process received gateway message: {err}") + } + } + } + GatewayMessages::Acks(acks) => { + for raw in acks { + if let Err(err) = self.on_received_ack(raw) { + warn!(target: "Monitor", "failed to process received gateway ack: {err}") + } + } + } + } + } + fn on_message(&mut self, mut message: Vec) -> Result<(), ProcessingError> { // if the nonce is none it means the packet was received during the 'waiting' for the // next test run @@ -111,19 +152,21 @@ impl ReceivedProcessor { packets_receiver: ReceivedProcessorReceiver, client_encryption_keypair: Arc, ) -> Self { - let inner: Arc>> = - Arc::new(Mutex::new(ReceivedProcessorInner { - test_nonce: None, - packets_receiver, - client_encryption_keypair, - message_receiver: R::new(), - received_packets: Vec::new(), - })); - - ReceivedProcessor { - permit_changer: None, - inner, - } + todo!() + // + // let inner: Arc>> = + // Arc::new(Mutex::new(ReceivedProcessorInner { + // test_nonce: None, + // packets_receiver, + // client_encryption_keypair, + // message_receiver: R::new(), + // received_packets: Vec::new(), + // })); + // + // ReceivedProcessor { + // permit_changer: None, + // inner, + // } } pub(crate) fn start_receiving(&mut self) { @@ -150,13 +193,7 @@ impl ReceivedProcessor { None => return, }, messages = inner.packets_receiver.next() => match messages { - Some(messages) => { - for message in messages { - if let Err(err) = inner.on_message(message) { - warn!(target: "Monitor", "failed to process received gateway message - {err}") - } - } - } + Some(messages) => inner.on_received(messages), None => return, }, } diff --git a/nym-api/src/network_monitor/monitor/receiver.rs b/nym-api/src/network_monitor/monitor/receiver.rs index ad9e90162f..299044e285 100644 --- a/nym-api/src/network_monitor/monitor/receiver.rs +++ b/nym-api/src/network_monitor/monitor/receiver.rs @@ -1,4 +1,4 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use crate::network_monitor::gateways_reader::{GatewayMessages, GatewaysReader}; @@ -66,12 +66,12 @@ impl PacketReceiver { // unwrap here is fine as it can only return a `None` if the PacketSender has died // and if that was the case, then the entire monitor is already in an undefined state update = self.clients_updater.next() => self.process_gateway_update(update.unwrap()), - gateway_message = self.gateways_reader.next() => { - let Some((_gateway_id, message)) = gateway_message else { + gateway_messages = self.gateways_reader.next() => { + let Some((_gateway_id, messages)) = gateway_messages else { log::error!("the gateways reader stream has terminated!"); continue }; - self.process_gateway_messages(message) + self.process_gateway_messages(messages) } } } diff --git a/nym-api/src/network_monitor/monitor/summary_producer.rs b/nym-api/src/network_monitor/monitor/summary_producer.rs index 704084c338..4096900cd5 100644 --- a/nym-api/src/network_monitor/monitor/summary_producer.rs +++ b/nym-api/src/network_monitor/monitor/summary_producer.rs @@ -1,10 +1,10 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::network_monitor::monitor::preparer::{InvalidNode, TestedNode}; -use crate::network_monitor::test_packet::{NodeType, TestPacket}; +use crate::network_monitor::monitor::preparer::InvalidNode; use crate::network_monitor::test_route::TestRoute; use nym_mixnet_contract_common::MixId; +use nym_node_tester_utils::node::TestableNode; use std::collections::HashMap; use std::fmt::{Display, Formatter}; @@ -281,8 +281,8 @@ impl SummaryProducer { pub(super) fn produce_summary( &self, - tested_mixnodes: Vec, - tested_gateways: Vec, + tested_mixnodes: Vec, + tested_gateways: Vec, received_packets: Vec, invalid_mixnodes: Vec, invalid_gateways: Vec, diff --git a/nym-api/src/network_monitor/test_packet.rs b/nym-api/src/network_monitor/test_packet.rs index 0a38e642b7..2bba7ddeb7 100644 --- a/nym-api/src/network_monitor/test_packet.rs +++ b/nym-api/src/network_monitor/test_packet.rs @@ -1,260 +1,40 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::network_monitor::monitor::preparer::TestedNode; -use nym_crypto::asymmetric::identity; -use nym_crypto::asymmetric::identity::Ed25519RecoveryError; -use nym_mixnet_contract_common::MixId; +use nym_node_tester_utils::error::NetworkTestingError; +use nym_node_tester_utils::TestMessage; use nym_topology::{gateway, mix}; -use std::convert::TryInto; -use std::fmt::{self, Display, Formatter}; -use std::hash::{Hash, Hasher}; -use std::mem; -use std::str::Utf8Error; -use thiserror::Error; +use serde::{Deserialize, Serialize}; -const MIXNODE_TYPE: u8 = 0; -const GATEWAY_TYPE: u8 = 1; +pub(crate) type NodeTestMessage = TestMessage; -#[repr(u8)] -#[derive(Eq, PartialEq, Debug, Hash, Clone, Copy)] -pub(crate) enum NodeType { - Mixnode(MixId), - Gateway, -} - -impl NodeType { - fn size(&self) -> usize { - match self { - NodeType::Mixnode(_) => 1 + mem::size_of::(), - NodeType::Gateway => 1, - } - } - - pub(crate) fn mix_id(&self) -> Option { - match self { - NodeType::Mixnode(mix_id) => Some(*mix_id), - NodeType::Gateway => None, - } - } - - pub(crate) fn into_bytes(self) -> Vec { - match self { - NodeType::Mixnode(mix_id) => { - let mut bytes = Vec::with_capacity(5); - bytes.push(MIXNODE_TYPE); - bytes.extend_from_slice(&mix_id.to_be_bytes()); - bytes - } - NodeType::Gateway => vec![GATEWAY_TYPE], - } - } - - pub(crate) fn try_from_bytes(b: &[u8]) -> Result { - if b.is_empty() { - return Err(TestPacketError::InvalidNodeType); - } - match b[0] { - t if t == MIXNODE_TYPE => { - if b.len() < (1 + mem::size_of::()) { - return Err(TestPacketError::InvalidNodeType); - } - Ok(NodeType::Mixnode(MixId::from_be_bytes( - b[1..1 + mem::size_of::()].try_into().unwrap(), - ))) - } - t if t == GATEWAY_TYPE => Ok(NodeType::Gateway), - _ => Err(TestPacketError::InvalidNodeType), - } - } -} - -#[derive(Debug, Error)] -pub(crate) enum TestPacketError { - #[error( - "the received packet was incomplete. Got {received} bytes but expected at least {min_expected}" - )] - IncompletePacket { - received: usize, - min_expected: usize, - }, - - // TODO: ideally this should contain more information but that'd require more refactoring than I'm willing to commit to now - #[error("the received packet did not contain a valid node type")] - InvalidNodeType, - - #[error("the received node identity key was malformed - {0}")] - InvalidNodeKey(#[from] Ed25519RecoveryError), - - #[error("the received packet contained malformed owner data - {0}")] - InvalidOwner(#[from] Utf8Error), -} - -#[derive(Eq, Clone, Debug)] -pub(crate) struct TestPacket { +#[derive(Serialize, Deserialize, Clone, Copy, Hash)] +pub(crate) struct NymApiTestMessageExt { pub(crate) route_id: u64, pub(crate) test_nonce: u64, - pub(crate) pub_key: identity::PublicKey, - pub(crate) owner: String, - pub(crate) node_type: NodeType, } -impl Display for TestPacket { - fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - write!( - f, - "TestPacket {{ pub_key: {}, owner: {}, route: {} test nonce: {} }}", - self.pub_key.to_base58_string(), - self.owner, - self.route_id, - self.test_nonce - ) - } -} - -impl Hash for TestPacket { - fn hash(&self, state: &mut H) { - self.route_id.hash(state); - self.test_nonce.hash(state); - self.pub_key.to_bytes().hash(state); - self.owner.hash(state); - self.node_type.hash(state); - } -} - -impl PartialEq for TestPacket { - fn eq(&self, other: &Self) -> bool { - self.route_id == other.route_id - && self.test_nonce == other.test_nonce - && self.pub_key.to_bytes() == other.pub_key.to_bytes() - && self.owner == other.owner - && self.node_type == other.node_type - } -} - -impl TestPacket { - pub(crate) fn from_mixnode(mix: &mix::Node, route_id: u64, test_nonce: u64) -> Self { - TestPacket { - pub_key: mix.identity_key, - owner: mix.owner.clone(), +impl NymApiTestMessageExt { + pub fn new(route_id: u64, test_nonce: u64) -> Self { + NymApiTestMessageExt { route_id, test_nonce, - node_type: NodeType::Mixnode(mix.mix_id), } } - pub(crate) fn from_gateway(gateway: &gateway::Node, route_id: u64, test_nonce: u64) -> Self { - TestPacket { - pub_key: gateway.identity_key, - owner: gateway.owner.clone(), - route_id, - test_nonce, - node_type: NodeType::Gateway, - } + pub fn mix_plaintexts( + &self, + node: &mix::Node, + test_packets: u32, + ) -> Result>, NetworkTestingError> { + NodeTestMessage::mix_plaintexts(node, test_packets, *self) } - pub(crate) fn new( - pub_key: identity::PublicKey, - owner: String, - route_id: u64, - test_nonce: u64, - node_type: NodeType, - ) -> Self { - TestPacket { - route_id, - test_nonce, - pub_key, - owner, - node_type, - } - } - - pub(crate) fn test_nonce(&self) -> u64 { - self.test_nonce - } - - pub(crate) fn to_bytes(&self) -> Vec { - IntoIterator::into_iter(self.route_id.to_be_bytes()) - .chain(IntoIterator::into_iter(self.test_nonce.to_be_bytes())) - .chain(self.node_type.into_bytes().iter().cloned()) - .chain(self.pub_key.to_bytes().iter().cloned()) - .chain(self.owner.as_bytes().iter().cloned()) - .collect() - } - - pub(crate) fn try_from_bytes(b: &[u8]) -> Result { - // route id + test nonce size - let n = mem::size_of::(); - - if b.len() < 2 * n + 1 + identity::PUBLIC_KEY_LENGTH { - return Err(TestPacketError::IncompletePacket { - received: b.len(), - min_expected: 2 * n + 1 + identity::PUBLIC_KEY_LENGTH, - }); - } - - // those unwraps can't fail as we've already checked for the size - let route_id = u64::from_be_bytes(b[0..n].try_into().unwrap()); - let test_nonce = u64::from_be_bytes(b[n..2 * n].try_into().unwrap()); - let node_type = NodeType::try_from_bytes(&b[2 * n..])?; - let type_size = node_type.size(); - - let pub_key = identity::PublicKey::from_bytes( - &b[2 * n + type_size..2 * n + type_size + identity::PUBLIC_KEY_LENGTH], - )?; - let owner = std::str::from_utf8(&b[2 * n + type_size + identity::PUBLIC_KEY_LENGTH..])?; - - Ok(TestPacket { - route_id, - node_type, - test_nonce, - pub_key, - owner: owner.to_owned(), - }) - } -} - -impl From for TestedNode { - fn from(packet: TestPacket) -> Self { - TestedNode { - identity: packet.pub_key.to_base58_string(), - owner: packet.owner, - node_type: packet.node_type, - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_packet_roundtrip() { - let mut rng = rand_07::thread_rng(); - let dummy_keypair = identity::KeyPair::new(&mut rng); - let owner = "some owner".to_string(); - let mix_packet = TestPacket::new( - *dummy_keypair.public_key(), - owner.clone(), - 42, - 123, - NodeType::Mixnode(1234), - ); - - let bytes = mix_packet.to_bytes(); - let recovered = TestPacket::try_from_bytes(&bytes).unwrap(); - assert_eq!(mix_packet, recovered); - - let gateway_packet = TestPacket::new( - *dummy_keypair.public_key(), - owner, - 42, - 123, - NodeType::Gateway, - ); - - let bytes = gateway_packet.to_bytes(); - let recovered = TestPacket::try_from_bytes(&bytes).unwrap(); - assert_eq!(gateway_packet, recovered); + pub fn gateway_plaintexts( + &self, + node: &gateway::Node, + test_packets: u32, + ) -> Result>, NetworkTestingError> { + NodeTestMessage::gateway_plaintexts(node, test_packets, *self) } } diff --git a/nym-api/src/network_monitor/test_route/mod.rs b/nym-api/src/network_monitor/test_route/mod.rs index c4b98ac5a0..26503227cc 100644 --- a/nym-api/src/network_monitor/test_route/mod.rs +++ b/nym-api/src/network_monitor/test_route/mod.rs @@ -1,7 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::network_monitor::test_packet::{NodeType, TestPacket}; +use crate::network_monitor::test_packet::NymApiTestMessageExt; use crate::network_monitor::ROUTE_TESTING_TEST_NONCE; use nym_crypto::asymmetric::identity; use nym_topology::{gateway, mix, NymTopology}; @@ -70,30 +70,21 @@ impl TestRoute { &self.nodes } - pub(crate) fn self_test_packet(&self) -> TestPacket { + pub(crate) fn test_message_ext(&self, test_nonce: u64) -> NymApiTestMessageExt { + NymApiTestMessageExt::new(self.id, test_nonce) + } + + pub(crate) fn self_test_messages(&self, count: usize) -> Vec> { // it doesn't really matter which node is "chosen" as the packet has to always // go through the same sequence of hops. // let's just use layer 1 mixnode for this (this choice is completely arbitrary) - let mix = &self.nodes.mixes()[&1][0]; - TestPacket::new( - mix.identity_key, - mix.owner.clone(), - self.id, - ROUTE_TESTING_TEST_NONCE, - NodeType::Mixnode(mix.mix_id), - ) - } + let mix = self.layer_one_mix(); - pub(crate) fn substitute_mix(&self, node: &mix::Node) -> NymTopology { - let mut topology = self.nodes.clone(); - topology.set_mixes_in_layer(node.layer as u8, vec![node.clone()]); - topology - } - - pub(crate) fn substitute_gateway(&self, gateway: &gateway::Node) -> NymTopology { - let mut topology = self.nodes.clone(); - topology.set_gateways(vec![gateway.clone()]); - topology + // the unwrap here is fine as the failure can only occur due to serialization and we're not + // using any custom implementations + NymApiTestMessageExt::new(self.id, ROUTE_TESTING_TEST_NONCE) + .mix_plaintexts(mix, count as u32) + .unwrap() } } @@ -104,16 +95,10 @@ impl Debug for TestRoute { "[v{}] Route {}: [G] {} => [M1] {} => [M2] {} => [M3] {}", self.system_version, self.id, - self.nodes.gateways()[0].identity().to_base58_string(), - self.nodes.mixes_in_layer(1)[0] - .identity_key - .to_base58_string(), - self.nodes.mixes_in_layer(2)[0] - .identity_key - .to_base58_string(), - self.nodes.mixes_in_layer(3)[0] - .identity_key - .to_base58_string() + self.gateway().identity().to_base58_string(), + self.layer_one_mix().identity_key.to_base58_string(), + self.layer_two_mix().identity_key.to_base58_string(), + self.layer_three_mix().identity_key.to_base58_string() ) } } From b94f2a483d89bc7eb1ca615383ae179e94b8b995 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Tue, 18 Apr 2023 17:21:29 +0100 Subject: [PATCH 2/5] nym-api compiling again --- common/node-tester-utils/src/node.rs | 14 +- common/node-tester-utils/src/processor.rs | 16 ++- common/node-tester-utils/src/receiver.rs | 27 +++- nym-api/src/network_monitor/chunker.rs | 79 ---------- .../src/network_monitor/gateways_reader.rs | 3 +- nym-api/src/network_monitor/mod.rs | 10 +- nym-api/src/network_monitor/monitor/mod.rs | 11 +- .../src/network_monitor/monitor/preparer.rs | 136 ++++-------------- .../src/network_monitor/monitor/processor.rs | 97 +++++-------- .../monitor/summary_producer.rs | 132 +++++++---------- nym-api/src/network_monitor/test_packet.rs | 10 +- 11 files changed, 173 insertions(+), 362 deletions(-) delete mode 100644 nym-api/src/network_monitor/chunker.rs diff --git a/common/node-tester-utils/src/node.rs b/common/node-tester-utils/src/node.rs index 70d328cfba..9d4e185d59 100644 --- a/common/node-tester-utils/src/node.rs +++ b/common/node-tester-utils/src/node.rs @@ -6,7 +6,7 @@ use nym_topology::{gateway, mix}; use serde::{Deserialize, Serialize}; use std::fmt::{Display, Formatter}; -#[derive(Serialize, Deserialize, Debug, Clone, Hash)] +#[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq)] pub struct TestableNode { pub encoded_identity: String, pub owner: String, @@ -31,6 +31,10 @@ impl TestableNode { pub fn new_gateway(encoded_identity: String, owner: String) -> Self { TestableNode::new(encoded_identity, owner, NodeType::Gateway) } + + pub fn is_mixnode(&self) -> bool { + self.typ.is_mixnode() + } } impl<'a> From<&'a mix::Node> for TestableNode { @@ -65,13 +69,19 @@ impl Display for TestableNode { } } -#[derive(Serialize, Deserialize, Hash, Clone, Copy, Debug)] +#[derive(Serialize, Deserialize, Hash, Clone, Copy, Debug, Eq, PartialEq)] #[serde(rename_all = "snake_case")] pub enum NodeType { Mixnode { mix_id: MixId }, Gateway, } +impl NodeType { + pub fn is_mixnode(&self) -> bool { + matches!(self, NodeType::Mixnode { .. }) + } +} + impl Display for NodeType { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { match self { diff --git a/common/node-tester-utils/src/processor.rs b/common/node-tester-utils/src/processor.rs index 809af452d5..0146c25258 100644 --- a/common/node-tester-utils/src/processor.rs +++ b/common/node-tester-utils/src/processor.rs @@ -45,12 +45,7 @@ impl TestPacketProcessor { local_encryption_keypair: Arc, ack_key: Arc, ) -> Self { - TestPacketProcessor { - local_encryption_keypair, - ack_key, - message_receiver: SphinxMessageReceiver::new(), - _ext_phantom: PhantomData, - } + Self::new(local_encryption_keypair, ack_key) } } @@ -58,6 +53,15 @@ impl TestPacketProcessor where R: MessageReceiver, { + pub fn new(local_encryption_keypair: Arc, ack_key: Arc) -> Self { + TestPacketProcessor { + local_encryption_keypair, + ack_key, + message_receiver: R::new(), + _ext_phantom: PhantomData, + } + } + pub fn process_mixnet_message( &mut self, mut raw_message: Vec, diff --git a/common/node-tester-utils/src/receiver.rs b/common/node-tester-utils/src/receiver.rs index 007b76f988..bf659bf0a5 100644 --- a/common/node-tester-utils/src/receiver.rs +++ b/common/node-tester-utils/src/receiver.rs @@ -35,21 +35,36 @@ impl SimpleMessageReceiver { acks_receiver: mpsc::UnboundedReceiver>>, received_sender: ReceivedSender, shutdown: TaskClient, + ) -> Self { + Self::new( + local_encryption_keypair, + ack_key, + mixnet_message_receiver, + acks_receiver, + received_sender, + shutdown, + ) + } +} + +impl SimpleMessageReceiver { + pub fn new( + local_encryption_keypair: Arc, + ack_key: Arc, + mixnet_message_receiver: mpsc::UnboundedReceiver>>, + acks_receiver: mpsc::UnboundedReceiver>>, + received_sender: ReceivedSender, + shutdown: TaskClient, ) -> Self { SimpleMessageReceiver { - message_processor: TestPacketProcessor::new_sphinx_processor( - local_encryption_keypair, - ack_key, - ), + message_processor: TestPacketProcessor::new(local_encryption_keypair, ack_key), mixnet_message_receiver, acks_receiver, received_sender, shutdown, } } -} -impl SimpleMessageReceiver { fn forward_received>>(&self, received: U) { // TODO: remove the unwrap once/if we do graceful shutdowns here self.received_sender diff --git a/nym-api/src/network_monitor/chunker.rs b/nym-api/src/network_monitor/chunker.rs deleted file mode 100644 index ef4f88e472..0000000000 --- a/nym-api/src/network_monitor/chunker.rs +++ /dev/null @@ -1,79 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use nym_sphinx::forwarding::packet::MixPacket; -use nym_sphinx::message::NymMessage; -use nym_sphinx::params::PacketSize; -use nym_sphinx::{ - acknowledgements::AckKey, addressing::clients::Recipient, preparer::MessagePreparer, -}; -use nym_topology::NymTopology; -use rand_07::rngs::OsRng; -use std::time::Duration; - -const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200); -const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200); - -#[derive(Clone)] -#[deprecated] -pub(crate) struct Chunker { - rng: OsRng, - packet_size: PacketSize, - message_preparer: MessagePreparer, -} - -impl Chunker { - pub(crate) fn new(tested_mix_me: Recipient) -> Self { - Chunker { - rng: OsRng, - // no point in using anything else for monitoring - // unless we should make it variable so mixnodes wouldn't know if - // non-default packet is for measurement or not - packet_size: PacketSize::RegularPacket, - message_preparer: MessagePreparer::new( - OsRng, - tested_mix_me, - DEFAULT_AVERAGE_PACKET_DELAY, - DEFAULT_AVERAGE_ACK_DELAY, - ), - } - } - - pub(crate) fn prepare_packets_from( - &mut self, - message: Vec, - topology: &NymTopology, - packet_sender: Recipient, - ) -> Vec { - // I really dislike how we have to overwrite the parameter of the `MessagePreparer` on each run - // but without some significant API changes in the `MessagePreparer` this was the easiest - // way to being able to have variable sender address. - self.message_preparer.set_sender_address(packet_sender); - self.prepare_packets(message, topology, packet_sender) - } - - fn prepare_packets( - &mut self, - message: Vec, - topology: &NymTopology, - packet_sender: Recipient, - ) -> Vec { - let ack_key: AckKey = AckKey::new(&mut self.rng); - - let split_message = self - .message_preparer - .pad_and_split_message(NymMessage::new_plain(message), self.packet_size); - - let mut mix_packets = Vec::with_capacity(split_message.len()); - for message_chunk in split_message { - // don't bother with acks etc. for time being - let prepared_fragment = self - .message_preparer - .prepare_chunk_for_sending(message_chunk, topology, &ack_key, &packet_sender) - .unwrap(); - - mix_packets.push(prepared_fragment.mix_packet); - } - mix_packets - } -} diff --git a/nym-api/src/network_monitor/gateways_reader.rs b/nym-api/src/network_monitor/gateways_reader.rs index 38956bade1..37a6fba194 100644 --- a/nym-api/src/network_monitor/gateways_reader.rs +++ b/nym-api/src/network_monitor/gateways_reader.rs @@ -48,7 +48,6 @@ impl Stream for GatewaysReader { type Item = (IdentityKey, GatewayMessages); fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // exhaust the ack map if possible match Pin::new(&mut self.ack_map).poll_next(cx) { Poll::Ready(None) => { // this should have never happened! @@ -56,7 +55,7 @@ impl Stream for GatewaysReader { } Poll::Ready(Some(ack_item)) => { // wake immediately in case there's an associated data message - cx.waker().wake(); + cx.waker().wake_by_ref(); return Poll::Ready(Some((ack_item.0, GatewayMessages::Acks(ack_item.1)))); } Poll::Pending => (), diff --git a/nym-api/src/network_monitor/mod.rs b/nym-api/src/network_monitor/mod.rs index f19cccc35e..6c2cb7c31f 100644 --- a/nym-api/src/network_monitor/mod.rs +++ b/nym-api/src/network_monitor/mod.rs @@ -20,11 +20,11 @@ use futures::channel::mpsc; use nym_bandwidth_controller::BandwidthController; use nym_credential_storage::persistent_storage::PersistentStorage; use nym_crypto::asymmetric::{encryption, identity}; +use nym_sphinx::acknowledgements::AckKey; use nym_sphinx::receiver::MessageReceiver; use nym_task::TaskManager; use std::sync::Arc; -pub(crate) mod chunker; pub(crate) mod gateways_reader; pub(crate) mod monitor; pub(crate) mod test_packet; @@ -83,6 +83,7 @@ impl<'a> NetworkMonitorBuilder<'a> { let identity_keypair = Arc::new(identity::KeyPair::new(&mut rng)); let encryption_keypair = Arc::new(encryption::KeyPair::new(&mut rng)); + let ack_key = Arc::new(AckKey::new(&mut rng)); let (gateway_status_update_sender, gateway_status_update_receiver) = mpsc::unbounded(); let (received_processor_sender_channel, received_processor_receiver_channel) = @@ -92,6 +93,7 @@ impl<'a> NetworkMonitorBuilder<'a> { &self.system_version, self.validator_cache, self.config.get_per_node_test_packets(), + Arc::clone(&ack_key), *identity_keypair.public_key(), *encryption_keypair.public_key(), ); @@ -118,6 +120,7 @@ impl<'a> NetworkMonitorBuilder<'a> { let received_processor = new_received_processor( received_processor_receiver_channel, Arc::clone(&encryption_keypair), + ack_key, ); let summary_producer = new_summary_producer(self.config.get_per_node_test_packets()); let packet_receiver = new_packet_receiver( @@ -164,6 +167,7 @@ fn new_packet_preparer( system_version: &str, validator_cache: NymContractCache, per_node_test_packets: usize, + ack_key: Arc, self_public_identity: identity::PublicKey, self_public_encryption: encryption::PublicKey, ) -> PacketPreparer { @@ -171,6 +175,7 @@ fn new_packet_preparer( system_version, validator_cache, per_node_test_packets, + ack_key, self_public_identity, self_public_encryption, ) @@ -199,8 +204,9 @@ fn new_packet_sender( fn new_received_processor( packets_receiver: ReceivedProcessorReceiver, client_encryption_keypair: Arc, + ack_key: Arc, ) -> ReceivedProcessor { - ReceivedProcessor::new(packets_receiver, client_encryption_keypair) + ReceivedProcessor::new(packets_receiver, client_encryption_keypair, ack_key) } fn new_summary_producer(per_node_test_packets: usize) -> SummaryProducer { diff --git a/nym-api/src/network_monitor/monitor/mod.rs b/nym-api/src/network_monitor/monitor/mod.rs index ebc7174d2a..787317db6c 100644 --- a/nym-api/src/network_monitor/monitor/mod.rs +++ b/nym-api/src/network_monitor/monitor/mod.rs @@ -1,11 +1,11 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use crate::network_monitor::monitor::preparer::PacketPreparer; use crate::network_monitor::monitor::processor::ReceivedProcessor; use crate::network_monitor::monitor::sender::PacketSender; use crate::network_monitor::monitor::summary_producer::{SummaryProducer, TestSummary}; -use crate::network_monitor::test_packet::TestPacket; +use crate::network_monitor::test_packet::NodeTestMessage; use crate::network_monitor::test_route::TestRoute; use crate::storage::NymApiStorage; use crate::support::config::Config; @@ -96,10 +96,13 @@ impl Monitor { } } - fn analyse_received_test_route_packets(&self, packets: &[TestPacket]) -> HashMap { + fn analyse_received_test_route_packets( + &self, + packets: &[NodeTestMessage], + ) -> HashMap { let mut received = HashMap::new(); for packet in packets { - *received.entry(packet.route_id).or_insert(0usize) += 1usize + *received.entry(packet.ext.route_id).or_insert(0usize) += 1usize } received diff --git a/nym-api/src/network_monitor/monitor/preparer.rs b/nym-api/src/network_monitor/monitor/preparer.rs index 7183c9f4c4..ecd9926f96 100644 --- a/nym-api/src/network_monitor/monitor/preparer.rs +++ b/nym-api/src/network_monitor/monitor/preparer.rs @@ -1,21 +1,19 @@ -// Copyright 2021 - Nym Technologies SA +// Copyright 2021-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::network_monitor::chunker::Chunker; use crate::network_monitor::monitor::sender::GatewayPackets; -use crate::network_monitor::test_packet::{NodeTestMessage, NymApiTestMessageExt}; use crate::network_monitor::test_route::TestRoute; use crate::nym_contract_cache::cache::NymContractCache; use log::info; use nym_crypto::asymmetric::{encryption, identity}; -use nym_mixnet_contract_common::{Addr, GatewayBond, IdentityKey, Layer, MixId, MixNodeBond}; -use nym_node_tester_utils::node::{NodeType, TestableNode}; +use nym_mixnet_contract_common::{GatewayBond, Layer, MixNodeBond}; +use nym_node_tester_utils::node::TestableNode; use nym_node_tester_utils::NodeTester; use nym_sphinx::acknowledgements::AckKey; use nym_sphinx::addressing::clients::Recipient; use nym_sphinx::forwarding::packet::MixPacket; use nym_sphinx::params::PacketSize; -use nym_topology::{gateway, mix, NymTopology}; +use nym_topology::{gateway, mix}; use rand_07::{rngs::ThreadRng, seq::SliceRandom, thread_rng, Rng}; use std::collections::{HashMap, HashSet}; use std::convert::TryInto; @@ -45,28 +43,14 @@ impl Display for InvalidNode { } } -// impl InvalidNode { -// pub(crate) fn mix_id(&self) -> Option { -// match self { -// InvalidNode::Outdated(_, _, node_type, _) => node_type.mix_id(), -// InvalidNode::Malformed(_, _, node_type) => node_type.mix_id(), -// } -// } -// -// pub(crate) fn identity(&self) -> String { -// match self { -// InvalidNode::Outdated(id, _, _, _) => id.clone(), -// InvalidNode::Malformed(id, _, _) => id.clone(), -// } -// } -// -// pub(crate) fn owner(&self) -> String { -// match self { -// InvalidNode::Outdated(_, owner, _, _) => owner.into(), -// InvalidNode::Malformed(_, owner, _) => owner.into(), -// } -// } -// } +impl From for TestableNode { + fn from(value: InvalidNode) -> Self { + match value { + InvalidNode::Outdated { node, .. } => node, + InvalidNode::Malformed { node } => node, + } + } +} pub(crate) struct PreparedPackets { /// All packets that are going to get sent during the test as well as the gateways through @@ -96,6 +80,7 @@ pub(crate) struct PacketPreparer { /// Number of test packets sent to each node per_node_test_packets: usize, + ack_key: Arc, // TODO: security: // in the future we should really create unique set of keys every time otherwise // gateways might recognise our "test" keys and take special care to always forward those packets @@ -109,18 +94,18 @@ impl PacketPreparer { system_version: &str, validator_cache: NymContractCache, per_node_test_packets: usize, + ack_key: Arc, self_public_identity: identity::PublicKey, self_public_encryption: encryption::PublicKey, ) -> Self { - todo!() - // PacketPreparer { - // system_version: system_version.to_owned(), - // chunker: None, - // validator_cache, - // per_node_test_packets, - // self_public_identity, - // self_public_encryption, - // } + PacketPreparer { + system_version: system_version.to_owned(), + validator_cache, + per_node_test_packets, + ack_key, + self_public_identity, + self_public_encryption, + } } fn ephemeral_tester( @@ -128,11 +113,7 @@ impl PacketPreparer { test_route: &TestRoute, self_address: Option, ) -> NodeTester { - let mut rng = thread_rng(); - - // TODO: we should make this key more long-term - let ack_key = Arc::new(AckKey::new(&mut rng)); - + let rng = thread_rng(); NodeTester::new( rng, // the topology here contains 3 mixnodes and 1 gateway so its cheap to clone it @@ -141,7 +122,7 @@ impl PacketPreparer { PacketSize::RegularPacket, DEFAULT_AVERAGE_PACKET_DELAY, DEFAULT_AVERAGE_ACK_DELAY, - ack_key, + self.ack_key.clone(), ) } @@ -342,26 +323,6 @@ impl PacketPreparer { ) } - // pub(crate) fn prepare_test_route_viability_packets( - // &mut self, - // route: &TestRoute, - // num: usize, - // ) -> GatewayPackets { - // let mut mix_packets = Vec::with_capacity(num); - // let test_packet = route.self_test_packet(); - // let recipient = self.create_packet_sender(route.gateway()); - // for _ in 0..num { - // let mix_packet = self.wrap_test_packet(&test_packet, route.topology(), recipient); - // mix_packets.push(mix_packet) - // } - // - // GatewayPackets::new( - // route.gateway_clients_address(), - // route.gateway_identity(), - // mix_packets, - // ) - // } - fn filter_outdated_and_malformed_mixnodes( &self, nodes: Vec, @@ -434,7 +395,6 @@ impl PacketPreparer { let gateway_address = test_route.gateway_clients_address(); let gateway_identity = test_route.gateway_identity(); - let recipient = self.create_packet_sender(test_route.gateway()); let mut mix_tester = self.ephemeral_mix_tester(test_route); // generate test packets for mixnodes @@ -486,54 +446,6 @@ impl PacketPreparer { .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity)); gateway_packets.push_packets(gateway_mix_packets); } - - // - // - // - - // // let gateway_identity = test_route.gateway_identity(); - // let gateway_address = test_route.gateway_clients_address(); - // - // // it's actually going to be a tiny bit more due to gateway testing, but it's a good enough approximation - // let mut mix_packets = Vec::with_capacity(mixnodes.len() * self.per_node_test_packets); - // - // // and for each mixnode... - // for mixnode in &mixnodes { - // let test_packet = TestPacket::from_mixnode(mixnode, test_route.id(), test_nonce); - // let topology = test_route.substitute_mix(mixnode); - // // produce n mix packets - // for _ in 0..self.per_node_test_packets { - // let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient); - // mix_packets.push(mix_packet); - // } - // } - // - // let gateway_packets = all_gateway_packets - // .entry(gateway_identity.to_bytes()) - // .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity)); - // gateway_packets.push_packets(mix_packets); - // - // // and for each gateway... - // for gateway in &gateways { - // let mut gateway_mix_packets = Vec::new(); - // let test_packet = TestPacket::from_gateway(gateway, test_route.id(), test_nonce); - // let gateway_identity = gateway.identity_key; - // let gateway_address = gateway.clients_address(); - // let recipient = self.create_packet_sender(gateway); - // let topology = test_route.substitute_gateway(gateway); - // // produce n mix packets - // for _ in 0..self.per_node_test_packets { - // let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient); - // gateway_mix_packets.push(mix_packet); - // } - // - // // and push it into existing struct (if it's a "core" gateway being tested against another route) - // // or create a new one - // let gateway_packets = all_gateway_packets - // .entry(gateway_identity.to_bytes()) - // .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity)); - // gateway_packets.push_packets(gateway_mix_packets); - // } } // convert our hashmap back into a vec diff --git a/nym-api/src/network_monitor/monitor/processor.rs b/nym-api/src/network_monitor/monitor/processor.rs index 5900543253..d7a24318d5 100644 --- a/nym-api/src/network_monitor/monitor/processor.rs +++ b/nym-api/src/network_monitor/monitor/processor.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::network_monitor::gateways_reader::GatewayMessages; -use crate::network_monitor::test_packet::NymApiTestMessageExt; +use crate::network_monitor::test_packet::{NodeTestMessage, NymApiTestMessageExt}; use crate::network_monitor::ROUTE_TESTING_TEST_NONCE; use futures::channel::mpsc; use futures::lock::{Mutex, MutexGuard}; @@ -11,6 +11,7 @@ use log::warn; use nym_crypto::asymmetric::encryption; use nym_node_tester_utils::error::NetworkTestingError; use nym_node_tester_utils::processor::TestPacketProcessor; +use nym_sphinx::acknowledgements::AckKey; use nym_sphinx::receiver::{MessageReceiver, MessageRecoveryError}; use std::mem; use std::sync::Arc; @@ -26,9 +27,6 @@ enum ProcessingError { )] MalformedPacketReceived(#[from] MessageRecoveryError), - #[error("received a mix packet that was NOT a proper network monitor test packet")] - NonTestPacketReceived, - #[error("the received test packet was malformed: {0}")] MalformedTestPacket(#[from] NetworkTestingError), @@ -53,17 +51,20 @@ struct ReceivedProcessorInner { packets_receiver: ReceivedProcessorReceiver, test_processor: TestPacketProcessor, - // TODO: rethinking - // /// Vector containing all received (and decrypted) packets in the current test run. - // received_packets: Vec, + + /// Vector containing all received (and decrypted) packets in the current test run. + // TODO: perhaps a different structure would be better here + received_packets: Vec, } impl ReceivedProcessorInner { - fn recover_test_message(&mut self, raw_message: Vec) { - // - } - fn on_received_data(&mut self, raw_message: Vec) -> Result<(), ProcessingError> { + // if the nonce is none it means the packet was received during the 'waiting' for the + // next test run + if self.test_nonce.is_none() { + return Err(ProcessingError::ReceivedOutsideTestRun); + } + let test_msg = self.test_processor.process_mixnet_message(raw_message)?; if test_msg.ext.test_nonce != self.test_nonce.unwrap() { @@ -73,10 +74,17 @@ impl ReceivedProcessorInner { }); } - todo!() + self.received_packets.push(test_msg); + Ok(()) } fn on_received_ack(&mut self, raw_ack: Vec) -> Result<(), ProcessingError> { + // if the nonce is none it means the packet was received during the 'waiting' for the + // next test run + if self.test_nonce.is_none() { + return Err(ProcessingError::ReceivedOutsideTestRun); + } + let frag_id = self.test_processor.process_ack(raw_ack)?; // TODO: hook it up at some point trace!("received a test ack with id {frag_id}. However, we're not goingn to do anything about it (just yet)"); @@ -103,40 +111,7 @@ impl ReceivedProcessorInner { } } - fn on_message(&mut self, mut message: Vec) -> Result<(), ProcessingError> { - // if the nonce is none it means the packet was received during the 'waiting' for the - // next test run - if self.test_nonce.is_none() { - return Err(ProcessingError::ReceivedOutsideTestRun); - } - - let plaintext = self - .message_receiver - .recover_plaintext_from_regular_packet( - self.client_encryption_keypair.private_key(), - &mut message, - )?; - let fragment = self.message_receiver.recover_fragment(plaintext)?; - let (recovered, _) = self - .message_receiver - .insert_new_fragment(fragment)? - .ok_or(ProcessingError::NonTestPacketReceived)?; // if it's a test packet it MUST BE reconstructed with single fragment - let test_packet = TestPacket::try_from_bytes(&recovered.into_inner_data())?; - - // we know nonce is NOT none - if test_packet.test_nonce() != self.test_nonce.unwrap() { - return Err(ProcessingError::NonMatchingNonce { - received: test_packet.test_nonce(), - expected: self.test_nonce.unwrap(), - }); - } - - self.received_packets.push(test_packet); - - Ok(()) - } - - fn finish_run(&mut self) -> Vec { + fn finish_run(&mut self) -> Vec { self.test_nonce = None; mem::take(&mut self.received_packets) } @@ -151,22 +126,20 @@ impl ReceivedProcessor { pub(crate) fn new( packets_receiver: ReceivedProcessorReceiver, client_encryption_keypair: Arc, + ack_key: Arc, ) -> Self { - todo!() - // - // let inner: Arc>> = - // Arc::new(Mutex::new(ReceivedProcessorInner { - // test_nonce: None, - // packets_receiver, - // client_encryption_keypair, - // message_receiver: R::new(), - // received_packets: Vec::new(), - // })); - // - // ReceivedProcessor { - // permit_changer: None, - // inner, - // } + let inner: Arc>> = + Arc::new(Mutex::new(ReceivedProcessorInner { + test_nonce: None, + packets_receiver, + test_processor: TestPacketProcessor::new(client_encryption_keypair, ack_key), + received_packets: Vec::new(), + })); + + ReceivedProcessor { + permit_changer: None, + inner, + } } pub(crate) fn start_receiving(&mut self) { @@ -247,7 +220,7 @@ impl ReceivedProcessor { .expect("processing task has died!"); } - pub(super) async fn return_received(&mut self) -> Vec { + pub(super) async fn return_received(&mut self) -> Vec { // ask for the lock back self.permit_changer .as_mut() diff --git a/nym-api/src/network_monitor/monitor/summary_producer.rs b/nym-api/src/network_monitor/monitor/summary_producer.rs index 4096900cd5..c4ed5c6dc5 100644 --- a/nym-api/src/network_monitor/monitor/summary_producer.rs +++ b/nym-api/src/network_monitor/monitor/summary_producer.rs @@ -2,14 +2,13 @@ // SPDX-License-Identifier: Apache-2.0 use crate::network_monitor::monitor::preparer::InvalidNode; +use crate::network_monitor::test_packet::NodeTestMessage; use crate::network_monitor::test_route::TestRoute; use nym_mixnet_contract_common::MixId; -use nym_node_tester_utils::node::TestableNode; +use nym_node_tester_utils::node::{NodeType, TestableNode}; use std::collections::HashMap; use std::fmt::{Display, Formatter}; -const INVALID_MIX_ID: u32 = u32::MAX; - // just some approximate measures to print to stdout (well, technically stderr since it's being printed via log) const EXCEPTIONAL_THRESHOLD: u8 = 95; // 95 - 100 const FINE_THRESHOLD: u8 = 80; // 80 - 95 @@ -60,12 +59,19 @@ impl GatewayResult { #[derive(Debug, Clone)] pub(crate) struct RouteResult { pub(crate) route: TestRoute, + + #[deprecated] reliability: u8, + performance: f32, } impl RouteResult { - pub(crate) fn new(route: TestRoute, reliability: u8) -> Self { - RouteResult { route, reliability } + pub(crate) fn new(route: TestRoute, reliability: u8, performance: f32) -> Self { + RouteResult { + route, + reliability, + performance, + } } } @@ -283,106 +289,76 @@ impl SummaryProducer { &self, tested_mixnodes: Vec, tested_gateways: Vec, - received_packets: Vec, + received_packets: Vec, invalid_mixnodes: Vec, invalid_gateways: Vec, test_routes: &[TestRoute], ) -> TestSummary { - let mut raw_mixnode_results = HashMap::new(); - let mut raw_gateway_results = HashMap::new(); - - let mut raw_route_results = HashMap::new(); - // we expect each route to receive this many packets in the ideal world let per_route_expected = (tested_mixnodes.len() + tested_gateways.len()) * self.per_node_test_packets; let per_node_expected = test_routes.len() * self.per_node_test_packets; - // TODO: whenever somebody feels like it, this should really get refactored. - // probably some wrapper struct would be appropriate here. - for tested_mixnode in tested_mixnodes { - raw_mixnode_results.insert( - ( - tested_mixnode.mix_id().unwrap_or(INVALID_MIX_ID), - tested_mixnode.identity, - tested_mixnode.owner, - ), - 0, - ); - } - - for tested_gateway in tested_gateways { - raw_gateway_results.insert((tested_gateway.identity, tested_gateway.owner), 0); - } - - for invalid_mixnode in invalid_mixnodes { - raw_mixnode_results.insert( - ( - invalid_mixnode.mix_id().unwrap_or(INVALID_MIX_ID), - invalid_mixnode.identity(), - invalid_mixnode.owner(), - ), - 0, - ); - } - - for invalid_gateway in invalid_gateways { - raw_gateway_results.insert((invalid_gateway.identity(), invalid_gateway.owner()), 0); - } - + let mut raw_route_results = HashMap::new(); for test_route in test_routes { raw_route_results.insert(test_route.id(), 0); } - for received in received_packets { - let pub_key = received.pub_key.to_base58_string(); + let mut raw_results = HashMap::new(); - match received.node_type { - NodeType::Mixnode(mix_id) => { - *raw_mixnode_results - .entry((mix_id, pub_key, received.owner)) - .or_default() += 1usize; - } - NodeType::Gateway => { - *raw_gateway_results - .entry((pub_key, received.owner)) - .or_default() += 1usize; - } - } - - *raw_route_results.entry(received.route_id).or_default() += 1usize; + for tested_mixnode in tested_mixnodes { + raw_results.insert(tested_mixnode, 0); } - let mixnode_results = raw_mixnode_results - .into_iter() - .map(|((mix_id, identity_key, owner), received)| { - let reliability = - (received as f32 / per_node_expected as f32 * 100.0).round() as u8; - MixnodeResult::new(mix_id, identity_key, owner, reliability) - }) - .collect(); + for tested_gateway in tested_gateways { + raw_results.insert(tested_gateway, 0); + } - let gateway_results = raw_gateway_results - .into_iter() - .map(|((identity_key, owner), received)| { - let reliability = - (received as f32 / per_node_expected as f32 * 100.0).round() as u8; - GatewayResult::new(identity_key, owner, reliability) - }) - .collect(); + for invalid_mixnode in invalid_mixnodes { + raw_results.insert(invalid_mixnode.into(), 0); + } + + for invalid_gateway in invalid_gateways { + raw_results.insert(invalid_gateway.into(), 0); + } + + for received in received_packets { + *raw_results.entry(received.tested_node).or_default() += 1usize; + *raw_route_results.entry(received.ext.route_id).or_default() += 1usize; + } + + let mut mixnode_results = Vec::new(); + let mut gateway_results = Vec::new(); + + for (node, received) in raw_results { + let performance = received as f32 / per_node_expected as f32 * 100.0; + let reliability = performance.round() as u8; + + match node.typ { + NodeType::Mixnode { mix_id } => { + let res = + MixnodeResult::new(mix_id, node.encoded_identity, node.owner, reliability); + mixnode_results.push(res) + } + NodeType::Gateway => { + let res = GatewayResult::new(node.encoded_identity, node.owner, reliability); + gateway_results.push(res) + } + } + } let route_results = raw_route_results .into_iter() .filter_map(|(id, received)| { - let reliability = - (received as f32 / per_route_expected as f32 * 100.0).round() as u8; + let performance = received as f32 / per_route_expected as f32 * 100.0; + let reliability = performance.round() as u8; // this might be suboptimal as we're going through the entire slice every time // but realistically this slice will never have more than ~ 10 elements AT MOST test_routes .iter() .find(|route| route.id() == id) - .map(|route| RouteResult::new(route.clone(), reliability)) + .map(|route| RouteResult::new(route.clone(), reliability, performance)) }) .collect(); diff --git a/nym-api/src/network_monitor/test_packet.rs b/nym-api/src/network_monitor/test_packet.rs index 2bba7ddeb7..289172d89b 100644 --- a/nym-api/src/network_monitor/test_packet.rs +++ b/nym-api/src/network_monitor/test_packet.rs @@ -3,7 +3,7 @@ use nym_node_tester_utils::error::NetworkTestingError; use nym_node_tester_utils::TestMessage; -use nym_topology::{gateway, mix}; +use nym_topology::mix; use serde::{Deserialize, Serialize}; pub(crate) type NodeTestMessage = TestMessage; @@ -29,12 +29,4 @@ impl NymApiTestMessageExt { ) -> Result>, NetworkTestingError> { NodeTestMessage::mix_plaintexts(node, test_packets, *self) } - - pub fn gateway_plaintexts( - &self, - node: &gateway::Node, - test_packets: u32, - ) -> Result>, NetworkTestingError> { - NodeTestMessage::gateway_plaintexts(node, test_packets, *self) - } } From 80d7285497cfff313b9a4381a0a2a5d8809163fc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 19 Apr 2023 11:23:01 +0100 Subject: [PATCH 3/5] further improvements + reduced log noise --- nym-api/src/main.rs | 3 - nym-api/src/network_monitor/mod.rs | 18 +----- nym-api/src/network_monitor/monitor/mod.rs | 2 +- .../src/network_monitor/monitor/preparer.rs | 23 ++------ .../src/network_monitor/monitor/processor.rs | 2 +- .../src/network_monitor/monitor/receiver.rs | 6 +- .../monitor/summary_producer.rs | 56 ++++++++----------- nym-api/src/network_monitor/test_route/mod.rs | 6 +- 8 files changed, 36 insertions(+), 80 deletions(-) diff --git a/nym-api/src/main.rs b/nym-api/src/main.rs index 73776b5dfd..55e1943404 100644 --- a/nym-api/src/main.rs +++ b/nym-api/src/main.rs @@ -58,8 +58,6 @@ async fn main() -> Result<(), Box> { async fn start_nym_api_tasks( config: Config, ) -> Result> { - let system_version = clap::crate_version!(); - let nyxd_client = nyxd::Client::new(&config); let mix_denom = nyxd_client.chain_details().await.mix_denom.base; @@ -133,7 +131,6 @@ async fn start_nym_api_tasks( nym_contract_cache_state, storage, nyxd_client.clone(), - system_version, &shutdown, ) .await; diff --git a/nym-api/src/network_monitor/mod.rs b/nym-api/src/network_monitor/mod.rs index 6c2cb7c31f..03e4ea65e8 100644 --- a/nym-api/src/network_monitor/mod.rs +++ b/nym-api/src/network_monitor/mod.rs @@ -37,12 +37,10 @@ pub(crate) fn setup<'a>( nym_contract_cache_state: &NymContractCache, storage: &NymApiStorage, nyxd_client: nyxd::Client, - system_version: &str, ) -> NetworkMonitorBuilder<'a> { NetworkMonitorBuilder::new( config, nyxd_client, - system_version, storage.to_owned(), nym_contract_cache_state.to_owned(), ) @@ -51,7 +49,6 @@ pub(crate) fn setup<'a>( pub(crate) struct NetworkMonitorBuilder<'a> { config: &'a Config, nyxd_client: nyxd::Client, - system_version: String, node_status_storage: NymApiStorage, validator_cache: NymContractCache, } @@ -60,14 +57,12 @@ impl<'a> NetworkMonitorBuilder<'a> { pub(crate) fn new( config: &'a Config, nyxd_client: nyxd::Client, - system_version: &str, node_status_storage: NymApiStorage, validator_cache: NymContractCache, ) -> Self { NetworkMonitorBuilder { config, nyxd_client, - system_version: system_version.to_string(), node_status_storage, validator_cache, } @@ -90,7 +85,6 @@ impl<'a> NetworkMonitorBuilder<'a> { mpsc::unbounded(); let packet_preparer = new_packet_preparer( - &self.system_version, self.validator_cache, self.config.get_per_node_test_packets(), Arc::clone(&ack_key), @@ -164,7 +158,6 @@ impl NetworkMonitorRunnables { } fn new_packet_preparer( - system_version: &str, validator_cache: NymContractCache, per_node_test_packets: usize, ack_key: Arc, @@ -172,7 +165,6 @@ fn new_packet_preparer( self_public_encryption: encryption::PublicKey, ) -> PacketPreparer { PacketPreparer::new( - system_version, validator_cache, per_node_test_packets, ack_key, @@ -229,16 +221,10 @@ pub(crate) async fn start( nym_contract_cache_state: &NymContractCache, storage: &NymApiStorage, nyxd_client: nyxd::Client, - system_version: &str, shutdown: &TaskManager, ) { - let monitor_builder = network_monitor::setup( - config, - nym_contract_cache_state, - storage, - nyxd_client, - system_version, - ); + let monitor_builder = + network_monitor::setup(config, nym_contract_cache_state, storage, nyxd_client); info!("Starting network monitor..."); let runnables: NetworkMonitorRunnables = monitor_builder.build().await; runnables.spawn_tasks(shutdown); diff --git a/nym-api/src/network_monitor/monitor/mod.rs b/nym-api/src/network_monitor/monitor/mod.rs index 787317db6c..d8ffea0643 100644 --- a/nym-api/src/network_monitor/monitor/mod.rs +++ b/nym-api/src/network_monitor/monitor/mod.rs @@ -261,7 +261,7 @@ impl Monitor { let received = self.received_processor.return_received().await; let total_received = received.len(); - info!("Test routes: {:?}", routes); + info!("Test routes: {:#?}", routes); info!("Received {}/{} packets", total_received, total_sent); let summary = self.summary_producer.produce_summary( diff --git a/nym-api/src/network_monitor/monitor/preparer.rs b/nym-api/src/network_monitor/monitor/preparer.rs index ecd9926f96..7343a06a6e 100644 --- a/nym-api/src/network_monitor/monitor/preparer.rs +++ b/nym-api/src/network_monitor/monitor/preparer.rs @@ -26,16 +26,12 @@ const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200); #[derive(Clone)] pub(crate) enum InvalidNode { - Outdated { node: TestableNode, version: String }, Malformed { node: TestableNode }, } impl Display for InvalidNode { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { match self { - InvalidNode::Outdated { node, version } => { - write!(f, "{node} is outdated. It runs on v{version}") - } InvalidNode::Malformed { node } => { write!(f, "{node} is malformed") } @@ -46,7 +42,6 @@ impl Display for InvalidNode { impl From for TestableNode { fn from(value: InvalidNode) -> Self { match value { - InvalidNode::Outdated { node, .. } => node, InvalidNode::Malformed { node } => node, } } @@ -74,7 +69,6 @@ pub(crate) struct PreparedPackets { #[derive(Clone)] pub(crate) struct PacketPreparer { - system_version: String, validator_cache: NymContractCache, /// Number of test packets sent to each node @@ -91,7 +85,6 @@ pub(crate) struct PacketPreparer { impl PacketPreparer { pub(crate) fn new( - system_version: &str, validator_cache: NymContractCache, per_node_test_packets: usize, ack_key: Arc, @@ -99,7 +92,6 @@ impl PacketPreparer { self_public_encryption: encryption::PublicKey, ) -> Self { PacketPreparer { - system_version: system_version.to_owned(), validator_cache, per_node_test_packets, ack_key, @@ -132,6 +124,7 @@ impl PacketPreparer { self.ephemeral_tester(test_route, Some(self_address)) } + #[allow(dead_code)] fn ephemeral_gateway_tester(&self, test_route: &TestRoute) -> NodeTester { self.ephemeral_tester(test_route, None) } @@ -276,16 +269,12 @@ impl PacketPreparer { continue }; - routes.push(TestRoute::new( - rng.gen(), - &self.system_version, - node_1, - node_2, - node_3, - gateway, - )) + routes.push(TestRoute::new(rng.gen(), node_1, node_2, node_3, gateway)) } - info!("{:?}", routes); + info!( + "The following routes will be used for testing: {:#?}", + routes + ); Some(routes) } diff --git a/nym-api/src/network_monitor/monitor/processor.rs b/nym-api/src/network_monitor/monitor/processor.rs index d7a24318d5..0e38eebd0b 100644 --- a/nym-api/src/network_monitor/monitor/processor.rs +++ b/nym-api/src/network_monitor/monitor/processor.rs @@ -87,7 +87,7 @@ impl ReceivedProcessorInner { let frag_id = self.test_processor.process_ack(raw_ack)?; // TODO: hook it up at some point - trace!("received a test ack with id {frag_id}. However, we're not goingn to do anything about it (just yet)"); + trace!("received a test ack with id {frag_id}. However, we're not going to do anything about it (just yet)"); Ok(()) } diff --git a/nym-api/src/network_monitor/monitor/receiver.rs b/nym-api/src/network_monitor/monitor/receiver.rs index 299044e285..344f8193b7 100644 --- a/nym-api/src/network_monitor/monitor/receiver.rs +++ b/nym-api/src/network_monitor/monitor/receiver.rs @@ -66,11 +66,7 @@ impl PacketReceiver { // unwrap here is fine as it can only return a `None` if the PacketSender has died // and if that was the case, then the entire monitor is already in an undefined state update = self.clients_updater.next() => self.process_gateway_update(update.unwrap()), - gateway_messages = self.gateways_reader.next() => { - let Some((_gateway_id, messages)) = gateway_messages else { - log::error!("the gateways reader stream has terminated!"); - continue - }; + Some((_gateway_id, messages)) = self.gateways_reader.next() => { self.process_gateway_messages(messages) } } diff --git a/nym-api/src/network_monitor/monitor/summary_producer.rs b/nym-api/src/network_monitor/monitor/summary_producer.rs index c4ed5c6dc5..2a68004e7b 100644 --- a/nym-api/src/network_monitor/monitor/summary_producer.rs +++ b/nym-api/src/network_monitor/monitor/summary_producer.rs @@ -59,19 +59,12 @@ impl GatewayResult { #[derive(Debug, Clone)] pub(crate) struct RouteResult { pub(crate) route: TestRoute, - - #[deprecated] - reliability: u8, performance: f32, } impl RouteResult { - pub(crate) fn new(route: TestRoute, reliability: u8, performance: f32) -> Self { - RouteResult { - route, - reliability, - performance, - } + pub(crate) fn new(route: TestRoute, performance: f32) -> Self { + RouteResult { route, performance } } } @@ -185,63 +178,63 @@ impl Display for TestReport { writeln!( f, "{:?}, reliability: {:.2}", - route_result.route, route_result.reliability + route_result.route, route_result.performance )?; } writeln!( f, - "Exceptional mixnodes (reliability >= {}): {}", - EXCEPTIONAL_THRESHOLD, self.exceptional_mixnodes + "Exceptional mixnodes (reliability >= {EXCEPTIONAL_THRESHOLD}): {}", + self.exceptional_mixnodes )?; writeln!( f, - "Exceptional gateways (reliability >= {}): {}", - EXCEPTIONAL_THRESHOLD, self.exceptional_gateways + "Exceptional gateways (reliability >= {EXCEPTIONAL_THRESHOLD}): {}", + self.exceptional_gateways )?; writeln!( f, - "Fine mixnodes (reliability {} - {}): {}", - FINE_THRESHOLD, EXCEPTIONAL_THRESHOLD, self.fine_mixnodes + "Fine mixnodes (reliability {FINE_THRESHOLD} - {EXCEPTIONAL_THRESHOLD}): {}", + self.fine_mixnodes )?; writeln!( f, - "Fine gateways (reliability {} - {}): {}", - FINE_THRESHOLD, EXCEPTIONAL_THRESHOLD, self.fine_gateways + "Fine gateways (reliability {FINE_THRESHOLD} - {EXCEPTIONAL_THRESHOLD}): {}", + self.fine_gateways )?; writeln!( f, - "Poor mixnodes (reliability {} - {}): {}", - POOR_THRESHOLD, FINE_THRESHOLD, self.poor_mixnodes + "Poor mixnodes (reliability {POOR_THRESHOLD} - {FINE_THRESHOLD}): {}", + self.poor_mixnodes )?; writeln!( f, - "Poor gateways (reliability {} - {}): {}", - POOR_THRESHOLD, FINE_THRESHOLD, self.poor_gateways + "Poor gateways (reliability {POOR_THRESHOLD} - {FINE_THRESHOLD}): {}", + self.poor_gateways )?; writeln!( f, - "Unreliable mixnodes (reliability {} - {}): {}", - UNRELIABLE_THRESHOLD, POOR_THRESHOLD, self.unreliable_mixnodes + "Unreliable mixnodes (reliability {UNRELIABLE_THRESHOLD} - {POOR_THRESHOLD}): {}", + self.unreliable_mixnodes )?; writeln!( f, - "Unreliable gateways (reliability {} - {}): {}", - UNRELIABLE_THRESHOLD, POOR_THRESHOLD, self.unreliable_gateways + "Unreliable gateways (reliability {UNRELIABLE_THRESHOLD} - {POOR_THRESHOLD}): {}", + self.unreliable_gateways )?; writeln!( f, - "Unroutable mixnodes (reliability < {}): {}", - UNRELIABLE_THRESHOLD, self.unroutable_mixnodes + "Unroutable mixnodes (reliability < {UNRELIABLE_THRESHOLD}): {}", + self.unroutable_mixnodes )?; writeln!( f, - "Unroutable gateways (reliability < {}): {}", - UNRELIABLE_THRESHOLD, self.unroutable_gateways + "Unroutable gateways (reliability < {UNRELIABLE_THRESHOLD}): {}", + self.unroutable_gateways )?; Ok(()) @@ -351,14 +344,13 @@ impl SummaryProducer { .into_iter() .filter_map(|(id, received)| { let performance = received as f32 / per_route_expected as f32 * 100.0; - let reliability = performance.round() as u8; // this might be suboptimal as we're going through the entire slice every time // but realistically this slice will never have more than ~ 10 elements AT MOST test_routes .iter() .find(|route| route.id() == id) - .map(|route| RouteResult::new(route.clone(), reliability, performance)) + .map(|route| RouteResult::new(route.clone(), performance)) }) .collect(); diff --git a/nym-api/src/network_monitor/test_route/mod.rs b/nym-api/src/network_monitor/test_route/mod.rs index 26503227cc..48eede18b5 100644 --- a/nym-api/src/network_monitor/test_route/mod.rs +++ b/nym-api/src/network_monitor/test_route/mod.rs @@ -10,14 +10,12 @@ use std::fmt::{Debug, Formatter}; #[derive(Clone)] pub(crate) struct TestRoute { id: u64, - system_version: String, nodes: NymTopology, } impl TestRoute { pub(crate) fn new( id: u64, - system_version: &str, l1_mix: mix::Node, l2_mix: mix::Node, l3_mix: mix::Node, @@ -33,7 +31,6 @@ impl TestRoute { TestRoute { id, - system_version: system_version.to_string(), nodes: NymTopology::new(layered_mixes, vec![gateway]), } } @@ -92,8 +89,7 @@ impl Debug for TestRoute { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!( f, - "[v{}] Route {}: [G] {} => [M1] {} => [M2] {} => [M3] {}", - self.system_version, + "Route {}: [G] {} => [M1] {} => [M2] {} => [M3] {}", self.id, self.gateway().identity().to_base58_string(), self.layer_one_mix().identity_key.to_base58_string(), From 9b36bccf0c1791e9816a830a27b280f1eeba3724 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 19 Apr 2023 11:26:41 +0100 Subject: [PATCH 4/5] wasm tester fixes --- clients/webassembly/src/tester/ephemeral_receiver.rs | 3 ++- clients/webassembly/src/tester/helpers.rs | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/clients/webassembly/src/tester/ephemeral_receiver.rs b/clients/webassembly/src/tester/ephemeral_receiver.rs index 1403176b36..4f2ce3cd8a 100644 --- a/clients/webassembly/src/tester/ephemeral_receiver.rs +++ b/clients/webassembly/src/tester/ephemeral_receiver.rs @@ -3,7 +3,8 @@ use crate::tester::helpers::{NodeTestResult, WasmTestMessageExt}; use futures::StreamExt; -use nym_node_tester_utils::receiver::{Received, ReceivedReceiver}; +use nym_node_tester_utils::processor::Received; +use nym_node_tester_utils::receiver::ReceivedReceiver; use nym_sphinx::chunking::fragment::FragmentIdentifier; use std::collections::HashSet; use std::time::Duration; diff --git a/clients/webassembly/src/tester/helpers.rs b/clients/webassembly/src/tester/helpers.rs index 59269b6921..9ee30047b7 100644 --- a/clients/webassembly/src/tester/helpers.rs +++ b/clients/webassembly/src/tester/helpers.rs @@ -4,7 +4,8 @@ // due to expansion of #[wasm_bindgen] macro on NodeTestResult #![allow(clippy::drop_non_drop)] -use nym_node_tester_utils::receiver::{Received, ReceivedReceiver}; +use nym_node_tester_utils::processor::Received; +use nym_node_tester_utils::receiver::ReceivedReceiver; use serde::{Deserialize, Serialize}; use std::fmt::{Display, Formatter}; use std::sync::atomic::{AtomicBool, Ordering}; From 221e1870e51d129ce0f6cdc8ec63e8090bf57556 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 19 Apr 2023 11:33:13 +0100 Subject: [PATCH 5/5] removed redundant trait --- common/node-tester-utils/src/error.rs | 2 +- common/node-tester-utils/src/message.rs | 12 +----------- nym-api/src/network_monitor/monitor/mod.rs | 2 +- 3 files changed, 3 insertions(+), 13 deletions(-) diff --git a/common/node-tester-utils/src/error.rs b/common/node-tester-utils/src/error.rs index 31bd77f98f..96337d0bc6 100644 --- a/common/node-tester-utils/src/error.rs +++ b/common/node-tester-utils/src/error.rs @@ -46,7 +46,7 @@ pub enum NetworkTestingError { #[error("received a packet that could not be reconstructed into a full message with a single fragment")] NonReconstructablePacket, - + #[error("the recipient of the test packet was never specified")] UnknownPacketRecipient, } diff --git a/common/node-tester-utils/src/message.rs b/common/node-tester-utils/src/message.rs index c4b63d753c..41dc517956 100644 --- a/common/node-tester-utils/src/message.rs +++ b/common/node-tester-utils/src/message.rs @@ -7,9 +7,8 @@ use nym_sphinx::message::NymMessage; use nym_topology::{gateway, mix}; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; -use std::hash::{Hash, Hasher}; -#[derive(Serialize, Deserialize, Hash, Clone, Copy)] +#[derive(Serialize, Deserialize, Clone, Copy)] pub struct Empty; #[derive(Serialize, Deserialize, Clone)] @@ -126,12 +125,3 @@ impl TestMessage { .map_err(|source| NetworkTestingError::MalformedTestMessageReceived { source }) } } - -impl Hash for TestMessage { - fn hash(&self, state: &mut H) { - self.tested_node.hash(state); - self.msg_id.hash(state); - self.total_msgs.hash(state); - self.ext.hash(state) - } -} diff --git a/nym-api/src/network_monitor/monitor/mod.rs b/nym-api/src/network_monitor/monitor/mod.rs index d8ffea0643..e3a774b166 100644 --- a/nym-api/src/network_monitor/monitor/mod.rs +++ b/nym-api/src/network_monitor/monitor/mod.rs @@ -274,7 +274,7 @@ impl Monitor { ); let report = summary.create_report(total_sent, total_received); - info!("{}", report); + info!("{report}"); self.submit_new_node_statuses(summary).await; }