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

wip

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2023-04-18 16:18:48 +01:00
parent eda223ed3d
commit f64cfb4cd1
31 changed files with 958 additions and 720 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+12
View File
@@ -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
+1
View File
@@ -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
+4 -15
View File
@@ -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
+8 -14
View File
@@ -1,8 +1,7 @@
// 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_sphinx::chunking::fragment::FragmentIdentifier;
@@ -21,7 +20,7 @@ pub(crate) struct EphemeralTestReceiver<'a> {
duplicate_acks: u32,
timeout_duration: Duration,
receiver_permit: AsyncMutexGuard<'a, ReceivedReceiver>,
receiver_permit: AsyncMutexGuard<'a, ReceivedReceiver<WasmTestMessageExt>>,
}
impl<'a> EphemeralTestReceiver<'a> {
@@ -38,7 +37,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 +52,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
+3 -3
View File
@@ -14,10 +14,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 +36,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
+7 -2
View File
@@ -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
+3
View File
@@ -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
+2
View File
@@ -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
+62 -24
View File
@@ -2,27 +2,19 @@
// SPDX-License-Identifier: Apache-2.0
use crate::error::NetworkTestingError;
use crate::MixId;
use crate::node::TestableNode;
use nym_sphinx::message::NymMessage;
use nym_topology::{gateway, mix};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::hash::{Hash, Hasher};
#[derive(Serialize, Deserialize, Hash, Clone, Copy)]
pub enum NodeType {
Mixnode(MixId),
Gateway,
}
#[derive(Serialize, Deserialize, Hash, Clone, Copy)]
pub struct Empty;
#[derive(Serialize, Deserialize, Clone)]
pub struct TestMessage<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 +26,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>
@@ -91,9 +129,9 @@ impl<T> TestMessage<T> {
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.tested_node.hash(state);
self.msg_id.hash(state);
self.total_msgs.hash(state);
self.ext.hash(state)
}
}
common/node-tester-utils/src/node.rs
+82
View File
@@ -0,0 +1,82 @@
// 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)]
pub struct TestableNode {
pub encoded_identity: String,
pub owner: String,
#[serde(rename = "type")]
pub typ: NodeType,
}
impl TestableNode {
pub fn new(encoded_identity: String, owner: String, typ: NodeType) -> Self {
TestableNode {
encoded_identity,
owner,
typ,
}
}
pub fn new_mixnode(encoded_identity: String, owner: String, mix_id: MixId) -> Self {
TestableNode::new(encoded_identity, owner, NodeType::Mixnode { mix_id })
}
pub fn new_gateway(encoded_identity: String, owner: String) -> Self {
TestableNode::new(encoded_identity, owner, NodeType::Gateway)
}
}
impl<'a> From<&'a mix::Node> for TestableNode {
fn from(value: &'a mix::Node) -> Self {
TestableNode {
encoded_identity: value.identity_key.to_base58_string(),
owner: value.owner.clone(),
typ: NodeType::Mixnode {
mix_id: value.mix_id,
},
}
}
}
impl<'a> From<&'a gateway::Node> for TestableNode {
fn from(value: &'a gateway::Node) -> Self {
TestableNode {
encoded_identity: value.identity_key.to_base58_string(),
owner: value.owner.clone(),
typ: NodeType::Gateway,
}
}
}
impl Display for TestableNode {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{} {} owned by {}",
self.typ, self.encoded_identity, self.owner
)
}
}
#[derive(Serialize, Deserialize, Hash, Clone, Copy, Debug)]
#[serde(rename_all = "snake_case")]
pub enum NodeType {
Mixnode { mix_id: MixId },
Gateway,
}
impl Display for NodeType {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
NodeType::Mixnode { mix_id } => write!(f, "mixnode (mix_id {mix_id})"),
NodeType::Gateway => write!(f, "gateway"),
}
}
}
common/node-tester-utils/src/processor.rs
+95
View File
@@ -0,0 +1,95 @@
// 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 {
TestPacketProcessor {
local_encryption_keypair,
ack_key,
message_receiver: SphinxMessageReceiver::new(),
_ext_phantom: PhantomData,
}
}
}
impl<T, R> TestPacketProcessor<T, R>
where
R: MessageReceiver,
{
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
+26 -60
View File
@@ -2,69 +2,45 @@
// SPDX-License-Identifier: Apache-2.0
use crate::error::NetworkTestingError;
use crate::processor::{Received, TestPacketProcessor};
use crate::{log_err, log_info, log_warn};
use futures::channel::mpsc;
use futures::StreamExt;
use nym_crypto::asymmetric::encryption;
use nym_sphinx::message::NymMessage;
use nym_sphinx::acknowledgements::AckKey;
use nym_sphinx::receiver::{MessageReceiver, SphinxMessageReceiver};
use nym_sphinx::{
acknowledgements::{identifier::recover_identifier, AckKey},
chunking::fragment::FragmentIdentifier,
};
use nym_task::TaskClient;
use serde::de::DeserializeOwned;
use std::sync::Arc;
pub type ReceivedSender = mpsc::UnboundedSender<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 {
SimpleMessageReceiver {
local_encryption_keypair,
ack_key,
message_receiver: SphinxMessageReceiver::new(),
message_processor: TestPacketProcessor::new_sphinx_processor(
local_encryption_keypair,
ack_key,
),
mixnet_message_receiver,
acks_receiver,
received_sender,
@@ -73,43 +49,33 @@ impl SimpleMessageReceiver<SphinxMessageReceiver> {
}
}
impl<R: MessageReceiver> SimpleMessageReceiver<R> {
fn forward_received<T: Into<Received>>(&self, received: T) {
impl<T, R: MessageReceiver> SimpleMessageReceiver<T, R> {
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
+104 -18
View File
@@ -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
+1
View File
@@ -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
+14
View File
@@ -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
+43
View File
@@ -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
+4 -2
View File
@@ -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
+6
View File
@@ -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
+1
View File
@@ -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
+36
View File
@@ -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
+10 -33
View File
@@ -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
+29
View File
@@ -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
+1
View File
@@ -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/network_monitor/chunker.rs
+1
View File
@@ -15,6 +15,7 @@ const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200);
const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200);
#[derive(Clone)]
#[deprecated]
pub(crate) struct Chunker {
rng: OsRng,
packet_size: PacketSize,
nym-api/src/network_monitor/gateways_reader.rs
+19 -8
View File
@@ -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,8 +45,7 @@ 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
@@ -51,10 +54,18 @@ impl Stream for GatewaysReader {
// this should have never happened!
return Poll::Ready(None);
}
Poll::Ready(Some(_item)) => (),
Poll::Ready(Some(ack_item)) => {
// wake immediately in case there's an associated data message
cx.waker().wake();
return Poll::Ready(Some((ack_item.0, GatewayMessages::Acks(ack_item.1))));
}
Poll::Pending => (),
}
Pin::new(&mut self.stream_map).poll_next(cx)
Pin::new(&mut self.stream_map)
.poll_next(cx)
.map(|maybe_data_item| {
maybe_data_item.map(|data_item| (data_item.0, GatewayMessages::Data(data_item.1)))
})
}
}
nym-api/src/network_monitor/monitor/preparer.rs
+266 -225
View File
@@ -3,112 +3,70 @@
use crate::network_monitor::chunker::Chunker;
use crate::network_monitor::monitor::sender::GatewayPackets;
use crate::network_monitor::test_packet::{NodeType, TestPacket};
use crate::network_monitor::test_packet::{NodeTestMessage, NymApiTestMessageExt};
use crate::network_monitor::test_route::TestRoute;
use crate::nym_contract_cache::cache::NymContractCache;
use log::info;
use nym_crypto::asymmetric::{encryption, identity};
use nym_mixnet_contract_common::{Addr, GatewayBond, Layer, MixId, MixNodeBond};
use nym_mixnet_contract_common::{Addr, GatewayBond, IdentityKey, Layer, MixId, MixNodeBond};
use nym_node_tester_utils::node::{NodeType, TestableNode};
use nym_node_tester_utils::NodeTester;
use nym_sphinx::acknowledgements::AckKey;
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::params::PacketSize;
use nym_topology::{gateway, mix, NymTopology};
use rand::seq::SliceRandom;
use rand::{thread_rng, Rng};
use rand_07::{rngs::ThreadRng, seq::SliceRandom, thread_rng, Rng};
use std::collections::{HashMap, HashSet};
use std::convert::TryInto;
use std::fmt::{self, Display, Formatter};
use std::sync::Arc;
use std::time::Duration;
// declared type aliases for easier code reasoning
type Version = String;
type Id = String;
type Owner = Addr;
const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200);
const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200);
#[derive(Clone)]
#[allow(dead_code)]
pub(crate) enum InvalidNode {
Outdated(Id, Owner, NodeType, Version),
Malformed(Id, Owner, NodeType),
Outdated { node: TestableNode, version: String },
Malformed { node: TestableNode },
}
impl Display for InvalidNode {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match self {
InvalidNode::Outdated(id, owner, _, version) => {
write!(
f,
"Node {} (v{}) owned by {} is outdated",
id, version, owner
)
InvalidNode::Outdated { node, version } => {
write!(f, "{node} is outdated. It runs on v{version}")
}
InvalidNode::Malformed(id, owner, _) => {
write!(f, "Node {} owned by {} is malformed", id, owner)
InvalidNode::Malformed { node } => {
write!(f, "{node} is malformed")
}
}
}
}
impl InvalidNode {
pub(crate) fn mix_id(&self) -> Option<MixId> {
match self {
InvalidNode::Outdated(_, _, node_type, _) => node_type.mix_id(),
InvalidNode::Malformed(_, _, node_type) => node_type.mix_id(),
}
}
pub(crate) fn identity(&self) -> String {
match self {
InvalidNode::Outdated(id, _, _, _) => id.clone(),
InvalidNode::Malformed(id, _, _) => id.clone(),
}
}
pub(crate) fn owner(&self) -> String {
match self {
InvalidNode::Outdated(_, owner, _, _) => owner.into(),
InvalidNode::Malformed(_, owner, _) => owner.into(),
}
}
}
#[derive(Eq, PartialEq, Debug, Hash, Clone)]
pub(crate) struct TestedNode {
pub(crate) identity: String,
pub(crate) owner: String,
pub(crate) node_type: NodeType,
}
impl TestedNode {
pub(crate) fn mix_id(&self) -> Option<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)
}
}
// 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(),
// }
// }
//
// pub(crate) fn identity(&self) -> String {
// match self {
// InvalidNode::Outdated(id, _, _, _) => id.clone(),
// InvalidNode::Malformed(id, _, _) => id.clone(),
// }
// }
//
// pub(crate) fn owner(&self) -> String {
// match self {
// InvalidNode::Outdated(_, owner, _, _) => owner.into(),
// InvalidNode::Malformed(_, owner, _) => owner.into(),
// }
// }
// }
pub(crate) struct PreparedPackets {
/// All packets that are going to get sent during the test as well as the gateways through
@@ -116,10 +74,10 @@ pub(crate) struct PreparedPackets {
pub(super) packets: Vec<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.
@@ -133,7 +91,6 @@ 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
@@ -155,39 +112,55 @@ impl PacketPreparer {
self_public_identity: identity::PublicKey,
self_public_encryption: encryption::PublicKey,
) -> Self {
PacketPreparer {
system_version: system_version.to_owned(),
chunker: None,
validator_cache,
per_node_test_packets,
self_public_identity,
self_public_encryption,
}
todo!()
// PacketPreparer {
// system_version: system_version.to_owned(),
// chunker: None,
// validator_cache,
// per_node_test_packets,
// self_public_identity,
// self_public_encryption,
// }
}
fn wrap_test_packet(
&mut self,
packet: &TestPacket,
topology: &NymTopology,
packet_recipient: Recipient,
) -> MixPacket {
// this should be done only once. We can't really do it at construction time
// as there's no sane Default for Recipient
if self.chunker.is_none() {
self.chunker = Some(Chunker::new(packet_recipient));
}
let mut mix_packets = self.chunker.as_mut().unwrap().prepare_packets_from(
packet.to_bytes(),
topology,
packet_recipient,
);
assert_eq!(
mix_packets.len(),
1,
"Our test packets data is longer than a single sphinx packet!"
);
fn ephemeral_tester(
&self,
test_route: &TestRoute,
self_address: Option<Recipient>,
) -> NodeTester<ThreadRng> {
let mut rng = thread_rng();
mix_packets.pop().unwrap()
// TODO: we should make this key more long-term
let ack_key = Arc::new(AckKey::new(&mut rng));
NodeTester::new(
rng,
// the topology here contains 3 mixnodes and 1 gateway so its cheap to clone it
test_route.topology().clone(),
self_address,
PacketSize::RegularPacket,
DEFAULT_AVERAGE_PACKET_DELAY,
DEFAULT_AVERAGE_ACK_DELAY,
ack_key,
)
}
// when we're testing mixnodes, the recipient is going to stay constant, so we can specify it ahead of time
fn ephemeral_mix_tester(&self, test_route: &TestRoute) -> NodeTester<ThreadRng> {
let self_address = self.create_packet_sender(test_route.gateway());
self.ephemeral_tester(test_route, Some(self_address))
}
fn ephemeral_gateway_tester(&self, test_route: &TestRoute) -> NodeTester<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 +175,30 @@ impl PacketPreparer {
let mixnodes = self.validator_cache.mixnodes_basic().await;
if gateways.len() < minimum_full_routes {
info!(
"Minimal topology is still not online. Going to check again in {:?}",
initialisation_backoff
);
tokio::time::sleep(initialisation_backoff).await;
self.topology_wait_backoff(initialisation_backoff).await;
continue;
}
let mut layered_mixes = HashMap::new();
let mut layer1_count = 0;
let mut layer2_count = 0;
let mut layer3_count = 0;
for mix in mixnodes {
let layer = mix.layer;
let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new);
mixes.push(mix)
match mix.layer {
Layer::One => layer1_count += 1,
Layer::Two => layer2_count += 1,
Layer::Three => layer3_count += 1,
}
}
// we remove the entries as this gives us the ownership and thus we can unwrap to default value
// which makes the code slightly nicer without having to deal with options
let layer1 = layered_mixes.remove(&Layer::One).unwrap_or_default();
let layer2 = layered_mixes.remove(&Layer::Two).unwrap_or_default();
let layer3 = layered_mixes.remove(&Layer::Three).unwrap_or_default();
if layer1.len() >= minimum_full_routes
&& layer2.len() >= minimum_full_routes
&& layer3.len() >= minimum_full_routes
if layer1_count >= minimum_full_routes
&& layer2_count >= minimum_full_routes
&& layer3_count >= minimum_full_routes
{
break;
}
info!(
"Minimal topology is still not online. Going to check again in {:?}",
initialisation_backoff
);
tokio::time::sleep(initialisation_backoff).await;
self.topology_wait_backoff(initialisation_backoff).await;
}
}
@@ -305,55 +269,43 @@ impl PacketPreparer {
if most_available == 0 {
error!("Cannot construct test routes. No nodes or gateways available");
None
} else {
trace!("Generating test routes...");
let mut routes = Vec::new();
for i in 0..most_available {
let node_1 = match self.try_parse_mix_bond(rand_l1[i]) {
Ok(node) => node,
Err(id) => {
blacklist.insert(id);
continue;
}
};
let node_2 = match self.try_parse_mix_bond(rand_l2[i]) {
Ok(node) => node,
Err(id) => {
blacklist.insert(id);
continue;
}
};
let node_3 = match self.try_parse_mix_bond(rand_l3[i]) {
Ok(node) => node,
Err(id) => {
blacklist.insert(id);
continue;
}
};
let gateway = match self.try_parse_gateway_bond(rand_gateways[i]) {
Ok(node) => node,
Err(id) => {
blacklist.insert(id);
continue;
}
};
routes.push(TestRoute::new(
rng.gen(),
&self.system_version,
node_1,
node_2,
node_3,
gateway,
))
}
info!("{:?}", routes);
Some(routes)
return None;
}
trace!("Generating test routes...");
let mut routes = Vec::new();
for i in 0..most_available {
let Ok(node_1) = self.try_parse_mix_bond(rand_l1[i]) else {
blacklist.insert(rand_l1[i].identity().to_owned());
continue
};
let Ok(node_2) = self.try_parse_mix_bond(rand_l2[i]) else {
blacklist.insert(rand_l2[i].identity().to_owned());
continue
};
let Ok(node_3) = self.try_parse_mix_bond(rand_l3[i]) else {
blacklist.insert(rand_l3[i].identity().to_owned());
continue
};
let Ok(gateway) = self.try_parse_gateway_bond(rand_gateways[i]) else {
blacklist.insert(rand_gateways[i].identity().to_owned());
continue
};
routes.push(TestRoute::new(
rng.gen(),
&self.system_version,
node_1,
node_2,
node_3,
gateway,
))
}
info!("{:?}", routes);
Some(routes)
}
fn create_packet_sender(&self, gateway: &gateway::Node) -> Recipient {
@@ -369,13 +321,19 @@ impl PacketPreparer {
route: &TestRoute,
num: usize,
) -> GatewayPackets {
let mut mix_packets = Vec::with_capacity(num);
let test_packet = route.self_test_packet();
let recipient = self.create_packet_sender(route.gateway());
for _ in 0..num {
let mix_packet = self.wrap_test_packet(&test_packet, route.topology(), recipient);
mix_packets.push(mix_packet)
}
let mut tester = self.ephemeral_mix_tester(route);
let topology = route.topology();
let plaintexts = route.self_test_messages(num);
// the unwrap here is fine as:
// 1. the topology is definitely valid (otherwise we wouldn't be here)
// 2. the recipient is specified (by calling **mix**_tester)
// 3. the test message is not too long, i.e. when serialized it will fit in a single sphinx packet
let mix_packets = plaintexts
.into_iter()
.map(|p| tester.wrap_plaintext_data(p, topology, None).unwrap())
.map(MixPacket::from)
.collect();
GatewayPackets::new(
route.gateway_clients_address(),
@@ -384,6 +342,26 @@ impl PacketPreparer {
)
}
// pub(crate) fn prepare_test_route_viability_packets(
// &mut self,
// route: &TestRoute,
// num: usize,
// ) -> GatewayPackets {
// let mut mix_packets = Vec::with_capacity(num);
// let test_packet = route.self_test_packet();
// let recipient = self.create_packet_sender(route.gateway());
// for _ in 0..num {
// let mix_packet = self.wrap_test_packet(&test_packet, route.topology(), recipient);
// mix_packets.push(mix_packet)
// }
//
// GatewayPackets::new(
// route.gateway_clients_address(),
// route.gateway_identity(),
// mix_packets,
// )
// }
fn filter_outdated_and_malformed_mixnodes(
&self,
nodes: Vec<MixNodeBond>,
@@ -394,11 +372,13 @@ impl PacketPreparer {
if let Ok(parsed_node) = (&mixnode).try_into() {
parsed_nodes.push(parsed_node)
} else {
invalid_nodes.push(InvalidNode::Malformed(
mixnode.mix_node.identity_key,
mixnode.owner,
NodeType::Mixnode(mixnode.mix_id),
));
invalid_nodes.push(InvalidNode::Malformed {
node: TestableNode::new_mixnode(
mixnode.identity().to_owned(),
mixnode.owner.clone().into_string(),
mixnode.mix_id,
),
});
}
}
(parsed_nodes, invalid_nodes)
@@ -414,11 +394,12 @@ impl PacketPreparer {
if let Ok(parsed_node) = (&gateway).try_into() {
parsed_nodes.push(parsed_node)
} else {
invalid_nodes.push(InvalidNode::Malformed(
gateway.gateway.identity_key,
gateway.owner,
NodeType::Gateway,
));
invalid_nodes.push(InvalidNode::Malformed {
node: TestableNode::new_gateway(
gateway.identity().to_owned(),
gateway.owner.clone().into_string(),
),
});
}
}
(parsed_nodes, invalid_nodes)
@@ -449,42 +430,54 @@ impl PacketPreparer {
// for each test route...
for test_route in test_routes {
let recipient = self.create_packet_sender(test_route.gateway());
let gateway_identity = test_route.gateway_identity();
let route_ext = test_route.test_message_ext(test_nonce);
let gateway_address = test_route.gateway_clients_address();
let gateway_identity = test_route.gateway_identity();
// it's actually going to be a tiny bit more due to gateway testing, but it's a good enough approximation
let mut mix_packets = Vec::with_capacity(mixnodes.len() * self.per_node_test_packets);
let recipient = self.create_packet_sender(test_route.gateway());
let mut mix_tester = self.ephemeral_mix_tester(test_route);
// and for each mixnode...
for mixnode in &mixnodes {
let test_packet = TestPacket::from_mixnode(mixnode, test_route.id(), test_nonce);
let topology = test_route.substitute_mix(mixnode);
// produce n mix packets
for _ in 0..self.per_node_test_packets {
let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient);
mix_packets.push(mix_packet);
}
}
// generate test packets for mixnodes
//
// the unwrap here is fine as:
// 1. the topology is definitely valid (otherwise we wouldn't be here)
// 2. the recipient is specified (by calling **mix**_tester)
// 3. the test message is not too long, i.e. when serialized it will fit in a single sphinx packet
let mixnode_test_packets = mix_tester
.mixnodes_test_packets(
&mixnodes,
route_ext,
self.per_node_test_packets as u32,
None,
)
.unwrap();
let mix_packets = mixnode_test_packets.into_iter().map(Into::into).collect();
let gateway_packets = all_gateway_packets
.entry(gateway_identity.to_bytes())
.or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity));
gateway_packets.push_packets(mix_packets);
// and for each gateway...
// and generate test packets for gateways (note the variable recipient)
for gateway in &gateways {
let mut gateway_mix_packets = Vec::new();
let test_packet = TestPacket::from_gateway(gateway, test_route.id(), test_nonce);
let recipient = self.create_packet_sender(gateway);
let gateway_identity = gateway.identity_key;
let gateway_address = gateway.clients_address();
let recipient = self.create_packet_sender(gateway);
let topology = test_route.substitute_gateway(gateway);
// produce n mix packets
for _ in 0..self.per_node_test_packets {
let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient);
gateway_mix_packets.push(mix_packet);
}
// the unwrap here is fine as:
// 1. the topology is definitely valid (otherwise we wouldn't be here)
// 2. the recipient is specified
// 3. the test message is not too long, i.e. when serialized it will fit in a single sphinx packet
let gateway_test_packets = mix_tester
.gateway_test_packets(
gateway,
route_ext,
self.per_node_test_packets as u32,
Some(recipient),
)
.unwrap();
let gateway_mix_packets =
gateway_test_packets.into_iter().map(Into::into).collect();
// and push it into existing struct (if it's a "core" gateway being tested against another route)
// or create a new one
@@ -493,6 +486,54 @@ impl PacketPreparer {
.or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity));
gateway_packets.push_packets(gateway_mix_packets);
}
//
//
//
// // let gateway_identity = test_route.gateway_identity();
// let gateway_address = test_route.gateway_clients_address();
//
// // it's actually going to be a tiny bit more due to gateway testing, but it's a good enough approximation
// let mut mix_packets = Vec::with_capacity(mixnodes.len() * self.per_node_test_packets);
//
// // and for each mixnode...
// for mixnode in &mixnodes {
// let test_packet = TestPacket::from_mixnode(mixnode, test_route.id(), test_nonce);
// let topology = test_route.substitute_mix(mixnode);
// // produce n mix packets
// for _ in 0..self.per_node_test_packets {
// let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient);
// mix_packets.push(mix_packet);
// }
// }
//
// let gateway_packets = all_gateway_packets
// .entry(gateway_identity.to_bytes())
// .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity));
// gateway_packets.push_packets(mix_packets);
//
// // and for each gateway...
// for gateway in &gateways {
// let mut gateway_mix_packets = Vec::new();
// let test_packet = TestPacket::from_gateway(gateway, test_route.id(), test_nonce);
// let gateway_identity = gateway.identity_key;
// let gateway_address = gateway.clients_address();
// let recipient = self.create_packet_sender(gateway);
// let topology = test_route.substitute_gateway(gateway);
// // produce n mix packets
// for _ in 0..self.per_node_test_packets {
// let mix_packet = self.wrap_test_packet(&test_packet, &topology, recipient);
// gateway_mix_packets.push(mix_packet);
// }
//
// // and push it into existing struct (if it's a "core" gateway being tested against another route)
// // or create a new one
// let gateway_packets = all_gateway_packets
// .entry(gateway_identity.to_bytes())
// .or_insert_with(|| GatewayPackets::empty(gateway_address, gateway_identity));
// gateway_packets.push_packets(gateway_mix_packets);
// }
}
// convert our hashmap back into a vec
nym-api/src/network_monitor/monitor/processor.rs
+71 -34
View File
@@ -1,14 +1,16 @@
// 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::NymApiTestMessageExt;
use crate::network_monitor::ROUTE_TESTING_TEST_NONCE;
use futures::channel::mpsc;
use futures::lock::{Mutex, MutexGuard};
use futures::{SinkExt, StreamExt};
use log::warn;
use nym_crypto::asymmetric::encryption;
use nym_node_tester_utils::error::NetworkTestingError;
use nym_node_tester_utils::processor::TestPacketProcessor;
use nym_sphinx::receiver::{MessageReceiver, MessageRecoveryError};
use std::mem;
use std::sync::Arc;
@@ -20,15 +22,15 @@ pub(crate) type ReceivedProcessorReceiver = mpsc::UnboundedReceiver<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,18 +52,57 @@ 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,
/// Vector containing all received (and decrypted) packets in the current test run.
received_packets: Vec<TestPacket>,
test_processor: TestPacketProcessor<NymApiTestMessageExt, R>,
// TODO: rethinking
// /// Vector containing all received (and decrypted) packets in the current test run.
// received_packets: Vec<TestPacket>,
}
impl<R: MessageReceiver> ReceivedProcessorInner<R> {
fn recover_test_message(&mut self, raw_message: Vec<u8>) {
//
}
fn on_received_data(&mut self, raw_message: Vec<u8>) -> Result<(), ProcessingError> {
let test_msg = self.test_processor.process_mixnet_message(raw_message)?;
if test_msg.ext.test_nonce != self.test_nonce.unwrap() {
return Err(ProcessingError::NonMatchingNonce {
received: test_msg.ext.test_nonce,
expected: self.test_nonce.unwrap(),
});
}
todo!()
}
fn on_received_ack(&mut self, raw_ack: Vec<u8>) -> Result<(), ProcessingError> {
let frag_id = self.test_processor.process_ack(raw_ack)?;
// TODO: hook it up at some point
trace!("received a test ack with id {frag_id}. However, we're not goingn to do anything about it (just yet)");
Ok(())
}
fn on_received(&mut self, messages: GatewayMessages) {
match messages {
GatewayMessages::Data(data_msgs) => {
for raw in data_msgs {
if let Err(err) = self.on_received_data(raw) {
warn!(target: "Monitor", "failed to process received gateway message: {err}")
}
}
}
GatewayMessages::Acks(acks) => {
for raw in acks {
if let Err(err) = self.on_received_ack(raw) {
warn!(target: "Monitor", "failed to process received gateway ack: {err}")
}
}
}
}
}
fn on_message(&mut self, mut message: Vec<u8>) -> Result<(), ProcessingError> {
// if the nonce is none it means the packet was received during the 'waiting' for the
// next test run
@@ -111,19 +152,21 @@ impl<R: MessageReceiver + Send + 'static> ReceivedProcessor<R> {
packets_receiver: ReceivedProcessorReceiver,
client_encryption_keypair: Arc<encryption::KeyPair>,
) -> Self {
let inner: Arc<Mutex<ReceivedProcessorInner<R>>> =
Arc::new(Mutex::new(ReceivedProcessorInner {
test_nonce: None,
packets_receiver,
client_encryption_keypair,
message_receiver: R::new(),
received_packets: Vec::new(),
}));
ReceivedProcessor {
permit_changer: None,
inner,
}
todo!()
//
// let inner: Arc<Mutex<ReceivedProcessorInner<R>>> =
// Arc::new(Mutex::new(ReceivedProcessorInner {
// test_nonce: None,
// packets_receiver,
// client_encryption_keypair,
// message_receiver: R::new(),
// received_packets: Vec::new(),
// }));
//
// ReceivedProcessor {
// permit_changer: None,
// inner,
// }
}
pub(crate) fn start_receiving(&mut self) {
@@ -150,13 +193,7 @@ impl<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,
},
}
nym-api/src/network_monitor/monitor/receiver.rs
+4 -4
View File
@@ -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,12 @@ impl PacketReceiver {
// unwrap here is fine as it can only return a `None` if the PacketSender has died
// and if that was the case, then the entire monitor is already in an undefined state
update = self.clients_updater.next() => self.process_gateway_update(update.unwrap()),
gateway_message = self.gateways_reader.next() => {
let Some((_gateway_id, message)) = gateway_message else {
gateway_messages = self.gateways_reader.next() => {
let Some((_gateway_id, messages)) = gateway_messages else {
log::error!("the gateways reader stream has terminated!");
continue
};
self.process_gateway_messages(message)
self.process_gateway_messages(messages)
}
}
}
nym-api/src/network_monitor/monitor/summary_producer.rs
+5 -5
View File
@@ -1,10 +1,10 @@
// 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_route::TestRoute;
use nym_mixnet_contract_common::MixId;
use nym_node_tester_utils::node::TestableNode;
use std::collections::HashMap;
use std::fmt::{Display, Formatter};
@@ -281,8 +281,8 @@ impl SummaryProducer {
pub(super) fn produce_summary(
&self,
tested_mixnodes: Vec<TestedNode>,
tested_gateways: Vec<TestedNode>,
tested_mixnodes: Vec<TestableNode>,
tested_gateways: Vec<TestableNode>,
received_packets: Vec<TestPacket>,
invalid_mixnodes: Vec<InvalidNode>,
invalid_gateways: Vec<InvalidNode>,
nym-api/src/network_monitor/test_packet.rs
+22 -242
View File
@@ -1,260 +1,40 @@
// 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_node_tester_utils::error::NetworkTestingError;
use nym_node_tester_utils::TestMessage;
use nym_topology::{gateway, mix};
use std::convert::TryInto;
use std::fmt::{self, Display, Formatter};
use std::hash::{Hash, Hasher};
use std::mem;
use std::str::Utf8Error;
use thiserror::Error;
use serde::{Deserialize, Serialize};
const MIXNODE_TYPE: u8 = 0;
const GATEWAY_TYPE: u8 = 1;
pub(crate) type NodeTestMessage = TestMessage<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 fn mix_plaintexts(
&self,
node: &mix::Node,
test_packets: u32,
) -> Result<Vec<Vec<u8>>, NetworkTestingError> {
NodeTestMessage::mix_plaintexts(node, test_packets, *self)
}
pub(crate) fn new(
pub_key: identity::PublicKey,
owner: String,
route_id: u64,
test_nonce: u64,
node_type: NodeType,
) -> Self {
TestPacket {
route_id,
test_nonce,
pub_key,
owner,
node_type,
}
}
pub(crate) fn test_nonce(&self) -> u64 {
self.test_nonce
}
pub(crate) fn to_bytes(&self) -> Vec<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 gateway_plaintexts(
&self,
node: &gateway::Node,
test_packets: u32,
) -> Result<Vec<Vec<u8>>, NetworkTestingError> {
NodeTestMessage::gateway_plaintexts(node, test_packets, *self)
}
}
nym-api/src/network_monitor/test_route/mod.rs
+16 -31
View File
@@ -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};
@@ -70,30 +70,21 @@ impl TestRoute {
&self.nodes
}
pub(crate) fn self_test_packet(&self) -> TestPacket {
pub(crate) fn test_message_ext(&self, test_nonce: u64) -> NymApiTestMessageExt {
NymApiTestMessageExt::new(self.id, test_nonce)
}
pub(crate) fn self_test_messages(&self, count: usize) -> Vec<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()
}
}
@@ -104,16 +95,10 @@ impl Debug for TestRoute {
"[v{}] Route {}: [G] {} => [M1] {} => [M2] {} => [M3] {}",
self.system_version,
self.id,
self.nodes.gateways()[0].identity().to_base58_string(),
self.nodes.mixes_in_layer(1)[0]
.identity_key
.to_base58_string(),
self.nodes.mixes_in_layer(2)[0]
.identity_key
.to_base58_string(),
self.nodes.mixes_in_layer(3)[0]
.identity_key
.to_base58_string()
self.gateway().identity().to_base58_string(),
self.layer_one_mix().identity_key.to_base58_string(),
self.layer_two_mix().identity_key.to_base58_string(),
self.layer_three_mix().identity_key.to_base58_string()
)
}
}