diff --git a/Cargo.lock b/Cargo.lock index c86b20097c..818a296b22 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..4f2ce3cd8a 100644 --- a/clients/webassembly/src/tester/ephemeral_receiver.rs +++ b/clients/webassembly/src/tester/ephemeral_receiver.rs @@ -1,10 +1,10 @@ // 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_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; @@ -21,7 +21,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 +38,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 +53,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..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}; @@ -14,10 +15,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 +37,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..96337d0bc6 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..41dc517956 100644 --- a/common/node-tester-utils/src/message.rs +++ b/common/node-tester-utils/src/message.rs @@ -2,27 +2,18 @@ // 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)] +#[derive(Serialize, Deserialize, 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 +25,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 @@ -88,12 +125,3 @@ impl TestMessage { .map_err(|source| NetworkTestingError::MalformedTestMessageReceived { source }) } } - -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.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..9d4e185d59 --- /dev/null +++ b/common/node-tester-utils/src/node.rs @@ -0,0 +1,92 @@ +// 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, Eq, PartialEq)] +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) + } + + pub fn is_mixnode(&self) -> bool { + self.typ.is_mixnode() + } +} + +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, 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 { + 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..0146c25258 --- /dev/null +++ b/common/node-tester-utils/src/processor.rs @@ -0,0 +1,99 @@ +// 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 { + Self::new(local_encryption_keypair, ack_key) + } +} + +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, + ) -> 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..bf659bf0a5 100644 --- a/common/node-tester-utils/src/receiver.rs +++ b/common/node-tester-utils/src/receiver.rs @@ -2,114 +2,95 @@ // 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 { + 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 { - local_encryption_keypair, - ack_key, - message_receiver: SphinxMessageReceiver::new(), + message_processor: TestPacketProcessor::new(local_encryption_keypair, ack_key), mixnet_message_receiver, acks_receiver, received_sender, shutdown, } } -} -impl SimpleMessageReceiver { - fn forward_received>(&self, received: T) { + 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 8ae9f77592..6c57738067 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/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/chunker.rs b/nym-api/src/network_monitor/chunker.rs deleted file mode 100644 index efce5e92d7..0000000000 --- a/nym-api/src/network_monitor/chunker.rs +++ /dev/null @@ -1,78 +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)] -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 ab521dd14e..37a6fba194 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,20 +45,26 @@ 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 match Pin::new(&mut self.ack_map).poll_next(cx) { Poll::Ready(None) => { // 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_by_ref(); + 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/mod.rs b/nym-api/src/network_monitor/mod.rs index f19cccc35e..03e4ea65e8 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; @@ -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, } @@ -83,15 +78,16 @@ 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) = 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), *identity_keypair.public_key(), *encryption_keypair.public_key(), ); @@ -118,6 +114,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( @@ -161,16 +158,16 @@ impl NetworkMonitorRunnables { } 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 { PacketPreparer::new( - system_version, validator_cache, per_node_test_packets, + ack_key, self_public_identity, self_public_encryption, ) @@ -199,8 +196,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 { @@ -223,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 ebc7174d2a..e3a774b166 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 @@ -258,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( @@ -271,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; } diff --git a/nym-api/src/network_monitor/monitor/preparer.rs b/nym-api/src/network_monitor/monitor/preparer.rs index a0df027e9a..7343a06a6e 100644 --- a/nym-api/src/network_monitor/monitor/preparer.rs +++ b/nym-api/src/network_monitor/monitor/preparer.rs @@ -1,113 +1,50 @@ -// 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::{NodeType, TestPacket}; 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::{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_topology::{gateway, mix, NymTopology}; -use rand::seq::SliceRandom; -use rand::{thread_rng, Rng}; +use nym_sphinx::params::PacketSize; +use nym_topology::{gateway, mix}; +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), + 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::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(), +impl From for TestableNode { + fn from(value: InvalidNode) -> Self { + match value { + InvalidNode::Malformed { node } => node, } } - - 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) - } } pub(crate) struct PreparedPackets { @@ -116,10 +53,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. @@ -132,13 +69,12 @@ 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 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 @@ -149,45 +85,56 @@ pub(crate) struct PacketPreparer { impl PacketPreparer { pub(crate) fn new( - 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 { PacketPreparer { - system_version: system_version.to_owned(), - chunker: None, validator_cache, per_node_test_packets, + ack_key, 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 rng = thread_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, + self.ack_key.clone(), + ) + } - mix_packets.pop().unwrap() + // 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)) + } + + #[allow(dead_code)] + 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 +149,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 +243,39 @@ 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(), node_1, node_2, node_3, gateway)) + } + info!( + "The following routes will be used for testing: {:#?}", + routes + ); + Some(routes) } fn create_packet_sender(&self, gateway: &gateway::Node) -> Recipient { @@ -369,13 +291,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(), @@ -394,11 +322,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 +344,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 +380,53 @@ 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 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 diff --git a/nym-api/src/network_monitor/monitor/processor.rs b/nym-api/src/network_monitor/monitor/processor.rs index 4ea4c2f880..0e38eebd0b 100644 --- a/nym-api/src/network_monitor/monitor/processor.rs +++ b/nym-api/src/network_monitor/monitor/processor.rs @@ -1,14 +1,17 @@ -// 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::{NodeTestMessage, 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::acknowledgements::AckKey; use nym_sphinx::receiver::{MessageReceiver, MessageRecoveryError}; use std::mem; use std::sync::Arc; @@ -20,15 +23,12 @@ 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, + test_processor: TestPacketProcessor, /// Vector containing all received (and decrypted) packets in the current test run. - received_packets: Vec, + // TODO: perhaps a different structure would be better here + received_packets: Vec, } impl ReceivedProcessorInner { - fn on_message(&mut self, mut message: Vec) -> Result<(), ProcessingError> { + 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 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())?; + let test_msg = self.test_processor.process_mixnet_message(raw_message)?; - // we know nonce is NOT none - if test_packet.test_nonce() != self.test_nonce.unwrap() { + if test_msg.ext.test_nonce != self.test_nonce.unwrap() { return Err(ProcessingError::NonMatchingNonce { - received: test_packet.test_nonce(), + received: test_msg.ext.test_nonce, expected: self.test_nonce.unwrap(), }); } - self.received_packets.push(test_packet); + 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 going to do anything about it (just yet)"); Ok(()) } - fn finish_run(&mut self) -> Vec { + 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 finish_run(&mut self) -> Vec { self.test_nonce = None; mem::take(&mut self.received_packets) } @@ -110,13 +126,13 @@ impl ReceivedProcessor { pub(crate) fn new( packets_receiver: ReceivedProcessorReceiver, client_encryption_keypair: Arc, + ack_key: Arc, ) -> Self { let inner: Arc>> = Arc::new(Mutex::new(ReceivedProcessorInner { test_nonce: None, packets_receiver, - client_encryption_keypair, - message_receiver: R::new(), + test_processor: TestPacketProcessor::new(client_encryption_keypair, ack_key), received_packets: Vec::new(), })); @@ -150,13 +166,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, }, } @@ -210,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/receiver.rs b/nym-api/src/network_monitor/monitor/receiver.rs index ad9e90162f..344f8193b7 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,8 @@ 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 { - log::error!("the gateways reader stream has terminated!"); - continue - }; - self.process_gateway_messages(message) + 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 704084c338..2a68004e7b 100644 --- a/nym-api/src/network_monitor/monitor/summary_producer.rs +++ b/nym-api/src/network_monitor/monitor/summary_producer.rs @@ -1,15 +1,14 @@ -// 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_packet::NodeTestMessage; use crate::network_monitor::test_route::TestRoute; use nym_mixnet_contract_common::MixId; +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,12 @@ impl GatewayResult { #[derive(Debug, Clone)] pub(crate) struct RouteResult { pub(crate) route: TestRoute, - reliability: u8, + performance: f32, } impl RouteResult { - pub(crate) fn new(route: TestRoute, reliability: u8) -> Self { - RouteResult { route, reliability } + pub(crate) fn new(route: TestRoute, performance: f32) -> Self { + RouteResult { route, performance } } } @@ -179,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(()) @@ -281,108 +280,77 @@ impl SummaryProducer { pub(super) fn produce_summary( &self, - tested_mixnodes: Vec, - tested_gateways: Vec, - received_packets: Vec, + tested_mixnodes: Vec, + tested_gateways: 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; // 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(), performance)) }) .collect(); diff --git a/nym-api/src/network_monitor/test_packet.rs b/nym-api/src/network_monitor/test_packet.rs index 0a38e642b7..289172d89b 100644 --- a/nym-api/src/network_monitor/test_packet.rs +++ b/nym-api/src/network_monitor/test_packet.rs @@ -1,260 +1,32 @@ -// 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_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 nym_node_tester_utils::error::NetworkTestingError; +use nym_node_tester_utils::TestMessage; +use nym_topology::mix; +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(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 mix_plaintexts( + &self, + node: &mix::Node, + test_packets: u32, + ) -> Result>, NetworkTestingError> { + NodeTestMessage::mix_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..48eede18b5 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}; @@ -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]), } } @@ -70,30 +67,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() } } @@ -101,19 +89,12 @@ 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.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() ) } }