Merge pull request #3328 from nymtech/feature/shared-network-monitor

Feature/shared network monitor

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",

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",

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<ClientState> {
                 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()]);

clients/webassembly/src/tester/ephemeral_receiver.rs

@@ -1,10 +1,10 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<WasmTestMessageExt>>,
 }
 
 impl<'a> EphemeralTestReceiver<'a> {
@@ -38,7 +38,7 @@ impl<'a> EphemeralTestReceiver<'a> {
     pub(crate) fn new(
         sent_packets: u32,
         expected_acks: HashSet<FragmentIdentifier>,
-        receiver_permit: AsyncMutexGuard<'a, ReceivedReceiver>,
+        receiver_permit: AsyncMutexGuard<'a, ReceivedReceiver<WasmTestMessageExt>>,
         timeout: Duration,
     ) -> Self {
         EphemeralTestReceiver {
@@ -53,23 +53,18 @@ impl<'a> EphemeralTestReceiver<'a> {
         }
     }
 
-    fn on_next_received_packet(&mut self, packet: Option<Received>) -> bool {
+    fn on_next_received_packet(&mut self, packet: Option<Received<WasmTestMessageExt>>) -> 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) {

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<AsyncMutex<ReceivedReceiver>>);
+pub(super) struct ReceivedReceiverWrapper(Arc<AsyncMutex<ReceivedReceiver<WasmTestMessageExt>>>);
 
 impl ReceivedReceiverWrapper {
-    pub(super) fn new(inner: ReceivedReceiver) -> Self {
+    pub(super) fn new(inner: ReceivedReceiver<WasmTestMessageExt>) -> 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<WasmTestMessageExt>> {
         self.0.lock().await
     }
 }

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

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,
 }

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;
 

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<T = Empty> {
-    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<T = Empty> {
 }
 
 impl<T> TestMessage<T> {
-    pub fn new_mix(node: &mix::Node, msg_id: u32, total_msgs: u32, ext: T) -> Self {
+    pub fn new<N: Into<TestableNode>>(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<N>(
+        node: N,
+        msg_id: u32,
+        total_msgs: u32,
+        ext: T,
+    ) -> Result<Vec<u8>, NetworkTestingError>
+    where
+        N: Into<TestableNode>,
+        T: Serialize,
+    {
+        Self::new(node, msg_id, total_msgs, ext).as_bytes()
+    }
+
+    pub fn new_plaintexts<N>(
+        node: &N,
+        total_msgs: u32,
+        ext: T,
+    ) -> Result<Vec<Vec<u8>>, NetworkTestingError>
+    where
+        for<'a> &'a N: Into<TestableNode>,
+        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<Vec<Vec<u8>>, 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<Vec<Vec<u8>>, NetworkTestingError>
+    where
+        T: Serialize + Clone,
+    {
+        Self::new_plaintexts(node, total_msgs, ext)
     }
 
     pub fn as_json_string(&self) -> Result<String, NetworkTestingError>
@@ -88,12 +125,3 @@ impl<T> TestMessage<T> {
             .map_err(|source| NetworkTestingError::MalformedTestMessageReceived { source })
     }
 }
-
-impl<T: Hash> Hash for TestMessage<T> {
-    fn hash<H: Hasher>(&self, state: &mut H) {
-        self.encoded_node_identity.hash(state);
-        self.node_owner.hash(state);
-        self.node_type.hash(state);
-        self.ext.hash(state)
-    }
-}

common/node-tester-utils/src/node.rs

@@ -0,0 +1,92 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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"),
+        }
+    }
+}

common/node-tester-utils/src/processor.rs

@@ -0,0 +1,99 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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<T> {
+    Message(TestMessage<T>),
+    Ack(FragmentIdentifier),
+}
+
+impl<T> From<TestMessage<T>> for Received<T> {
+    fn from(value: TestMessage<T>) -> Self {
+        Received::Message(value)
+    }
+}
+
+impl<T> From<FragmentIdentifier> for Received<T> {
+    fn from(value: FragmentIdentifier) -> Self {
+        Received::Ack(value)
+    }
+}
+
+pub struct TestPacketProcessor<T, R: MessageReceiver = SphinxMessageReceiver> {
+    local_encryption_keypair: Arc<encryption::KeyPair>,
+    ack_key: Arc<AckKey>,
+
+    /// Structure responsible for decrypting and recovering plaintext message from received ciphertexts.
+    message_receiver: R,
+
+    _ext_phantom: PhantomData<T>,
+}
+
+impl<T> TestPacketProcessor<T, SphinxMessageReceiver> {
+    pub fn new_sphinx_processor(
+        local_encryption_keypair: Arc<encryption::KeyPair>,
+        ack_key: Arc<AckKey>,
+    ) -> Self {
+        Self::new(local_encryption_keypair, ack_key)
+    }
+}
+
+impl<T, R> TestPacketProcessor<T, R>
+where
+    R: MessageReceiver,
+{
+    pub fn new(local_encryption_keypair: Arc<encryption::KeyPair>, ack_key: Arc<AckKey>) -> 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<u8>,
+    ) -> Result<TestMessage<T>, 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<u8>,
+    ) -> Result<FragmentIdentifier, NetworkTestingError> {
+        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 })
+    }
+}

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<Received>;
-pub type ReceivedReceiver = mpsc::UnboundedReceiver<Received>;
-
-// simple enum containing aggregated processed results
-pub enum Received {
-    Message(NymMessage),
-    Ack(FragmentIdentifier),
-}
-
-impl From<NymMessage> for Received {
-    fn from(value: NymMessage) -> Self {
-        Received::Message(value)
-    }
-}
-
-impl From<FragmentIdentifier> for Received {
-    fn from(value: FragmentIdentifier) -> Self {
-        Received::Ack(value)
-    }
-}
+pub type ReceivedSender<T> = mpsc::UnboundedSender<Received<T>>;
+pub type ReceivedReceiver<T> = mpsc::UnboundedReceiver<Received<T>>;
 
 // the 'Simple' bit comes from the fact that it expects all received messages to consist of a single `Fragment`
