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] 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() ) } }