-pub struct SimpleMessageReceiver<R: MessageReceiver = SphinxMessageReceiver> {
-    local_encryption_keypair: Arc<encryption::KeyPair>,
-
-    ack_key: Arc<AckKey>,
-
-    /// Structure responsible for decrypting and recovering plaintext message from received ciphertexts.
-    message_receiver: R,
+pub struct SimpleMessageReceiver<T, R: MessageReceiver = SphinxMessageReceiver> {
+    message_processor: TestPacketProcessor<T, R>,
 
     mixnet_message_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
     acks_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
 
-    received_sender: ReceivedSender,
+    received_sender: ReceivedSender<T>,
     shutdown: TaskClient,
 }
 
-impl SimpleMessageReceiver<SphinxMessageReceiver> {
+impl<T> SimpleMessageReceiver<T, SphinxMessageReceiver> {
     pub fn new_sphinx_receiver(
         local_encryption_keypair: Arc<encryption::KeyPair>,
         ack_key: Arc<AckKey>,
         mixnet_message_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
         acks_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
-        received_sender: ReceivedSender,
+        received_sender: ReceivedSender<T>,
+        shutdown: TaskClient,
+    ) -> Self {
+        Self::new(
+            local_encryption_keypair,
+            ack_key,
+            mixnet_message_receiver,
+            acks_receiver,
+            received_sender,
+            shutdown,
+        )
+    }
+}
+
+impl<T, R: MessageReceiver> SimpleMessageReceiver<T, R> {
+    pub fn new(
+        local_encryption_keypair: Arc<encryption::KeyPair>,
+        ack_key: Arc<AckKey>,
+        mixnet_message_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
+        acks_receiver: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
+        received_sender: ReceivedSender<T>,
         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<R: MessageReceiver> SimpleMessageReceiver<R> {
-    fn forward_received<T: Into<Received>>(&self, received: T) {
+    fn forward_received<U: Into<Received<T>>>(&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<u8>) -> 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<u8>) -> 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<u8>) -> 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;

common/node-tester-utils/src/tester.rs

@@ -21,7 +21,10 @@ pub struct NodeTester<R> {
 
     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<Recipient>,
 
     packet_size: PacketSize,
 
@@ -47,7 +50,7 @@ where
     pub fn new(
         rng: R,
         base_topology: NymTopology,
-        recipient: Recipient,
+        self_address: Option<Recipient>,
         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<Vec<PreparedFragment>, NetworkTestingError> {
-        self.mixnode_test_packets(mix, Empty, test_packets)
+        self.mixnode_test_packets(mix, Empty, test_packets, None)
     }
 
     pub fn mixnode_test_packets<T>(
@@ -97,6 +100,7 @@ where
         mix: &mix::Node,
         msg_ext: T,
         test_packets: u32,
+        custom_recipient: Option<Recipient>,
     ) -> Result<Vec<PreparedFragment>, 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<T>(
+        &mut self,
+        nodes: &[mix::Node],
+        msg_ext: T,
+        test_packets: u32,
+        custom_recipient: Option<Recipient>,
+    ) -> Result<Vec<PreparedFragment>, 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<Recipient>,
     ) -> Result<Vec<PreparedFragment>, 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<T>(
@@ -133,6 +164,7 @@ where
         encoded_mix_identity: String,
         msg_ext: T,
         test_packets: u32,
+        custom_recipient: Option<Recipient>,
     ) -> Result<Vec<PreparedFragment>, 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<T>(
+    pub fn gateway_test_packets<T>(
         &mut self,
-        message: &TestMessage<T>,
-        topology: &NymTopology,
-    ) -> Result<PreparedFragment, NetworkTestingError>
+        gateway: &gateway::Node,
+        msg_ext: T,
+        test_packets: u32,
+        custom_recipient: Option<Recipient>,
+    ) -> Result<Vec<PreparedFragment>, 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<T>(
+        &mut self,
+        encoded_gateway_identity: String,
+        msg_ext: T,
+        test_packets: u32,
+        custom_recipient: Option<Recipient>,
+    ) -> Result<Vec<PreparedFragment>, 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<u8>,
+        topology: &NymTopology,
+        custom_recipient: Option<Recipient>,
+    ) -> Result<PreparedFragment, NetworkTestingError> {
+        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<T>(
+        &mut self,
+        message: &TestMessage<T>,
+        topology: &NymTopology,
+        custom_recipient: Option<Recipient>,
+    ) -> Result<PreparedFragment, NetworkTestingError>
+    where
+        T: Serialize,
+    {
+        let serialized = message.as_bytes()?;
+        self.wrap_plaintext_data(serialized, topology, custom_recipient)
+    }
 }
 
 impl<R: CryptoRng + Rng> FragmentPreparer for NodeTester<R> {

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

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" }

common/nymsphinx/routing/src/lib.rs

@@ -0,0 +1,43 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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<Vec<Node>, 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<Node> {
+    type Error = InvalidNumberOfHops;
+
+    fn sphinx_route(
+        &mut self,
+        hops: u8,
+        _destination: &Recipient,
+    ) -> Result<Vec<Node>, 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())
+        }
+    }
+}

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;

common/nymsphinx/src/preparer/mod.rs

@@ -38,6 +38,12 @@ pub struct PreparedFragment {
     pub fragment_identifier: FragmentIdentifier,
 }
 
+impl From<PreparedFragment> 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 {

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]

common/topology/src/error.rs

@@ -0,0 +1,36 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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)>,
+    },
+}

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<MixLayer, Vec<mix::Node>> {
         &self.mixes
     }
@@ -182,8 +160,7 @@ impl NymTopology {
         num_mix_hops: u8,
     ) -> Result<Vec<SphinxNode>, 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;
 

common/topology/src/random_route_provider.rs

@@ -0,0 +1,29 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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<R> {
+    rng: R,
+    inner: NymTopology,
+}
+
+impl<R> SphinxRouteMaker for NymTopologyRouteProvider<R>
+where
+    R: Rng + CryptoRng,
+{
+    type Error = NymTopologyError;
+
+    fn sphinx_route(
+        &mut self,
+        hops: u8,
+        destination: &Recipient,
+    ) -> Result<Vec<Node>, NymTopologyError> {
+        self.inner
+            .random_route_to_gateway(&mut self.rng, hops, destination.gateway())
+    }
+}

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 = []

nym-api/src/main.rs

@@ -58,8 +58,6 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 async fn start_nym_api_tasks(
     config: Config,
 ) -> Result<ShutdownHandles, Box<dyn Error + Send + Sync>> {
-    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;

nym-api/src/network_monitor/chunker.rs

@@ -1,78 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// 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<OsRng>,
-}
-
-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<u8>,
-        topology: &NymTopology,
-        packet_sender: Recipient,
-    ) -> Vec<MixPacket> {
-        // 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<u8>,
-        topology: &NymTopology,
-        packet_sender: Recipient,
-    ) -> Vec<MixPacket> {
-        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
-    }
-}

nym-api/src/network_monitor/gateways_reader.rs

@@ -1,18 +1,22 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<Vec<u8>>;
+pub(crate) enum GatewayMessages {
+    Data(Vec<Vec<u8>>),
+    Acks(Vec<Vec<u8>>),
+}
 
 pub(crate) struct GatewaysReader {
-    ack_map: StreamMap<String, AcknowledgementReceiver>,
-    stream_map: StreamMap<String, MixnetMessageReceiver>,
+    ack_map: StreamMap<IdentityKey, AcknowledgementReceiver>,
+    stream_map: StreamMap<IdentityKey, MixnetMessageReceiver>,
 }
 
 impl GatewaysReader {
@@ -41,20 +45,26 @@ impl GatewaysReader {
 }
 
 impl Stream for GatewaysReader {
-    // just return whatever is returned by our main `stream_map`
-    type Item = <StreamMap<String, MixnetMessageReceiver> as Stream>::Item;
+    type Item = (IdentityKey, GatewayMessages);
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        // 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)))
+            })
     }
 }

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<R: MessageReceiver + Send + 'static> NetworkMonitorRunnables<R> {
 }
 
 fn new_packet_preparer(
-    system_version: &str,
     validator_cache: NymContractCache,
     per_node_test_packets: usize,
+    ack_key: Arc<AckKey>,
     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<R: MessageReceiver + Send + 'static>(
     packets_receiver: ReceivedProcessorReceiver,
     client_encryption_keypair: Arc<encryption::KeyPair>,
+    ack_key: Arc<AckKey>,
 ) -> ReceivedProcessor<R> {
-    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<R: MessageReceiver + Send + 'static>(
     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<R> = monitor_builder.build().await;
     runnables.spawn_tasks(shutdown);

nym-api/src/network_monitor/monitor/mod.rs

@@ -1,11 +1,11 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<R: MessageReceiver + Send> Monitor<R> {
         }
     }
 
-    fn analyse_received_test_route_packets(&self, packets: &[TestPacket]) -> HashMap<u64, usize> {
+    fn analyse_received_test_route_packets(
+        &self,
+        packets: &[NodeTestMessage],
+    ) -> HashMap<u64, usize> {
         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<R: MessageReceiver + Send> Monitor<R> {
 
         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<R: MessageReceiver + Send> Monitor<R> {
         );
 
         let report = summary.create_report(total_sent, total_received);
-        info!("{}", report);
+        info!("{report}");
 
         self.submit_new_node_statuses(summary).await;
     }

nym-api/src/network_monitor/monitor/preparer.rs

@@ -1,113 +1,50 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<MixId> {
-        match self {
-            InvalidNode::Outdated(_, _, node_type, _) => node_type.mix_id(),
-            InvalidNode::Malformed(_, _, node_type) => node_type.mix_id(),
+impl From<InvalidNode> 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<MixId> {
-        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<GatewayPackets>,
 
     /// Vector containing list of public keys and owners of all nodes mixnodes being tested.
-    pub(super) tested_mixnodes: Vec<TestedNode>,
+    pub(super) tested_mixnodes: Vec<TestableNode>,
 
     /// Vector containing list of public keys and owners of all gateways being tested.
-    pub(super) tested_gateways: Vec<TestedNode>,
+    pub(super) tested_gateways: Vec<TestableNode>,
 
     /// 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<Chunker>,
     validator_cache: NymContractCache,
 
     /// Number of test packets sent to each node
     per_node_test_packets: usize,
 
+    ack_key: Arc<AckKey>,
     // 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<AckKey>,
         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<Recipient>,
+    ) -> NodeTester<ThreadRng> {
+        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<ThreadRng> {
+        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<ThreadRng> {
+        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

nym-api/src/network_monitor/monitor/processor.rs

@@ -1,14 +1,17 @@
-// Copyright 2021-2022 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<GatewayMessa
 #[derive(Error, Debug)]
 enum ProcessingError {
     #[error(
-        "could not recover underlying data from the received packet since it was malformed - {0}"
+        "could not recover underlying data from the received packet since it was malformed: {0}"
     )]
     MalformedPacketReceived(#[from] MessageRecoveryError),
 
-    #[error("received a mix packet that was NOT a proper network monitor test packet")]
-    NonTestPacketReceived,
-
-    #[error("the received test packet was malformed - {0}")]
-    MalformedTestPacket(#[from] TestPacketError),
+    #[error("the received test packet was malformed: {0}")]
+    MalformedTestPacket(#[from] NetworkTestingError),
 
     #[error("received packet with an unexpected nonce. Got: {received}, expected: {expected}")]
     NonMatchingNonce { received: u64, expected: u64 },
@@ -50,52 +50,68 @@ struct ReceivedProcessorInner<R: MessageReceiver> {
     /// 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<encryption::KeyPair>,
-
-    /// Structure responsible for decrypting and recovering plaintext message from received ciphertexts.
-    message_receiver: R,
+    test_processor: TestPacketProcessor<NymApiTestMessageExt, R>,
 
     /// Vector containing all received (and decrypted) packets in the current test run.
-    received_packets: Vec<TestPacket>,
+    // TODO: perhaps a different structure would be better here
+    received_packets: Vec<NodeTestMessage>,
 }
 
 impl<R: MessageReceiver> ReceivedProcessorInner<R> {
-    fn on_message(&mut self, mut message: Vec<u8>) -> Result<(), ProcessingError> {
+    fn on_received_data(&mut self, raw_message: Vec<u8>) -> 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<u8>) -> 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<TestPacket> {
+    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<NodeTestMessage> {
         self.test_nonce = None;
         mem::take(&mut self.received_packets)
     }
@@ -110,13 +126,13 @@ impl<R: MessageReceiver + Send + 'static> ReceivedProcessor<R> {
     pub(crate) fn new(
         packets_receiver: ReceivedProcessorReceiver,
         client_encryption_keypair: Arc<encryption::KeyPair>,
+        ack_key: Arc<AckKey>,
     ) -> Self {
         let inner: Arc<Mutex<ReceivedProcessorInner<R>>> =
             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<R: MessageReceiver + Send + 'static> ReceivedProcessor<R> {
                             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<R: MessageReceiver + Send + 'static> ReceivedProcessor<R> {
             .expect("processing task has died!");
     }
 
-    pub(super) async fn return_received(&mut self) -> Vec<TestPacket> {
+    pub(super) async fn return_received(&mut self) -> Vec<NodeTestMessage> {
         // ask for the lock back
         self.permit_changer
             .as_mut()

nym-api/src/network_monitor/monitor/receiver.rs

@@ -1,4 +1,4 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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)
                 }
             }
         }

nym-api/src/network_monitor/monitor/summary_producer.rs

@@ -1,15 +1,14 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<TestedNode>,
-        tested_gateways: Vec<TestedNode>,
-        received_packets: Vec<TestPacket>,
+        tested_mixnodes: Vec<TestableNode>,
+        tested_gateways: Vec<TestableNode>,
+        received_packets: Vec<NodeTestMessage>,
         invalid_mixnodes: Vec<InvalidNode>,
         invalid_gateways: Vec<InvalidNode>,
         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();
 

nym-api/src/network_monitor/test_packet.rs

@@ -1,260 +1,32 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<NymApiTestMessageExt>;
 
-#[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::<MixId>(),
-            NodeType::Gateway => 1,
-        }
-    }
-
-    pub(crate) fn mix_id(&self) -> Option<MixId> {
-        match self {
-            NodeType::Mixnode(mix_id) => Some(*mix_id),
-            NodeType::Gateway => None,
-        }
-    }
-
-    pub(crate) fn into_bytes(self) -> Vec<u8> {
-        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<Self, TestPacketError> {
-        if b.is_empty() {
-            return Err(TestPacketError::InvalidNodeType);
-        }
-        match b[0] {
-            t if t == MIXNODE_TYPE => {
-                if b.len() < (1 + mem::size_of::<MixId>()) {
-                    return Err(TestPacketError::InvalidNodeType);
-                }
-                Ok(NodeType::Mixnode(MixId::from_be_bytes(
-                    b[1..1 + mem::size_of::<MixId>()].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<H: Hasher>(&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<u8> {
-        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<Self, TestPacketError> {
-        // route id + test nonce size
-        let n = mem::size_of::<u64>();
-
-        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<TestPacket> 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<Vec<Vec<u8>>, NetworkTestingError> {
+        NodeTestMessage::mix_plaintexts(node, test_packets, *self)
     }
 }

nym-api/src/network_monitor/test_route/mod.rs

@@ -1,7 +1,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // 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<Vec<u8>> {
         // 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()
         )
     }
 }