From e6460208d37e8643edb0307fbb8a84e8d8060c32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Fri, 2 Jul 2021 11:31:29 +0100 Subject: [PATCH] Removed standalone network-monitor (#669) It has been integrated into the validator-api --- network-monitor/Cargo.toml | 35 -- network-monitor/src/chunker.rs | 75 --- network-monitor/src/gateways_reader.rs | 183 ------- network-monitor/src/main.rs | 356 ------------- network-monitor/src/monitor/mod.rs | 220 -------- network-monitor/src/monitor/preparer.rs | 477 ------------------ network-monitor/src/monitor/processor.rs | 227 --------- network-monitor/src/monitor/receiver.rs | 69 --- network-monitor/src/monitor/sender.rs | 448 ---------------- .../src/monitor/summary_producer.rs | 270 ---------- network-monitor/src/node_status_api/client.rs | 110 ---- network-monitor/src/node_status_api/mod.rs | 73 --- network-monitor/src/node_status_api/models.rs | 68 --- network-monitor/src/test_packet.rs | 214 -------- .../src/tested_network/good_topology.rs | 29 -- network-monitor/src/tested_network/mod.rs | 72 --- 16 files changed, 2926 deletions(-) delete mode 100644 network-monitor/Cargo.toml delete mode 100644 network-monitor/src/chunker.rs delete mode 100644 network-monitor/src/gateways_reader.rs delete mode 100644 network-monitor/src/main.rs delete mode 100644 network-monitor/src/monitor/mod.rs delete mode 100644 network-monitor/src/monitor/preparer.rs delete mode 100644 network-monitor/src/monitor/processor.rs delete mode 100644 network-monitor/src/monitor/receiver.rs delete mode 100644 network-monitor/src/monitor/sender.rs delete mode 100644 network-monitor/src/monitor/summary_producer.rs delete mode 100644 network-monitor/src/node_status_api/client.rs delete mode 100644 network-monitor/src/node_status_api/mod.rs delete mode 100644 network-monitor/src/node_status_api/models.rs delete mode 100644 network-monitor/src/test_packet.rs delete mode 100644 network-monitor/src/tested_network/good_topology.rs delete mode 100644 network-monitor/src/tested_network/mod.rs diff --git a/network-monitor/Cargo.toml b/network-monitor/Cargo.toml deleted file mode 100644 index ef7b52c305..0000000000 --- a/network-monitor/Cargo.toml +++ /dev/null @@ -1,35 +0,0 @@ -# Copyright 2020 - Nym Technologies SA -# SPDX-License-Identifier: Apache-2.0 - -[package] -name = "nym-validator-api" -version = "0.10.1" -authors = ["Dave Hrycyszyn ", "Jędrzej Stuczyński "] -edition = "2018" - -# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html - -[dependencies] -clap = "2.33.0" -dotenv = "0.15.0" -futures = "0.3" -log = "0.4" -pin-project = "1.0" -pretty_env_logger = "0.4" -rand = "0.7" -reqwest = { version = "0.11", features = ["json"] } -serde = "1.0" -serde_json = "1.0" -tokio = { version = "1.4", features = ["rt-multi-thread", "macros", "signal", "time"] } - -## internal -crypto = { path = "../common/crypto" } -gateway-client = { path = "../common/client-libs/gateway-client" } -gateway-requests = { path = "../gateway/gateway-requests" } -mixnet-contract = { path = "../common/mixnet-contract" } -nymsphinx = { path = "../common/nymsphinx" } -topology = { path = "../common/topology" } -validator-client = { path = "../common/client-libs/validator-client" } -version-checker = { path = "../common/version-checker" } - -[dev-dependencies] diff --git a/network-monitor/src/chunker.rs b/network-monitor/src/chunker.rs deleted file mode 100644 index b2bc7c1736..0000000000 --- a/network-monitor/src/chunker.rs +++ /dev/null @@ -1,75 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use nymsphinx::forwarding::packet::MixPacket; -use nymsphinx::params::PacketMode; -use nymsphinx::{ - acknowledgements::AckKey, addressing::clients::Recipient, preparer::MessagePreparer, -}; -use rand::rngs::OsRng; -use std::time::Duration; -use topology::NymTopology; - -const DEFAULT_AVERAGE_PACKET_DELAY: Duration = Duration::from_millis(200); -const DEFAULT_AVERAGE_ACK_DELAY: Duration = Duration::from_millis(200); - -pub(crate) struct Chunker { - rng: OsRng, - message_preparer: MessagePreparer, -} - -impl Chunker { - pub(crate) fn new(tested_mix_me: Recipient) -> Self { - Chunker { - rng: OsRng, - message_preparer: MessagePreparer::new( - OsRng, - tested_mix_me, - DEFAULT_AVERAGE_PACKET_DELAY, - DEFAULT_AVERAGE_ACK_DELAY, - PacketMode::Mix, - None, - ), - } - } - - pub(crate) async fn prepare_packets_from( - &mut self, - message: Vec, - topology: &NymTopology, - packet_sender: Recipient, - ) -> Vec { - // I really dislike how we have to overwrite the parameter of the `MessagePreparer` on each run - // but without some significant API changes in the `MessagePreparer` this was the easiest - // way to being able to have variable sender address. - self.message_preparer.set_sender_address(packet_sender); - self.prepare_packets(message, topology, packet_sender).await - } - - async fn prepare_packets( - &mut self, - message: Vec, - topology: &NymTopology, - packet_sender: Recipient, - ) -> Vec { - let ack_key: AckKey = AckKey::new(&mut self.rng); - - let (split_message, _reply_keys) = self - .message_preparer - .prepare_and_split_message(message, false, topology) - .expect("failed to split the message"); - - 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) - .await - .unwrap(); - - mix_packets.push(prepared_fragment.mix_packet); - } - mix_packets - } -} diff --git a/network-monitor/src/gateways_reader.rs b/network-monitor/src/gateways_reader.rs deleted file mode 100644 index e4288caec4..0000000000 --- a/network-monitor/src/gateways_reader.rs +++ /dev/null @@ -1,183 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crypto::asymmetric::identity; -use futures::stream::Stream; -use futures::task::Context; -use gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver}; -use log::*; -use std::pin::Pin; -use std::task::{Poll, Waker}; - -/// Constant used to determine maximum number of times the GatewayReader can poll. It basically -/// tries to solve the same problem that `FuturesUnordered` has: https://github.com/rust-lang/futures-rs/issues/2047 -const YIELD_EVERY: usize = 32; - -// TODO: Originally I set it to (identity::PublicKey, Vec>) and I definitely -// had a reason for doing so, but right now I can't remember what that was... -pub(crate) type GatewayMessages = Vec>; - -pub(crate) struct GatewayChannel { - id: identity::PublicKey, - message_receiver: MixnetMessageReceiver, - ack_receiver: AcknowledgementReceiver, - is_closed: bool, -} - -impl GatewayChannel { - pub(crate) fn new( - id: identity::PublicKey, - message_receiver: MixnetMessageReceiver, - ack_receiver: AcknowledgementReceiver, - ) -> Self { - GatewayChannel { - id, - message_receiver, - ack_receiver, - is_closed: false, - } - } -} - -impl Stream for GatewayChannel { - type Item = Vec>; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - if self.is_closed { - return Poll::Ready(None); - } - - let mut polled = 0; - - // empty the ack channel if anything is on it (we don't care about the content at all at the - // moment) - while let Poll::Ready(_ack) = Pin::new(&mut self.ack_receiver).poll_next(cx) { - polled += 1; - } - - let item = futures::ready!(Pin::new(&mut self.message_receiver).poll_next(cx)); - - match item { - None => Poll::Ready(None), - // if we managed to get an item, try to also read additional ones - Some(mut messages) => { - while let Poll::Ready(new_item) = Pin::new(&mut self.message_receiver).poll_next(cx) - { - polled += 1; - match new_item { - None => { - self.is_closed = true; - cx.waker().wake_by_ref(); - return Poll::Ready(Some(messages)); - } - Some(mut additional_messages) => { - messages.append(&mut additional_messages); - - // it is fine enough to use the same constant here as in the main GatewayReader - // as only a single channel, i.e. the main gateway will be capable - // of returning more than 2 values - if polled == YIELD_EVERY { - cx.waker().wake_by_ref(); - break; - } - } - } - } - Poll::Ready(Some(messages)) - } - } - } -} - -pub(crate) struct GatewaysReader { - latest_read: usize, - // channels: FuturesUnordered, - channels: Vec, - waker: Option, -} - -impl GatewaysReader { - pub(crate) fn new() -> Self { - GatewaysReader { - latest_read: 0, - channels: Vec::new(), - waker: None, - } - } - - fn remove_nth(&mut self, i: usize) { - self.channels.remove(i); - } - - // todo: if we find that this method is called frequently, perhaps the vector should get - // replaced with different data structure - pub(crate) fn remove_by_key(&mut self, key: identity::PublicKey) { - match self.channels.iter().position(|item| item.id == key) { - Some(i) => { - self.channels.remove(i); - } - // this shouldn't ever get thrown, so perhaps a panic would be more in order? - None => error!( - "tried to remove gateway reader {} but it doesn't exist!", - key.to_base58_string() - ), - } - } - - fn poll_nth( - &mut self, - cx: &mut Context<'_>, - i: usize, - ) -> Option>> { - if let Poll::Ready(item) = Pin::new(&mut self.channels[i]).poll_next(cx) { - self.latest_read = i; - match item { - // Some(messages) => return Some(Poll::Ready(Some((self.channels[i].id, messages)))), - Some(messages) => return Some(Poll::Ready(Some(messages))), - // remove dead channel - None => self.remove_nth(i), - } - } - None - } - - pub(crate) fn insert_channel(&mut self, channel: GatewayChannel) { - self.channels.push(channel); - if let Some(waker) = self.waker.take() { - waker.wake() - } - } -} - -// TODO: not sure if this will scale well, but I don't know what would be a good alternative, -// perhaps try to somehow incorporate FuturesUnordered? -// also, perhaps reading should be done in parallel? -impl Stream for GatewaysReader { - // item represents gateway that returned messages alongside the actual messages - type Item = GatewayMessages; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - if self.latest_read >= self.channels.len() { - self.latest_read = 0; - } - - // don't start reading from beginning each time to at least slightly help with the bias - for i in self.latest_read..self.channels.len() { - if let Some(item) = self.poll_nth(cx, i) { - return item; - } - } - - for i in 0..self.latest_read { - if let Some(item) = self.poll_nth(cx, i) { - return item; - } - } - - // if we have no channels available, store the waker to be woken when a new one is pushed - if self.channels.is_empty() { - self.waker = Some(cx.waker().clone()) - } - Poll::Pending - } -} diff --git a/network-monitor/src/main.rs b/network-monitor/src/main.rs deleted file mode 100644 index 1813ebecfa..0000000000 --- a/network-monitor/src/main.rs +++ /dev/null @@ -1,356 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::monitor::preparer::PacketPreparer; -use crate::monitor::processor::{ - ReceivedProcessor, ReceivedProcessorReceiver, ReceivedProcessorSender, -}; -use crate::monitor::receiver::{ - GatewayClientUpdateReceiver, GatewayClientUpdateSender, PacketReceiver, -}; -use crate::monitor::sender::PacketSender; -use crate::monitor::summary_producer::SummaryProducer; -use crate::tested_network::good_topology::parse_topology_file; -use crate::tested_network::TestedNetwork; -use clap::{App, Arg, ArgMatches}; -use crypto::asymmetric::{encryption, identity}; -use futures::channel::mpsc; -use log::*; -use nymsphinx::addressing::clients::Recipient; -use std::sync::Arc; -use std::time::Duration; -use topology::NymTopology; - -mod chunker; -pub(crate) mod gateways_reader; -mod monitor; -mod node_status_api; -mod test_packet; -mod tested_network; - -const V4_TOPOLOGY_ARG: &str = "v4-topology-filepath"; -const V6_TOPOLOGY_ARG: &str = "v6-topology-filepath"; -const VALIDATORS_ARG: &str = "validators"; -const NODE_STATUS_API_ARG: &str = "node-status-api"; -const DETAILED_REPORT_ARG: &str = "detailed-report"; -const GATEWAY_SENDING_RATE_ARG: &str = "gateway-rate"; -const MIXNET_CONTRACT_ARG: &str = "mixnet-contract"; - -const DEFAULT_VALIDATORS: &[&str] = &[ - // "http://testnet-finney-validator.nymtech.net:1317", - "http://testnet-finney-validator2.nymtech.net:1317", - "http://mixnet.club:1317", -]; - -const DEFAULT_NODE_STATUS_API: &str = "http://localhost:8081"; -const DEFAULT_GATEWAY_SENDING_RATE: usize = 500; -const DEFAULT_MIXNET_CONTRACT: &str = "hal1k0jntykt7e4g3y88ltc60czgjuqdy4c9c6gv94"; - -pub(crate) const TIME_CHUNK_SIZE: Duration = Duration::from_millis(50); -pub(crate) const PENALISE_OUTDATED: bool = false; - -// TODO: let's see how it goes and whether those new adjusting -const MAX_CONCURRENT_GATEWAY_CLIENTS: Option = Some(50); -const GATEWAY_RESPONSE_TIMEOUT: Duration = Duration::from_millis(1_500); -pub(crate) const GATEWAY_CONNECTION_TIMEOUT: Duration = Duration::from_millis(2_500); - -fn parse_args<'a>() -> ArgMatches<'a> { - App::new("Nym Network Monitor") - .author("Nymtech") - .arg( - Arg::with_name(V4_TOPOLOGY_ARG) - .help("location of .json file containing IPv4 'good' network topology") - .long(V4_TOPOLOGY_ARG) - .takes_value(true) - .required(true), - ) - .arg( - Arg::with_name(V6_TOPOLOGY_ARG) - .help("location of .json file containing IPv6 'good' network topology") - .long(V6_TOPOLOGY_ARG) - .takes_value(true) - .required(true), - ) - .arg( - Arg::with_name(VALIDATORS_ARG) - .help("REST endpoint of the validator the monitor will grab nodes to test") - .long(VALIDATORS_ARG) - .takes_value(true) - ) - .arg(Arg::with_name("mixnet-contract") - .long(MIXNET_CONTRACT_ARG) - .help("Address of the validator contract managing the network") - .takes_value(true), - ) - .arg( - Arg::with_name(NODE_STATUS_API_ARG) - .help("Address of the node status api to submit results to. Most likely it's a local address") - .long(NODE_STATUS_API_ARG) - .takes_value(true) - ) - .arg( - Arg::with_name(DETAILED_REPORT_ARG) - .help("specifies whether a detailed report should be printed after each run") - .long(DETAILED_REPORT_ARG) - ) - .arg(Arg::with_name(GATEWAY_SENDING_RATE_ARG) - .help("specifies maximum rate (in packets per second) of test packets being sent to gateway") - .takes_value(true) - .long(GATEWAY_SENDING_RATE_ARG) - .short("r") - ) - .get_matches() -} - -#[tokio::main] -async fn main() { - println!("Network monitor starting..."); - let matches = parse_args(); - let v4_topology_path = matches.value_of(V4_TOPOLOGY_ARG).unwrap(); - let v6_topology_path = matches.value_of(V6_TOPOLOGY_ARG).unwrap(); - - let v4_topology = parse_topology_file(v4_topology_path); - let v6_topology = parse_topology_file(v6_topology_path); - - let validators_rest_uris_borrowed = matches - .values_of(VALIDATORS_ARG) - .map(|args| args.collect::>()) - .unwrap_or_else(|| DEFAULT_VALIDATORS.to_vec()); - - let validators_rest_uris = validators_rest_uris_borrowed - .into_iter() - .map(|uri| uri.to_string()) - .collect::>(); - - let node_status_api_uri = matches - .value_of(NODE_STATUS_API_ARG) - .unwrap_or(DEFAULT_NODE_STATUS_API); - - let mixnet_contract = matches - .value_of(MIXNET_CONTRACT_ARG) - .unwrap_or(DEFAULT_MIXNET_CONTRACT); - - let detailed_report = matches.is_present(DETAILED_REPORT_ARG); - let sending_rate = matches - .value_of(GATEWAY_SENDING_RATE_ARG) - .map(|v| v.parse().unwrap()) - .unwrap_or_else(|| DEFAULT_GATEWAY_SENDING_RATE); - - check_if_up_to_date(&v4_topology, &v6_topology); - setup_logging(); - - println!("* validator servers: {:?}", validators_rest_uris); - println!("* node status api server: {}", node_status_api_uri); - println!("* mixnet contract: {}", mixnet_contract); - println!("* detailed report printing: {}", detailed_report); - println!("* gateway sending rate: {} packets/s", sending_rate); - - // TODO: in the future I guess this should somehow change to distribute the load - let tested_mix_gateway = v4_topology.gateways()[0].clone(); - println!( - "* gateway for testing mixnodes: {}", - tested_mix_gateway.identity_key.to_base58_string() - ); - - // TODO: those keys change constant throughout the whole execution of the monitor. - // and on top of that, they are used with ALL the gateways -> presumably this should change - // in the future - let mut rng = rand::rngs::OsRng; - - let identity_keypair = Arc::new(identity::KeyPair::new(&mut rng)); - let encryption_keypair = Arc::new(encryption::KeyPair::new(&mut rng)); - - let test_mixnode_sender = Recipient::new( - *identity_keypair.public_key(), - *encryption_keypair.public_key(), - tested_mix_gateway.identity_key, - ); - - let tested_network = TestedNetwork::new_good(v4_topology, v6_topology); - let validator_client = new_validator_client(validators_rest_uris, mixnet_contract); - let node_status_api_client = new_node_status_api_client(node_status_api_uri); - - 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( - validator_client, - tested_network.clone(), - test_mixnode_sender, - *identity_keypair.public_key(), - *encryption_keypair.public_key(), - ); - - let packet_sender = new_packet_sender( - gateway_status_update_sender, - Arc::clone(&identity_keypair), - sending_rate, - ); - let received_processor = new_received_processor( - received_processor_receiver_channel, - Arc::clone(&encryption_keypair), - ); - let summary_producer = new_summary_producer(detailed_report); - let mut packet_receiver = new_packet_receiver( - gateway_status_update_receiver, - received_processor_sender_channel, - ); - - let mut monitor = monitor::Monitor::new( - packet_preparer, - packet_sender, - received_processor, - summary_producer, - node_status_api_client, - tested_network, - ); - - tokio::spawn(async move { packet_receiver.run().await }); - - tokio::spawn(async move { monitor.run().await }); - - wait_for_interrupt().await -} - -async fn wait_for_interrupt() { - if let Err(e) = tokio::signal::ctrl_c().await { - error!( - "There was an error while capturing SIGINT - {:?}. We will terminate regardless", - e - ); - } - println!("Received SIGINT - the network monitor will terminate now"); -} - -fn new_packet_preparer( - validator_client: validator_client::Client, - tested_network: TestedNetwork, - test_mixnode_sender: Recipient, - self_public_identity: identity::PublicKey, - self_public_encryption: encryption::PublicKey, -) -> PacketPreparer { - PacketPreparer::new( - validator_client, - tested_network, - test_mixnode_sender, - self_public_identity, - self_public_encryption, - ) -} - -fn new_packet_sender( - gateways_status_updater: GatewayClientUpdateSender, - local_identity: Arc, - max_sending_rate: usize, -) -> PacketSender { - PacketSender::new( - gateways_status_updater, - local_identity, - GATEWAY_RESPONSE_TIMEOUT, - MAX_CONCURRENT_GATEWAY_CLIENTS, - max_sending_rate, - ) -} - -fn new_received_processor( - packets_receiver: ReceivedProcessorReceiver, - client_encryption_keypair: Arc, -) -> ReceivedProcessor { - ReceivedProcessor::new(packets_receiver, client_encryption_keypair) -} - -fn new_summary_producer(detailed_report: bool) -> SummaryProducer { - // right now always print the basic report. If we feel like we need to change it, it can - // be easily adjusted by adding some flag or something - let summary_producer = SummaryProducer::default().with_report(); - if detailed_report { - summary_producer.with_detailed_report() - } else { - summary_producer - } -} - -fn new_packet_receiver( - gateways_status_updater: GatewayClientUpdateReceiver, - processor_packets_sender: ReceivedProcessorSender, -) -> PacketReceiver { - PacketReceiver::new(gateways_status_updater, processor_packets_sender) -} - -fn new_validator_client( - validator_rest_uris: Vec, - mixnet_contract: &str, -) -> validator_client::Client { - let config = validator_client::Config::new(validator_rest_uris, mixnet_contract); - validator_client::Client::new(config) -} - -fn new_node_status_api_client>(base_url: S) -> node_status_api::Client { - let config = node_status_api::Config::new(base_url); - node_status_api::Client::new(config) -} - -fn setup_logging() { - let mut log_builder = pretty_env_logger::formatted_timed_builder(); - if let Ok(s) = ::std::env::var("RUST_LOG") { - log_builder.parse_filters(&s); - } else { - // default to 'Info' - log_builder.filter(None, log::LevelFilter::Info); - } - - log_builder - .filter_module("hyper", log::LevelFilter::Warn) - .filter_module("tokio_reactor", log::LevelFilter::Warn) - .filter_module("reqwest", log::LevelFilter::Warn) - .filter_module("mio", log::LevelFilter::Warn) - .filter_module("want", log::LevelFilter::Warn) - .filter_module("sled", log::LevelFilter::Warn) - .filter_module("tungstenite", log::LevelFilter::Warn) - .filter_module("tokio_tungstenite", log::LevelFilter::Warn) - .init(); -} - -fn check_if_up_to_date(v4_topology: &NymTopology, v6_topology: &NymTopology) { - let monitor_version = env!("CARGO_PKG_VERSION"); - for (_, layer_mixes) in v4_topology.mixes().iter() { - for mix in layer_mixes.iter() { - if !version_checker::is_minor_version_compatible(monitor_version, &*mix.version) { - panic!( - "Our good topology is not compatible with monitor! Mix runs {}, we have {}", - mix.version, monitor_version - ) - } - } - } - - for gateway in v4_topology.gateways().iter() { - if !version_checker::is_minor_version_compatible(monitor_version, &*gateway.version) { - panic!( - "Our good topology is not compatible with monitor! Gateway runs {}, we have {}", - gateway.version, monitor_version - ) - } - } - - for (_, layer_mixes) in v6_topology.mixes().iter() { - for mix in layer_mixes.iter() { - if !version_checker::is_minor_version_compatible(monitor_version, &*mix.version) { - panic!( - "Our good topology is not compatible with monitor! Mix runs {}, we have {}", - mix.version, monitor_version - ) - } - } - } - - for gateway in v6_topology.gateways().iter() { - if !version_checker::is_minor_version_compatible(monitor_version, &*gateway.version) { - panic!( - "Our good topology is not compatible with monitor! Gateway runs {}, we have {}", - gateway.version, monitor_version - ) - } - } -} diff --git a/network-monitor/src/monitor/mod.rs b/network-monitor/src/monitor/mod.rs deleted file mode 100644 index a3c1a6ca96..0000000000 --- a/network-monitor/src/monitor/mod.rs +++ /dev/null @@ -1,220 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::monitor::preparer::{PacketPreparer, TestedNode}; -use crate::monitor::processor::ReceivedProcessor; -use crate::monitor::sender::PacketSender; -use crate::monitor::summary_producer::{SummaryProducer, TestReport}; -use crate::node_status_api; -use crate::node_status_api::models::{BatchGatewayStatus, BatchMixStatus}; -use crate::test_packet::NodeType; -use crate::tested_network::TestedNetwork; -use log::*; -use tokio::time::{sleep, Duration, Instant}; - -pub(crate) mod preparer; -pub(crate) mod processor; -pub(crate) mod receiver; -pub(crate) mod sender; -pub(crate) mod summary_producer; - -const PACKET_DELIVERY_TIMEOUT: Duration = Duration::from_secs(20); -const MONITOR_RUN_INTERVAL: Duration = Duration::from_secs(15 * 60); -const GATEWAY_PING_INTERVAL: Duration = Duration::from_secs(60); - -pub(super) struct Monitor { - nonce: u64, - packet_preparer: PacketPreparer, - packet_sender: PacketSender, - received_processor: ReceivedProcessor, - summary_producer: SummaryProducer, - node_status_api_client: node_status_api::Client, - tested_network: TestedNetwork, -} - -impl Monitor { - pub(super) fn new( - packet_preparer: PacketPreparer, - packet_sender: PacketSender, - received_processor: ReceivedProcessor, - summary_producer: SummaryProducer, - node_status_api_client: node_status_api::Client, - tested_network: TestedNetwork, - ) -> Self { - Monitor { - nonce: 1, - packet_preparer, - packet_sender, - received_processor, - summary_producer, - node_status_api_client, - tested_network, - } - } - - // while it might have been cleaner to put this into a separate `Notifier` structure, - // I don't see much point considering it's only a single, small, method - async fn notify_node_status_api( - &self, - mix_status: BatchMixStatus, - gateway_status: BatchGatewayStatus, - ) { - if let Err(err) = self - .node_status_api_client - .post_batch_mix_status(mix_status) - .await - { - warn!( - "Failed to send batch mix status to node status api - {:?}", - err - ) - } - - if let Err(err) = self - .node_status_api_client - .post_batch_gateway_status(gateway_status) - .await - { - warn!( - "Failed to send batch mix status to node status api - {:?}", - err - ) - } - } - - // checking it this way with a TestReport is rather suboptimal but given the fact we're only - // doing this fewer than 10 times, it's not that problematic - fn check_good_nodes_status(&self, report: &TestReport) -> bool { - for v4_mixes in self.tested_network.v4_topology().mixes().values() { - for v4_mix in v4_mixes { - let node = &TestedNode { - identity: v4_mix.identity_key.to_base58_string(), - owner: v4_mix.owner.clone(), - node_type: NodeType::Mixnode, - }; - if !report.fully_working_mixes.contains(node) { - return false; - } - } - } - - for v4_gateway in self.tested_network.v4_topology().gateways() { - let node = &TestedNode { - identity: v4_gateway.identity_key.to_base58_string(), - owner: v4_gateway.owner.clone(), - node_type: NodeType::Gateway, - }; - if !report.fully_working_gateways.contains(node) { - return false; - } - } - - for v6_mixes in self.tested_network.v6_topology().mixes().values() { - for v6_mix in v6_mixes { - let node = &TestedNode { - identity: v6_mix.identity_key.to_base58_string(), - owner: v6_mix.owner.clone(), - node_type: NodeType::Mixnode, - }; - if !report.fully_working_mixes.contains(node) { - return false; - } - } - } - - for v6_gateway in self.tested_network.v6_topology().gateways() { - let node = &TestedNode { - identity: v6_gateway.identity_key.to_base58_string(), - owner: v6_gateway.owner.clone(), - node_type: NodeType::Gateway, - }; - if !report.fully_working_gateways.contains(node) { - return false; - } - } - - true - } - - async fn test_run(&mut self) { - info!(target: "Monitor", "Starting test run no. {}", self.nonce); - - debug!(target: "Monitor", "preparing mix packets to all nodes..."); - let prepared_packets = match self.packet_preparer.prepare_test_packets(self.nonce).await { - Ok(packets) => packets, - Err(err) => { - error!("failed to create packets for the test run - {:?}", err); - // TODO: return error? - return; - } - }; - - self.received_processor.set_new_expected(self.nonce).await; - - info!(target: "Monitor", "starting to send all the packets..."); - self.packet_sender - .send_packets(prepared_packets.packets) - .await; - - info!(target: "Monitor", "sending is over, waiting for {:?} before checking what we received", PACKET_DELIVERY_TIMEOUT); - - // give the packets some time to traverse the network - sleep(PACKET_DELIVERY_TIMEOUT).await; - - let received = self.received_processor.return_received().await; - - let test_summary = self.summary_producer.produce_summary( - prepared_packets.tested_nodes, - received, - prepared_packets.invalid_nodes, - ); - - // our "good" nodes MUST be working correctly otherwise we cannot trust the results - if self.check_good_nodes_status(&test_summary.test_report) { - self.notify_node_status_api( - test_summary.batch_mix_status, - test_summary.batch_gateway_status, - ) - .await; - } else { - error!("our own 'good' nodes did not pass the check - we are not going to submit results to the node status API"); - } - - self.nonce += 1; - } - - async fn ping_all_gateways(&mut self) { - self.packet_sender.ping_all_active_gateways().await; - } - - pub(crate) async fn run(&mut self) { - // start from 0 to run test immediately on startup - let test_delay = sleep(Duration::from_secs(0)); - tokio::pin!(test_delay); - - let ping_delay = sleep(GATEWAY_PING_INTERVAL); - tokio::pin!(ping_delay); - - loop { - tokio::select! { - _ = &mut test_delay => { - self.test_run().await; - info!(target: "Monitor", "Next test run will happen in {:?}", MONITOR_RUN_INTERVAL); - - let now = Instant::now(); - test_delay.as_mut().reset(now + MONITOR_RUN_INTERVAL); - // since we just sent packets through gateways, there's no need to ping them - ping_delay.as_mut().reset(now + GATEWAY_PING_INTERVAL); - - } - _ = &mut ping_delay => { - info!(target: "Monitor", "Pinging all active gateways"); - self.ping_all_gateways().await; - - let now = Instant::now(); - ping_delay.as_mut().reset(now + GATEWAY_PING_INTERVAL); - } - } - } - } -} diff --git a/network-monitor/src/monitor/preparer.rs b/network-monitor/src/monitor/preparer.rs deleted file mode 100644 index a1aba7b924..0000000000 --- a/network-monitor/src/monitor/preparer.rs +++ /dev/null @@ -1,477 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::chunker::Chunker; -use crate::monitor::sender::GatewayPackets; -use crate::test_packet::{NodeType, TestPacket}; -use crate::tested_network::TestedNetwork; -use crypto::asymmetric::{encryption, identity}; -use log::*; -use mixnet_contract::{GatewayBond, MixNodeBond}; -use nymsphinx::addressing::clients::Recipient; -use nymsphinx::forwarding::packet::MixPacket; -use std::convert::TryInto; -use std::fmt::{self, Display, Formatter}; -use topology::{gateway, mix}; -use validator_client::ValidatorClientError; - -#[derive(Debug)] -pub(super) enum PacketPreparerError { - ValidatorClientError(ValidatorClientError), -} - -// declared type aliases for easier code reasoning -type Version = String; -type Id = String; -type Owner = String; - -#[derive(Clone)] -pub(crate) enum InvalidNode { - OutdatedMix(Id, Owner, Version), - MalformedMix(Id, Owner), - OutdatedGateway(Id, Owner, Version), - MalformedGateway(Id, Owner), -} - -impl Display for InvalidNode { - fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - match self { - InvalidNode::OutdatedMix(id, owner, version) => { - write!( - f, - "Mixnode {} (v {}) owned by {} is outdated", - id, version, owner - ) - } - InvalidNode::MalformedMix(id, owner) => { - write!(f, "Mixnode {} owner by {} is malformed", id, owner) - } - InvalidNode::OutdatedGateway(id, owner, version) => { - write!( - f, - "Gateway {} (v {}) owned by {} is outdated", - id, version, owner - ) - } - InvalidNode::MalformedGateway(id, owner) => { - write!(f, "Gateway {} owned by {} is malformed", id, owner) - } - } - } -} - -enum PreparedNode { - TestedGateway(gateway::Node, [TestPacket; 2]), - TestedMix(mix::Node, [TestPacket; 2]), - Invalid(InvalidNode), -} - -#[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 new_mix(identity: String, owner: String) -> Self { - TestedNode { - identity, - owner, - node_type: NodeType::Mixnode, - } - } - - pub(crate) fn new_gateway(identity: String, owner: String) -> Self { - TestedNode { - identity, - owner, - node_type: NodeType::Gateway, - } - } - - pub(crate) fn from_raw_mix(identity: S1, owner: S2) -> Self - where - S1: Into, - S2: Into, - { - TestedNode { - identity: identity.into(), - owner: owner.into(), - node_type: NodeType::Mixnode, - } - } - - pub(crate) fn from_raw_gateway(identity: S1, owner: S2) -> Self - where - S1: Into, - S2: Into, - { - TestedNode { - identity: identity.into(), - owner: owner.into(), - node_type: NodeType::Gateway, - } - } - - pub(crate) fn is_gateway(&self) -> bool { - self.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 { - /// All packets that are going to get sent during the test as well as the gateways through - /// which they ought to be sent. - pub(super) packets: Vec, - - /// Vector containing list of public keys and owners of all nodes (mixnodes and gateways) being tested. - /// We do not need to specify the exact test parameters as for each of them we expect to receive - /// two packets back: ipv4 and ipv6 regardless of which gateway they originate. - pub(super) tested_nodes: Vec, - - /// All nodes that failed to get parsed correctly. They will be marked to the validator as being - /// down on ipv4 and ipv6. - pub(super) invalid_nodes: Vec, -} - -pub(crate) struct PacketPreparer { - chunker: Chunker, - validator_client: validator_client::Client, - tested_network: TestedNetwork, - - // currently all test MIXNODE packets are sent via the same gateway - test_mixnode_sender: Recipient, - - // keys required to create sender of any other gateway - self_public_identity: identity::PublicKey, - self_public_encryption: encryption::PublicKey, -} - -impl PacketPreparer { - pub(crate) fn new( - validator_client: validator_client::Client, - tested_network: TestedNetwork, - test_mixnode_sender: Recipient, - self_public_identity: identity::PublicKey, - self_public_encryption: encryption::PublicKey, - ) -> Self { - PacketPreparer { - chunker: Chunker::new(test_mixnode_sender), - validator_client, - tested_network, - test_mixnode_sender, - self_public_identity, - self_public_encryption, - } - } - - async fn get_network_nodes( - &mut self, - ) -> Result<(Vec, Vec), PacketPreparerError> { - info!(target: "Monitor", "Obtaining network topology..."); - - let mixnodes = match self.validator_client.get_mix_nodes().await { - Err(err) => { - error!("failed to get network mixnodes - {}", err); - return Err(PacketPreparerError::ValidatorClientError(err)); - } - Ok(mixes) => mixes, - }; - - let gateways = match self.validator_client.get_gateways().await { - Err(err) => { - error!("failed to get network gateways - {}", err); - return Err(PacketPreparerError::ValidatorClientError(err)); - } - Ok(gateways) => gateways, - }; - - info!(target: "Monitor", "Obtained network topology"); - - Ok((mixnodes, gateways)) - } - - fn check_version_compatibility(&self, mix_version: &str) -> bool { - let semver_compatibility = version_checker::is_minor_version_compatible( - mix_version, - self.tested_network.system_version(), - ); - - if semver_compatibility { - // this can't fail as we know it's semver compatible - let version = version_checker::parse_version(mix_version).unwrap(); - - // check if it's at least 0.9.2 - reject anything below it due to significant bugs present - // if it's 1.Y.Z it's definitely >= 0.9.2 - if version.major >= 1 { - return true; - } - // if it's 0.10.Z it's definitely >= 0.9.2 - if version.major == 0 && version.minor >= 10 { - return true; - } - // if it's 0.9.Z, ensure Z >= 2 - version.minor == 9 && version.patch >= 2 - } else { - false - } - } - - fn mix_into_prepared_node(&self, nonce: u64, mixnode_bond: &MixNodeBond) -> PreparedNode { - if !self.check_version_compatibility(&mixnode_bond.mix_node().version) { - return PreparedNode::Invalid(InvalidNode::OutdatedMix( - mixnode_bond.mix_node().identity_key.clone(), - mixnode_bond.owner.to_string(), - mixnode_bond.mix_node().version.clone(), - )); - } - match TryInto::::try_into(mixnode_bond) { - Ok(mix) => { - let v4_packet = TestPacket::new_v4( - mix.identity_key, - mix.owner.clone(), - nonce, - NodeType::Mixnode, - ); - let v6_packet = TestPacket::new_v6( - mix.identity_key, - mix.owner.clone(), - nonce, - NodeType::Mixnode, - ); - PreparedNode::TestedMix(mix, [v4_packet, v6_packet]) - } - Err(err) => { - warn!(target: "bad node", - "mix {} is malformed - {}", - mixnode_bond.mix_node().identity_key, - err - ); - PreparedNode::Invalid(InvalidNode::MalformedMix( - mixnode_bond.mix_node().identity_key.clone(), - mixnode_bond.owner.to_string(), - )) - } - } - } - - fn gateway_into_prepared_node(&self, nonce: u64, gateway_bond: &GatewayBond) -> PreparedNode { - if !self.check_version_compatibility(&gateway_bond.gateway().version) { - return PreparedNode::Invalid(InvalidNode::OutdatedGateway( - gateway_bond.gateway().identity_key.clone(), - gateway_bond.owner.to_string(), - gateway_bond.gateway().version.clone(), - )); - } - match TryInto::::try_into(gateway_bond) { - Ok(gateway) => { - let v4_packet = TestPacket::new_v4( - gateway.identity_key, - gateway.owner.clone(), - nonce, - NodeType::Gateway, - ); - let v6_packet = TestPacket::new_v6( - gateway.identity_key, - gateway.owner.clone(), - nonce, - NodeType::Gateway, - ); - PreparedNode::TestedGateway(gateway, [v4_packet, v6_packet]) - } - Err(err) => { - warn!(target: "bad node", - "gateway {} is malformed - {:?}", - gateway_bond.gateway().identity_key, - err - ); - PreparedNode::Invalid(InvalidNode::MalformedGateway( - gateway_bond.gateway().identity_key.clone(), - gateway_bond.owner.to_string(), - )) - } - } - } - - fn prepare_mixnodes(&self, nonce: u64, nodes: &[MixNodeBond]) -> Vec { - nodes - .iter() - .map(|mix| self.mix_into_prepared_node(nonce, mix)) - .collect() - } - - fn prepare_gateways(&self, nonce: u64, nodes: &[GatewayBond]) -> Vec { - nodes - .iter() - .map(|gateway| self.gateway_into_prepared_node(nonce, gateway)) - .collect() - } - - fn tested_nodes(&self, nodes: &[PreparedNode]) -> Vec { - nodes - .iter() - .filter_map(|node| match node { - PreparedNode::TestedGateway(gateway, _) => Some(TestedNode::new_gateway( - gateway.identity_key.to_base58_string(), - gateway.owner.clone(), - )), - PreparedNode::TestedMix(mix, _) => Some(TestedNode::new_mix( - mix.identity_key.to_base58_string(), - mix.owner.clone(), - )), - PreparedNode::Invalid(..) => None, - }) - .collect() - } - - fn create_packet_sender(&self, gateway: &gateway::Node) -> Recipient { - Recipient::new( - self.self_public_identity, - self.self_public_encryption, - gateway.identity_key, - ) - } - - async fn create_mixnode_mix_packets( - &mut self, - mixes: Vec, - invalid: &mut Vec, - ) -> Vec { - // all of the mixnode mix packets are going to get sent via our one 'main' gateway - // TODO: in the future this should probably be changed... - - // this might be slightly overestimating the number of packets we are going to produce, - // however, it should be negligible as we don't expect to ever see high number of - // invalid nodes (and even if we do see them, they shouldn't persist for long) - let mut packets = Vec::with_capacity(mixes.len() * 2); - - for mix in mixes.into_iter() { - match mix { - PreparedNode::TestedMix(node, test_packets) => { - for test_packet in test_packets.iter() { - let topology_to_test = self - .tested_network - .substitute_mix(node.clone(), test_packet.ip_version()); - let mix_message = test_packet.to_bytes(); - let mut mix_packet = self - .chunker - .prepare_packets_from( - mix_message, - &topology_to_test, - self.test_mixnode_sender, - ) - .await; - debug_assert_eq!(mix_packet.len(), 1); - packets.push(mix_packet.pop().unwrap()); - } - } - PreparedNode::Invalid(node) => invalid.push(node), - // `prepare_mixnodes` should NEVER return prepared gateways - _ => unreachable!(), - } - } - packets - } - - async fn create_gateway_mix_packets( - &mut self, - gateways: Vec, - invalid: &mut Vec, - ) -> Vec { - // again, there might be a slight overestimation here, but in the grand scheme of things - // it will be negligible - let mut packets = Vec::with_capacity(gateways.len()); - - // unfortunately this can't be done more cleanly with iterators as we require an async call - for gateway in gateways.into_iter() { - match gateway { - PreparedNode::TestedGateway(node, test_packets) => { - let mut gateway_packets = Vec::with_capacity(2); - for test_packet in test_packets.iter() { - let packet_sender = self.create_packet_sender(&node); - let topology_to_test = self - .tested_network - .substitute_gateway(node.clone(), test_packet.ip_version()); - let mix_message = test_packet.to_bytes(); - let mut mix_packet = self - .chunker - .prepare_packets_from(mix_message, &topology_to_test, packet_sender) - .await; - debug_assert_eq!(mix_packet.len(), 1); - - gateway_packets.push(mix_packet.pop().unwrap()); - } - packets.push(GatewayPackets::new( - node.clients_address(), - node.identity_key, - gateway_packets, - )) - } - PreparedNode::Invalid(node) => invalid.push(node), - // `prepare_gateways` should NEVER return prepared mixnodes - _ => unreachable!(), - } - } - - packets - } - - pub(super) async fn prepare_test_packets( - &mut self, - nonce: u64, - ) -> Result { - let (mixnode_bonds, gateway_bonds) = self.get_network_nodes().await?; - - let mut invalid_nodes = Vec::new(); - let mixes = self.prepare_mixnodes(nonce, &mixnode_bonds); - let gateways = self.prepare_gateways(nonce, &gateway_bonds); - - // get the keys of all nodes that will get tested this round - let tested_nodes: Vec<_> = self - .tested_nodes(&mixes) - .into_iter() - .chain(self.tested_nodes(&gateways).into_iter()) - .collect(); - - // those packets are going to go to our 'main' gateway - let mix_packets = self - .create_mixnode_mix_packets(mixes, &mut invalid_nodes) - .await; - - let main_gateway_id = self.tested_network.main_v4_gateway().identity_key; - - let mut gateway_packets = self - .create_gateway_mix_packets(gateways, &mut invalid_nodes) - .await; - - // check whether our 'good' gateway is being tested - if let Some(tested_main_gateway_packets) = gateway_packets - .iter_mut() - .find(|gateway| gateway.gateway_address() == main_gateway_id) - { - // we are testing the gateway we specified in our 'good' topology - tested_main_gateway_packets.push_packets(mix_packets); - } else { - // we are not testing the gateway from our 'good' topology -> it's probably - // situation similar to using 'good' qa-topology but testing testnet nodes. - let main_gateway_packets = GatewayPackets::new( - self.tested_network.main_v4_gateway().clients_address(), - main_gateway_id, - mix_packets, - ); - gateway_packets.push(main_gateway_packets); - } - - Ok(PreparedPackets { - packets: gateway_packets, - tested_nodes, - invalid_nodes, - }) - } -} diff --git a/network-monitor/src/monitor/processor.rs b/network-monitor/src/monitor/processor.rs deleted file mode 100644 index f2e9da15ee..0000000000 --- a/network-monitor/src/monitor/processor.rs +++ /dev/null @@ -1,227 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::gateways_reader::GatewayMessages; -use crate::test_packet::TestPacket; -use crypto::asymmetric::encryption; -use futures::channel::mpsc; -use futures::lock::{Mutex, MutexGuard}; -use futures::{SinkExt, StreamExt}; -use log::*; -use nymsphinx::receiver::MessageReceiver; -use std::fmt::{self, Display, Formatter}; -use std::mem; -use std::sync::Arc; - -pub(crate) type ReceivedProcessorSender = mpsc::UnboundedSender; -pub(crate) type ReceivedProcessorReceiver = mpsc::UnboundedReceiver; - -#[derive(Debug)] -enum ProcessingError { - MalformedPacketReceived, - NonTestPacketReceived, - NonMatchingNonce(u64), - ReceivedOutsideTestRun, -} - -impl Display for ProcessingError { - fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - match self { - ProcessingError::MalformedPacketReceived => write!(f, "received malformed packet"), - ProcessingError::NonTestPacketReceived => write!(f, "received a non-test packet"), - ProcessingError::NonMatchingNonce(nonce) => write!( - f, - "received packet with nonce {} which is different than the expected", - nonce - ), - ProcessingError::ReceivedOutsideTestRun => write!( - f, - "received packet while the test is currently not in progress" - ), - } - } -} - -// we can't use Notify due to possible edge case where both notification are consumed at once -enum LockPermit { - Release, - Free, -} - -struct ReceivedProcessorInner { - /// Nonce of the current test run indicating which packets should get rejected. - nonce: Option, - - /// Channel for receiving packets/messages from the gateway clients - packets_receiver: ReceivedProcessorReceiver, - - // TODO: right now it's identical for each gateway we send through, but should it? - /// Encryption key of the clients sending through the gateways. - client_encryption_keypair: Arc, - - /// Structure responsible for decrypting and recovering plaintext message from received ciphertexts. - message_receiver: MessageReceiver, - - /// Vector containing all received (and decrypted) packets in the current test run. - received_packets: Vec, -} - -impl ReceivedProcessorInner { - fn on_message(&mut self, message: Vec) -> Result<(), ProcessingError> { - // if the nonce is none it means the packet was received during the 'waiting' for the - // next test run - if self.nonce.is_none() { - return Err(ProcessingError::ReceivedOutsideTestRun); - } - - let encrypted_bytes = self - .message_receiver - .recover_plaintext(self.client_encryption_keypair.private_key(), message) - .map_err(|_| ProcessingError::MalformedPacketReceived)?; - let fragment = self - .message_receiver - .recover_fragment(&encrypted_bytes) - .map_err(|_| ProcessingError::MalformedPacketReceived)?; - let (recovered, _) = self - .message_receiver - .insert_new_fragment(fragment) - .map_err(|_| ProcessingError::MalformedPacketReceived)? - .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.message) - .map_err(|_| ProcessingError::MalformedPacketReceived)?; - - // we know nonce is NOT none - if test_packet.nonce() != self.nonce.unwrap() { - return Err(ProcessingError::NonMatchingNonce(test_packet.nonce())); - } - - self.received_packets.push(test_packet); - - Ok(()) - } - - fn finish_run(&mut self) -> Vec { - self.nonce = None; - mem::take(&mut self.received_packets) - } -} - -pub(crate) struct ReceivedProcessor { - permit_changer: mpsc::Sender, - inner: Arc>, -} - -impl ReceivedProcessor { - pub(crate) fn new( - packets_receiver: ReceivedProcessorReceiver, - client_encryption_keypair: Arc, - ) -> Self { - let inner: Arc> = - Arc::new(Mutex::new(ReceivedProcessorInner { - nonce: None, - packets_receiver, - client_encryption_keypair, - message_receiver: MessageReceiver::new(), - received_packets: Vec::new(), - })); - - // TODO: perhaps it should be using 0 size instead? - let (permit_sender, permit_receiver) = mpsc::channel(1); - - Self::start_receiving(Arc::clone(&inner), permit_receiver); - - ReceivedProcessor { - permit_changer: permit_sender, - inner, - } - } - - fn start_receiving( - inner: Arc>, - mut permit_change: mpsc::Receiver, - ) { - tokio::spawn(async move { - loop { - let permit = wait_for_permit(&mut permit_change, &*inner).await; - receive_or_release_permit(&mut permit_change, permit).await; - } - - async fn receive_or_release_permit( - permit_change: &mut mpsc::Receiver, - mut inner: MutexGuard<'_, ReceivedProcessorInner>, - ) { - loop { - tokio::select! { - permit_change = permit_change.next() => match permit_change.unwrap() { - LockPermit::Release => return, - LockPermit::Free => error!("somehow we got notification that the lock is free to take while we already hold it!"), - }, - messages = inner.packets_receiver.next() => { - for message in messages.expect("packet receiver has died!") { - if let Err(err) = inner.on_message(message) { - warn!(target: "Monitor", "failed to process received gateway message - {}", err) - } - } - }, - } - } - } - - // this lint really looks like a false positive because when lifetimes are elided, - // the compiler can't figure out appropriate lifetime bounds - #[allow(clippy::needless_lifetimes)] - async fn wait_for_permit<'a>( - permit_change: &mut mpsc::Receiver, - inner: &'a Mutex, - ) -> MutexGuard<'a, ReceivedProcessorInner> { - loop { - match permit_change.next().await.unwrap() { - // we should only ever get this on the very first run - LockPermit::Release => debug!( - "somehow got request to drop our lock permit while we do not hold it!" - ), - LockPermit::Free => return inner.lock().await, - } - } - } - }); - } - - pub(super) async fn set_new_expected(&mut self, nonce: u64) { - // ask for the lock back - self.permit_changer - .send(LockPermit::Release) - .await - .expect("processing task has died!"); - let mut inner = self.inner.lock().await; - - inner.nonce = Some(nonce); - - // give the permit back - drop(inner); - self.permit_changer - .send(LockPermit::Free) - .await - .expect("processing task has died!"); - } - - pub(super) async fn return_received(&mut self) -> Vec { - // ask for the lock back - self.permit_changer - .send(LockPermit::Release) - .await - .expect("processing task has died!"); - let mut inner = self.inner.lock().await; - - let received = inner.finish_run(); - - // give the permit back - drop(inner); - self.permit_changer - .send(LockPermit::Free) - .await - .expect("processing task has died!"); - - received - } -} diff --git a/network-monitor/src/monitor/receiver.rs b/network-monitor/src/monitor/receiver.rs deleted file mode 100644 index 230f4e7989..0000000000 --- a/network-monitor/src/monitor/receiver.rs +++ /dev/null @@ -1,69 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::gateways_reader::{GatewayChannel, GatewayMessages, GatewaysReader}; -use crate::monitor::processor::ReceivedProcessorSender; -use crypto::asymmetric::identity; -use futures::channel::mpsc; -use futures::StreamExt; -use gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver}; - -pub(crate) type GatewayClientUpdateSender = mpsc::UnboundedSender; -pub(crate) type GatewayClientUpdateReceiver = mpsc::UnboundedReceiver; - -pub(crate) enum GatewayClientUpdate { - Failure(identity::PublicKey), - New( - identity::PublicKey, - (MixnetMessageReceiver, AcknowledgementReceiver), - ), -} - -pub(crate) struct PacketReceiver { - gateways_reader: GatewaysReader, - clients_updater: GatewayClientUpdateReceiver, - processor_sender: ReceivedProcessorSender, -} - -impl PacketReceiver { - pub(crate) fn new( - clients_updater: GatewayClientUpdateReceiver, - processor_sender: ReceivedProcessorSender, - ) -> Self { - PacketReceiver { - gateways_reader: GatewaysReader::new(), - clients_updater, - processor_sender, - } - } - - fn process_gateway_update(&mut self, update: GatewayClientUpdate) { - match update { - GatewayClientUpdate::New(id, (message_receiver, ack_receiver)) => { - let channel = GatewayChannel::new(id, message_receiver, ack_receiver); - self.gateways_reader.insert_channel(channel); - } - GatewayClientUpdate::Failure(id) => self.gateways_reader.remove_by_key(id), - } - } - - fn process_gateway_messages(&self, messages: GatewayMessages) { - self.processor_sender - .unbounded_send(messages) - .expect("packet processor seems to have crashed!"); - } - - pub(crate) async fn run(&mut self) { - loop { - tokio::select! { - // 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()), - // similarly gateway reader will never return a `None` as it's implemented - // as an infinite stream that returns Poll::Pending if it doesn't have anything - // to return - messages = self.gateways_reader.next() => self.process_gateway_messages(messages.unwrap()), - } - } - } -} diff --git a/network-monitor/src/monitor/sender.rs b/network-monitor/src/monitor/sender.rs deleted file mode 100644 index 8c97b3a993..0000000000 --- a/network-monitor/src/monitor/sender.rs +++ /dev/null @@ -1,448 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::monitor::receiver::{GatewayClientUpdate, GatewayClientUpdateSender}; -use crate::{GATEWAY_CONNECTION_TIMEOUT, TIME_CHUNK_SIZE}; -use crypto::asymmetric::identity::{self, PUBLIC_KEY_LENGTH}; -use futures::channel::mpsc; -use futures::stream::{self, FuturesUnordered, StreamExt}; -use futures::task::Context; -use futures::{Future, Stream}; -use gateway_client::error::GatewayClientError; -use gateway_client::{AcknowledgementReceiver, GatewayClient, MixnetMessageReceiver}; -use log::*; -use nymsphinx::forwarding::packet::MixPacket; -use pin_project::pin_project; -use std::collections::HashMap; -use std::mem; -use std::num::NonZeroUsize; -use std::pin::Pin; -use std::sync::Arc; -use std::task::Poll; -use std::time::Duration; -use tokio::time::Instant; - -pub(crate) struct GatewayPackets { - /// Network address of the target gateway if wanted to be accessed by the client. - /// It is a websocket address. - clients_address: String, - - /// Public key of the target gateway. - pub_key: identity::PublicKey, - - /// All the packets that are going to get sent to the gateway. - packets: Vec, -} - -impl GatewayPackets { - pub(crate) fn new( - clients_address: String, - pub_key: identity::PublicKey, - packets: Vec, - ) -> Self { - GatewayPackets { - clients_address, - pub_key, - packets, - } - } - - pub(super) fn push_packets(&mut self, mut packets: Vec) { - self.packets.append(&mut packets) - } - - pub(super) fn gateway_address(&self) -> identity::PublicKey { - self.pub_key - } -} - -// struct consisting of all external data required to construct a fresh gateway client -struct FreshGatewayClientData { - gateways_status_updater: GatewayClientUpdateSender, - local_identity: Arc, - gateway_response_timeout: Duration, -} - -pub(crate) struct PacketSender { - // TODO: this has a potential long-term issue. If we keep those clients cached between runs, - // malicious gateways could figure out which traffic comes from the network monitor and always - // forward that traffic while dropping the rest. However, at the current stage such sophisticated - // behaviour is unlikely. - active_gateway_clients: HashMap<[u8; PUBLIC_KEY_LENGTH], GatewayClient>, - - fresh_gateway_client_data: Arc, - max_concurrent_clients: Option, - max_sending_rate: usize, -} - -impl PacketSender { - pub(crate) fn new( - gateways_status_updater: GatewayClientUpdateSender, - local_identity: Arc, - gateway_response_timeout: Duration, - max_concurrent_clients: Option, - max_sending_rate: usize, - ) -> Self { - PacketSender { - active_gateway_clients: HashMap::new(), - fresh_gateway_client_data: Arc::new(FreshGatewayClientData { - gateways_status_updater, - local_identity, - gateway_response_timeout, - }), - max_concurrent_clients, - max_sending_rate, - } - } - - fn new_gateway_client( - address: String, - identity: identity::PublicKey, - fresh_gateway_client_data: &FreshGatewayClientData, - ) -> ( - GatewayClient, - (MixnetMessageReceiver, AcknowledgementReceiver), - ) { - // TODO: future optimization: if we're remaking client for a gateway to which we used to be connected in the past, - // use old shared keys - let (message_sender, message_receiver) = mpsc::unbounded(); - // currently we do not care about acks at all, but we must keep the channel alive - // so that the gateway client would not crash - let (ack_sender, ack_receiver) = mpsc::unbounded(); - ( - GatewayClient::new( - address, - Arc::clone(&fresh_gateway_client_data.local_identity), - identity, - None, - message_sender, - ack_sender, - fresh_gateway_client_data.gateway_response_timeout, - ), - (message_receiver, ack_receiver), - ) - } - - async fn attempt_to_send_packets( - client: &mut GatewayClient, - mut mix_packets: Vec, - max_sending_rate: usize, - ) -> Result<(), GatewayClientError> { - let gateway_id = client.gateway_identity().to_base58_string(); - info!(target: "MessageSender", "Got {} packets to send to gateway {}", mix_packets.len(), gateway_id); - - if mix_packets.len() <= max_sending_rate { - debug!(target: "MessageSender", "Everything is going to get sent as one."); - client.batch_send_mix_packets(mix_packets).await?; - } else { - let packets_per_time_chunk = - (max_sending_rate as f64 * TIME_CHUNK_SIZE.as_secs_f64()) as usize; - - let total_expected_time = - Duration::from_secs_f64(mix_packets.len() as f64 / max_sending_rate as f64); - info!(target: "MessageSender", - "With our rate of {} packets/s it should take around {:?} to send it all to {} ...", - max_sending_rate, total_expected_time, gateway_id - ); - - fn split_off_vec(vec: &mut Vec, at: usize) -> Option> { - if vec.is_empty() { - None - } else { - if at >= vec.len() { - return Some(Vec::new()); - } - Some(vec.split_off(at)) - } - } - - // TODO future consideration: perhaps allow gateway client to take the packets by reference? - // this way we won't have to do reallocations in here as they're unavoidable when - // splitting a vector into multiple vectors - while let Some(retained) = split_off_vec(&mut mix_packets, packets_per_time_chunk) { - debug!(target: "MessageSender", "Sending {} packets...", mix_packets.len()); - - if mix_packets.len() == 1 { - client.send_mix_packet(mix_packets.pop().unwrap()).await?; - } else { - client.batch_send_mix_packets(mix_packets).await?; - } - - tokio::time::sleep(TIME_CHUNK_SIZE).await; - - mix_packets = retained; - } - debug!(target: "MessageSender", "Done sending"); - } - - Ok(()) - } - - // TODO: perhaps it should be spawned as a task to execute it in parallel rather - // than just concurrently? - async fn send_gateway_packets( - packets: GatewayPackets, - fresh_gateway_client_data: Arc, - client: Option, - max_sending_rate: usize, - ) -> Option { - let was_present = client.is_some(); - - let (mut client, gateway_channels) = if let Some(client) = client { - (client, None) - } else { - let (mut new_client, (message_receiver, ack_receiver)) = Self::new_gateway_client( - packets.clients_address, - packets.pub_key, - &fresh_gateway_client_data, - ); - - // Put this in timeout in case the gateway has incorrectly set their ulimit and our connection - // gets stuck in their TCP queue and just hangs on our end but does not terminate - // (an actual bug we experienced) - match tokio::time::timeout( - GATEWAY_CONNECTION_TIMEOUT, - new_client.authenticate_and_start(), - ) - .await - { - Ok(Ok(_)) => {} - Ok(Err(err)) => { - warn!( - "failed to authenticate with new gateway ({}) - {}", - packets.pub_key.to_base58_string(), - err - ); - // we failed to create a client, can't do much here - return None; - } - Err(_) => { - warn!( - "timed out while trying to authenticate with new gateway ({})", - packets.pub_key.to_base58_string() - ); - return None; - } - }; - - (new_client, Some((message_receiver, ack_receiver))) - }; - - if let Err(err) = - Self::attempt_to_send_packets(&mut client, packets.packets, max_sending_rate).await - { - warn!( - "failed to send packets to {} - {:?}", - packets.pub_key.to_base58_string(), - err - ); - // if this was a fresh client, there's no need to do anything as it was never - // registered to get read - if was_present { - fresh_gateway_client_data - .gateways_status_updater - .unbounded_send(GatewayClientUpdate::Failure(packets.pub_key)) - .expect("packet receiver seems to have died!"); - } - return None; - } else if !was_present { - // this is a fresh and working client - fresh_gateway_client_data - .gateways_status_updater - .unbounded_send(GatewayClientUpdate::New( - packets.pub_key, - gateway_channels - .expect("we created a new client, yet the channels are a None!"), - )) - .expect("packet receiver seems to have died!") - } - Some(client) - } - - pub(super) async fn send_packets(&mut self, packets: Vec) { - // we know that each of the elements in the packets array will only ever access a single, - // unique element from the existing clients - - // while it may seem weird that each time we send packets we remove the entries from the map, - // and then put them back in, this way we remove the need for having locks instead, like - // Arc>>> - let max_concurrent_clients = self.max_concurrent_clients; - let max_sending_rate = self.max_sending_rate; - - // can't chain it all nicely together as there's no adapter method defined on Stream directly - // for ForEachConcurrentClientUse - let stream = stream::iter(packets.into_iter().map(|packets| { - let existing_client = self - .active_gateway_clients - .remove(&(packets.pub_key.to_bytes())); - ( - packets, - Arc::clone(&self.fresh_gateway_client_data), - existing_client, - ) - })); - - ForEachConcurrentClientUse::new( - stream, - max_concurrent_clients, - |(packets, fresh_data, client)| async move { - Self::send_gateway_packets(packets, fresh_data, client, max_sending_rate).await - }, - ) - .await - .into_iter() - .for_each(|client| { - if let Some(client) = client { - if let Some(existing) = self - .active_gateway_clients - .insert(client.gateway_identity().to_bytes(), client) - { - panic!( - "we got duplicate gateway client for {}!", - existing.gateway_identity().to_base58_string() - ); - } - } - }) - } - - pub(super) async fn ping_all_active_gateways(&mut self) { - if self.active_gateway_clients.is_empty() { - info!(target: "Monitor", "no gateways to ping"); - return; - } - - let ping_start = Instant::now(); - - let mut clients_to_purge = Vec::new(); - - // since we don't need to wait for response, we can just ping all gateways sequentially - // if it becomes problem later on, we can adjust it. - for (gateway_id, active_client) in self.active_gateway_clients.iter_mut() { - if let Err(err) = active_client.send_ping_message().await { - warn!( - target: "Monitor", - "failed to send ping message to gateway {} - {} - assuming the connection is dead.", - active_client.gateway_identity().to_base58_string(), - err, - ); - clients_to_purge.push(*gateway_id); - } - } - - // purge all dead connections - for gateway_id in clients_to_purge.into_iter() { - // if this unwrap failed it means something extremely weird is going on - // and we got some solar flare bitflip type of corruption - let gateway_key = identity::PublicKey::from_bytes(&gateway_id) - .expect("failed to recover gateways public key from valid bytes"); - - // remove the gateway listener channels - self.fresh_gateway_client_data - .gateways_status_updater - .unbounded_send(GatewayClientUpdate::Failure(gateway_key)) - .expect("packet receiver seems to have died!"); - - // and remove it from our cache - self.active_gateway_clients.remove(&gateway_id); - } - - let ping_end = Instant::now(); - let time_taken = ping_end.duration_since(ping_start); - debug!(target: "Monitor", "pinging all active gateways took {:?}", time_taken); - } -} - -// A slightly modified and less generic version of the futures' ForEachConcurrent that allows the futures to return -// gateway clients back -#[pin_project] -struct ForEachConcurrentClientUse { - #[pin] - stream: Option, - f: F, - futures: FuturesUnordered, - limit: Option, - result: Vec>, -} - -impl ForEachConcurrentClientUse -where - St: Stream, - F: FnMut(St::Item) -> Fut, - Fut: Future>, -{ - pub(super) fn new(stream: St, limit: Option, f: F) -> Self { - let size_hint = stream.size_hint(); - Self { - stream: Some(stream), - // Note: `limit` = 0 gets ignored. - limit: limit.and_then(NonZeroUsize::new), - f, - futures: FuturesUnordered::new(), - result: Vec::with_capacity(size_hint.1.unwrap_or(size_hint.0)), - } - } -} - -impl Future for ForEachConcurrentClientUse -where - St: Stream, - F: FnMut(St::Item) -> Fut, - Fut: Future>, -{ - type Output = Vec>; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let mut this = self.project(); - loop { - let mut made_progress_this_iter = false; - - // Check if we've already created a number of futures greater than `limit` - if this - .limit - .map(|limit| limit.get() > this.futures.len()) - .unwrap_or(true) - { - let mut stream_completed = false; - let elem = if let Some(stream) = this.stream.as_mut().as_pin_mut() { - match stream.poll_next(cx) { - Poll::Ready(Some(elem)) => { - made_progress_this_iter = true; - Some(elem) - } - Poll::Ready(None) => { - stream_completed = true; - None - } - Poll::Pending => None, - } - } else { - None - }; - if stream_completed { - this.stream.set(None); - } - if let Some(elem) = elem { - this.futures.push((this.f)(elem)); - } - } - - match this.futures.poll_next_unpin(cx) { - Poll::Ready(Some(client)) => { - this.result.push(client); - made_progress_this_iter = true - } - Poll::Ready(None) => { - if this.stream.is_none() { - return Poll::Ready(mem::take(this.result)); - } - } - Poll::Pending => {} - } - - if !made_progress_this_iter { - return Poll::Pending; - } - } - } -} diff --git a/network-monitor/src/monitor/summary_producer.rs b/network-monitor/src/monitor/summary_producer.rs deleted file mode 100644 index ca4a4c575e..0000000000 --- a/network-monitor/src/monitor/summary_producer.rs +++ /dev/null @@ -1,270 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::monitor::preparer::{InvalidNode, TestedNode}; -use crate::node_status_api::models::{ - BatchGatewayStatus, BatchMixStatus, GatewayStatus, MixStatus, -}; -use crate::test_packet::{NodeType, TestPacket}; -use crate::PENALISE_OUTDATED; -use log::*; -use std::collections::HashMap; - -#[derive(Default)] -struct NodeResult { - ip_v4_compatible: bool, - ip_v6_compatible: bool, -} - -impl NodeResult { - fn into_mix_status(self, pub_key: String, owner: String) -> Vec { - let v4_status = MixStatus { - owner: owner.clone(), - pub_key: pub_key.clone(), - ip_version: "4".to_string(), - up: self.ip_v4_compatible, - }; - - let v6_status = MixStatus { - owner, - pub_key, - ip_version: "6".to_string(), - up: self.ip_v6_compatible, - }; - - vec![v4_status, v6_status] - } - - fn into_gateway_status(self, pub_key: String, owner: String) -> Vec { - let v4_status = GatewayStatus { - owner: owner.clone(), - pub_key: pub_key.clone(), - ip_version: "4".to_string(), - up: self.ip_v4_compatible, - }; - - let v6_status = GatewayStatus { - owner, - pub_key, - ip_version: "6".to_string(), - up: self.ip_v6_compatible, - }; - - vec![v4_status, v6_status] - } -} - -#[derive(Default)] -pub(crate) struct TestReport { - pub(crate) total_sent: usize, - pub(crate) total_received: usize, - pub(crate) malformed: Vec, - - // below are only populated if we're going to be printing the report - pub(crate) only_ipv4_compatible_mixes: Vec, // can't speak v6, but can speak v4 - pub(crate) only_ipv6_compatible_mixes: Vec, // can't speak v4, but can speak v6 - pub(crate) completely_unroutable_mixes: Vec, // can't speak either v4 or v6 - pub(crate) fully_working_mixes: Vec, - - pub(crate) only_ipv4_compatible_gateways: Vec, // can't speak v6, but can speak v4 - pub(crate) only_ipv6_compatible_gateways: Vec, // can't speak v4, but can speak v6 - pub(crate) completely_unroutable_gateways: Vec, // can't speak either v4 or v6 - pub(crate) fully_working_gateways: Vec, -} - -impl TestReport { - fn print(&self, detailed: bool) { - info!(target: "Test Report", "Sent total of {} packets", self.total_sent); - info!(target: "Test Report", "Received total of {} packets", self.total_received); - info!(target: "Test Report", "{} nodes are invalid", self.malformed.len()); - - info!(target: "Test Report", "{} mixnodes speak ONLY IPv4 (NO IPv6 connectivity)", self.only_ipv4_compatible_mixes.len()); - info!(target: "Test Report", "{} mixnodes speak ONLY IPv6 (NO IPv4 connectivity)", self.only_ipv6_compatible_mixes.len()); - info!(target: "Test Report", "{} mixnodes are totally unroutable!", self.completely_unroutable_mixes.len()); - info!(target: "Test Report", "{} mixnodes work fine!", self.fully_working_mixes.len()); - - info!(target: "Test Report", "{} gateways speak ONLY IPv4 (NO IPv6 connectivity)", self.only_ipv4_compatible_gateways.len()); - info!(target: "Test Report", "{} gateways speak ONLY IPv6 (NO IPv4 connectivity)", self.only_ipv6_compatible_gateways.len()); - info!(target: "Test Report", "{} gateways are totally unroutable!", self.completely_unroutable_gateways.len()); - info!(target: "Test Report", "{} gateways work fine!", self.fully_working_gateways.len()); - - if detailed { - info!(target: "Detailed report", "full summary:"); - for malformed in self.malformed.iter() { - info!(target: "Invalid node", "{}", malformed) - } - - for v4_node in self.only_ipv4_compatible_mixes.iter() { - info!(target: "IPv4-only mixnode", "{}", v4_node) - } - - for v6_node in self.only_ipv6_compatible_mixes.iter() { - info!(target: "IPv6-only mixnode", "{}", v6_node) - } - - for unroutable in self.completely_unroutable_mixes.iter() { - info!(target: "Unroutable mixnode", "{}", unroutable) - } - - for working in self.fully_working_mixes.iter() { - info!(target: "Fully working mixnode", "{}", working) - } - - for v4_node in self.only_ipv4_compatible_gateways.iter() { - info!(target: "IPv4-only gateway", "{}", v4_node) - } - - for v6_node in self.only_ipv6_compatible_gateways.iter() { - info!(target: "IPv6-only gateway", "{}", v6_node) - } - - for unroutable in self.completely_unroutable_gateways.iter() { - info!(target: "Unroutable gateway", "{}", unroutable) - } - - for working in self.fully_working_gateways.iter() { - info!(target: "Fully working gateway", "{}", working) - } - } - } - - fn parse_summary(&mut self, summary: &HashMap) { - for (node, result) in summary.iter() { - let owned_node = node.clone(); - if node.is_gateway() { - if result.ip_v4_compatible && result.ip_v6_compatible { - self.fully_working_gateways.push(owned_node) - } else if result.ip_v4_compatible { - self.only_ipv4_compatible_gateways.push(owned_node) - } else if result.ip_v6_compatible { - self.only_ipv6_compatible_gateways.push(owned_node) - } else { - self.completely_unroutable_gateways.push(owned_node) - } - } else if result.ip_v4_compatible && result.ip_v6_compatible { - self.fully_working_mixes.push(owned_node) - } else if result.ip_v4_compatible { - self.only_ipv4_compatible_mixes.push(owned_node) - } else if result.ip_v6_compatible { - self.only_ipv6_compatible_mixes.push(owned_node) - } else { - self.completely_unroutable_mixes.push(owned_node) - } - } - } -} - -pub(crate) struct TestSummary { - pub(crate) batch_mix_status: BatchMixStatus, - pub(crate) batch_gateway_status: BatchGatewayStatus, - pub(crate) test_report: TestReport, -} - -#[derive(Default)] -pub(crate) struct SummaryProducer { - print_report: bool, - print_detailed_report: bool, -} - -impl SummaryProducer { - pub(crate) fn with_report(mut self) -> Self { - self.print_report = true; - self - } - - pub(crate) fn with_detailed_report(mut self) -> Self { - self.print_report = true; - self.print_detailed_report = true; - self - } - - pub(super) fn produce_summary( - &self, - expected_nodes: Vec, - received_packets: Vec, - invalid_nodes: Vec, - ) -> TestSummary { - let expected_nodes_count = expected_nodes.len(); - let received_packets_count = received_packets.len(); - - // contains map of all (seemingly valid) nodes and whether they speak ipv4/ipv6 - let mut summary: HashMap = HashMap::new(); - - // update based on data we actually got - for received_status in received_packets.into_iter() { - let is_received_v4 = received_status.ip_version().is_v4(); - let entry = summary.entry(received_status.into()).or_default(); - if is_received_v4 { - entry.ip_v4_compatible = true - } else { - entry.ip_v6_compatible = true - } - } - - // insert entries we didn't get but were expecting - for expected in expected_nodes.into_iter() { - summary.entry(expected).or_default(); - } - - // finally insert malformed nodes - for malformed in invalid_nodes.iter() { - match malformed { - InvalidNode::OutdatedMix(id, owner, _) - | InvalidNode::OutdatedGateway(id, owner, _) => { - if PENALISE_OUTDATED { - summary.insert(TestedNode::from_raw_gateway(id, owner), Default::default()); - } - } - InvalidNode::MalformedMix(id, owner) | InvalidNode::MalformedGateway(id, owner) => { - summary.insert(TestedNode::from_raw_mix(id, owner), Default::default()); - } - } - } - - let mut report = TestReport { - total_sent: expected_nodes_count * 2, // we sent two packets per node (one ipv4 and one ipv6) - total_received: received_packets_count, - malformed: invalid_nodes, - - ..Default::default() - }; - - report.parse_summary(&summary); - - if self.print_report { - report.print(self.print_detailed_report); - } - - let (mixes, gateways): (Vec<_>, Vec<_>) = summary - .into_iter() - .partition(|(node, _)| node.node_type == NodeType::Mixnode); - - let mix_statuses = mixes - .into_iter() - .flat_map(|(node, result)| { - result - .into_mix_status(node.identity, node.owner) - .into_iter() - }) - .collect(); - - let gateway_statuses = gateways - .into_iter() - .flat_map(|(node, result)| { - result - .into_gateway_status(node.identity, node.owner) - .into_iter() - }) - .collect(); - - TestSummary { - batch_mix_status: BatchMixStatus { - status: mix_statuses, - }, - batch_gateway_status: BatchGatewayStatus { - status: gateway_statuses, - }, - test_report: report, - } - } -} diff --git a/network-monitor/src/node_status_api/client.rs b/network-monitor/src/node_status_api/client.rs deleted file mode 100644 index b7698f6c07..0000000000 --- a/network-monitor/src/node_status_api/client.rs +++ /dev/null @@ -1,110 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::node_status_api::models::{BatchGatewayStatus, BatchMixStatus, DefaultRestResponse}; -use crate::node_status_api::NodeStatusApiClientError; - -pub(crate) struct Config { - base_url: String, -} - -impl Config { - pub(crate) fn new>(base_url: S) -> Self { - Config { - base_url: base_url.into(), - } - } -} - -pub(crate) struct Client { - config: Config, - reqwest_client: reqwest::Client, -} - -impl Client { - pub(crate) fn new(config: Config) -> Self { - let reqwest_client = reqwest::Client::new(); - Client { - config, - reqwest_client, - } - } - - // Potentially, down the line, this could be moved to /common/client-libs - // and additional methods could be added like GET for report data, but currently - // we have absolutely no use for that in Rust. - - pub(crate) async fn post_batch_mix_status( - &self, - batch_status: BatchMixStatus, - ) -> Result<(), NodeStatusApiClientError> { - const RELATIVE_PATH: &str = "api/status/mixnode/batch"; - - let url = format!("{}/{}", self.config.base_url, RELATIVE_PATH); - - let response = self - .reqwest_client - .post(url) - .json(&batch_status) - .send() - .await?; - - if response.status().is_success() { - let response_content: DefaultRestResponse = response.json().await?; - match response_content { - DefaultRestResponse::Ok(ok_response) => { - if ok_response.ok { - Ok(()) - } else { - Err(NodeStatusApiClientError::NodeStatusApiError( - "received an ok response with false status".into(), - )) - } - } - DefaultRestResponse::Error(err_response) => Err(err_response.into()), - } - } else { - Err(NodeStatusApiClientError::NodeStatusApiError(format!( - "received response with status {}", - response.status() - ))) - } - } - - pub(crate) async fn post_batch_gateway_status( - &self, - batch_status: BatchGatewayStatus, - ) -> Result<(), NodeStatusApiClientError> { - const RELATIVE_PATH: &str = "api/status/gateway/batch"; - - let url = format!("{}/{}", self.config.base_url, RELATIVE_PATH); - - let response = self - .reqwest_client - .post(url) - .json(&batch_status) - .send() - .await?; - - if response.status().is_success() { - let response_content: DefaultRestResponse = response.json().await?; - match response_content { - DefaultRestResponse::Ok(ok_response) => { - if ok_response.ok { - Ok(()) - } else { - Err(NodeStatusApiClientError::NodeStatusApiError( - "received an ok response with false status".into(), - )) - } - } - DefaultRestResponse::Error(err_response) => Err(err_response.into()), - } - } else { - Err(NodeStatusApiClientError::NodeStatusApiError(format!( - "received response with status {}", - response.status() - ))) - } - } -} diff --git a/network-monitor/src/node_status_api/mod.rs b/network-monitor/src/node_status_api/mod.rs deleted file mode 100644 index 6570a3ebc1..0000000000 --- a/network-monitor/src/node_status_api/mod.rs +++ /dev/null @@ -1,73 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::node_status_api::models::ErrorResponses; -use std::fmt::{self, Display, Formatter}; - -mod client; -pub(crate) mod models; - -pub(crate) use client::{Client, Config}; - -const MAX_SANE_UNEXPECTED_PRINT: usize = 100; - -#[derive(Debug)] -pub enum NodeStatusApiClientError { - ReqwestClientError(reqwest::Error), - NodeStatusApiError(String), - UnexpectedResponse(String), -} - -impl From for NodeStatusApiClientError { - fn from(err: reqwest::Error) -> Self { - NodeStatusApiClientError::ReqwestClientError(err) - } -} - -impl From for NodeStatusApiClientError { - fn from(err: ErrorResponses) -> Self { - match err { - ErrorResponses::Error(err_message) => { - NodeStatusApiClientError::NodeStatusApiError(err_message.error) - } - ErrorResponses::Unexpected(received) => { - NodeStatusApiClientError::UnexpectedResponse(received.to_string()) - } - } - } -} - -impl Display for NodeStatusApiClientError { - fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - match self { - NodeStatusApiClientError::ReqwestClientError(err) => { - write!(f, "there was an issue with the REST request - {}", err) - } - NodeStatusApiClientError::NodeStatusApiError(err) => { - write!( - f, - "there was an issue with the node status api client - {}", - err - ) - } - NodeStatusApiClientError::UnexpectedResponse(received) => { - if received.len() < MAX_SANE_UNEXPECTED_PRINT { - write!( - f, - "received data was completely unexpected. got: {}", - received - ) - } else { - write!( - f, - "received data was completely unexpected. got: {}...", - received - .chars() - .take(MAX_SANE_UNEXPECTED_PRINT) - .collect::() - ) - } - } - } - } -} diff --git a/network-monitor/src/node_status_api/models.rs b/network-monitor/src/node_status_api/models.rs deleted file mode 100644 index 93e39ec470..0000000000 --- a/network-monitor/src/node_status_api/models.rs +++ /dev/null @@ -1,68 +0,0 @@ -// Copyright 2021 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use serde::{Deserialize, Serialize}; - -#[derive(Debug, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -/// A notification sent to the validators to let them know whether a given mix is -/// currently up or down (based on whether it's mixing packets) -pub struct MixStatus { - pub pub_key: String, - pub owner: String, - pub ip_version: String, - pub up: bool, -} - -#[derive(Debug, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -/// A notification sent to the validators to let them know whether a given set of mixes is -/// currently up or down (based on whether it's mixing packets) -pub struct BatchMixStatus { - pub status: Vec, -} - -#[derive(Debug, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -/// A notification sent to the validators to let them know whether a given gateway is -/// currently up or down (based on whether it's mixing packets) -pub struct GatewayStatus { - pub pub_key: String, - pub owner: String, - pub ip_version: String, - pub up: bool, -} - -#[derive(Debug, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -/// A notification sent to the validators to let them know whether a given set of gateways is -/// currently up or down (based on whether it's mixing packets) -pub struct BatchGatewayStatus { - pub status: Vec, -} - -#[derive(Deserialize, Debug)] -#[serde(rename_all = "camelCase", untagged)] -pub(crate) enum ErrorResponses { - Error(ErrorResponse), - Unexpected(serde_json::Value), -} - -#[derive(Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub(crate) struct ErrorResponse { - pub(crate) error: String, -} - -#[derive(Deserialize, Debug)] -#[serde(rename_all = "camelCase")] -pub(crate) struct OkResponse { - pub(crate) ok: bool, -} - -#[derive(Deserialize, Debug)] -#[serde(rename_all = "camelCase", untagged)] -pub(crate) enum DefaultRestResponse { - Ok(OkResponse), - Error(ErrorResponses), -} diff --git a/network-monitor/src/test_packet.rs b/network-monitor/src/test_packet.rs deleted file mode 100644 index 08ef1bcad7..0000000000 --- a/network-monitor/src/test_packet.rs +++ /dev/null @@ -1,214 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::monitor::preparer::TestedNode; -use crypto::asymmetric::identity; -use std::convert::{TryFrom, TryInto}; -use std::fmt::{self, Display, Formatter}; -use std::hash::{Hash, Hasher}; -use std::mem; -use std::str::Utf8Error; - -#[repr(u8)] -#[derive(Eq, PartialEq, Debug, Hash, Clone, Copy)] -pub(crate) enum NodeType { - Mixnode = 0, - Gateway = 1, -} - -impl TryFrom for NodeType { - type Error = TestPacketError; - - fn try_from(value: u8) -> Result { - match value { - _ if value == (Self::Mixnode as u8) => Ok(Self::Mixnode), - _ if value == (Self::Gateway as u8) => Ok(Self::Gateway), - _ => Err(TestPacketError::InvalidNodeType), - } - } -} - -#[derive(Debug)] -pub(crate) enum TestPacketError { - IncompletePacket, - InvalidIpVersion, - InvalidNodeType, - InvalidNodeKey, - InvalidOwner(Utf8Error), -} - -impl From for TestPacketError { - fn from(_: identity::KeyRecoveryError) -> Self { - TestPacketError::InvalidNodeKey - } -} - -impl From for TestPacketError { - fn from(err: Utf8Error) -> Self { - TestPacketError::InvalidOwner(err) - } -} - -#[repr(u8)] -#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)] -pub(crate) enum IpVersion { - V4 = 4, - V6 = 6, -} - -impl TryFrom for IpVersion { - type Error = TestPacketError; - - fn try_from(value: u8) -> Result { - match value { - _ if value == (Self::V4 as u8) => Ok(Self::V4), - _ if value == (Self::V6 as u8) => Ok(Self::V6), - _ => Err(TestPacketError::InvalidIpVersion), - } - } -} - -impl IpVersion { - pub(crate) fn is_v4(&self) -> bool { - *self == IpVersion::V4 - } -} - -impl From for String { - fn from(ipv: IpVersion) -> Self { - format!("{}", ipv) - } -} - -impl Display for IpVersion { - fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - write!(f, "{}", *self as u8) - } -} - -#[derive(Eq, Clone, Debug)] -pub(crate) struct TestPacket { - ip_version: IpVersion, - nonce: u64, - pub_key: identity::PublicKey, - owner: String, - node_type: NodeType, -} - -impl Display for TestPacket { - fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - write!( - f, - "TestPacket {{ ip: {}, pub_key: {}, owner: {}, nonce: {} }}", - self.ip_version, - self.pub_key.to_base58_string(), - self.owner, - self.nonce - ) - } -} - -impl Hash for TestPacket { - fn hash(&self, state: &mut H) { - self.ip_version.hash(state); - self.nonce.hash(state); - self.pub_key.to_bytes().hash(state); - self.owner.hash(state); - } -} - -impl PartialEq for TestPacket { - fn eq(&self, other: &Self) -> bool { - self.ip_version == other.ip_version - && self.nonce == other.nonce - && self.pub_key.to_bytes() == other.pub_key.to_bytes() - } -} - -impl TestPacket { - pub(crate) fn new_v4( - pub_key: identity::PublicKey, - owner: String, - nonce: u64, - node_type: NodeType, - ) -> Self { - TestPacket { - ip_version: IpVersion::V4, - nonce, - pub_key, - owner, - node_type, - } - } - - pub(crate) fn new_v6( - pub_key: identity::PublicKey, - owner: String, - nonce: u64, - node_type: NodeType, - ) -> Self { - TestPacket { - ip_version: IpVersion::V6, - nonce, - pub_key, - owner, - node_type, - } - } - - pub(crate) fn nonce(&self) -> u64 { - self.nonce - } - - pub(crate) fn ip_version(&self) -> IpVersion { - self.ip_version - } - - pub(crate) fn to_bytes(&self) -> Vec { - self.nonce - .to_be_bytes() - .iter() - .cloned() - .chain(std::iter::once(self.node_type as u8)) - .chain(std::iter::once(self.ip_version as u8)) - .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 { - // nonce size - let n = mem::size_of::(); - - if b.len() < n + 1 + identity::PUBLIC_KEY_LENGTH { - return Err(TestPacketError::IncompletePacket); - } - - // this unwrap can't fail as we've already checked for the size - let nonce = u64::from_be_bytes(b[0..n].try_into().unwrap()); - let node_type = NodeType::try_from(b[n])?; - - let ip_version = IpVersion::try_from(b[n + 1])?; - let pub_key = - identity::PublicKey::from_bytes(&b[n + 2..n + 2 + identity::PUBLIC_KEY_LENGTH])?; - let owner = std::str::from_utf8(&b[n + 2 + identity::PUBLIC_KEY_LENGTH..])?; - - Ok(TestPacket { - node_type, - ip_version, - nonce, - pub_key, - owner: owner.to_owned(), - }) - } -} - -impl From for TestedNode { - fn from(packet: TestPacket) -> Self { - TestedNode { - identity: packet.pub_key.to_base58_string(), - owner: packet.owner, - node_type: packet.node_type, - } - } -} diff --git a/network-monitor/src/tested_network/good_topology.rs b/network-monitor/src/tested_network/good_topology.rs deleted file mode 100644 index 3fa91c8e0b..0000000000 --- a/network-monitor/src/tested_network/good_topology.rs +++ /dev/null @@ -1,29 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use mixnet_contract::{GatewayBond, MixNodeBond}; -use serde::Deserialize; -use std::fs; -use topology::{nym_topology_from_bonds, NymTopology}; - -#[derive(Deserialize)] -struct GoodTopology { - mixnodes: Vec, - gateways: Vec, -} - -pub(crate) fn parse_topology_file(file_path: &str) -> NymTopology { - let file_content = - fs::read_to_string(file_path).expect("specified topology file does not exist"); - let good_topology = serde_json::from_str::(&file_content) - .expect("topology in specified file is malformed"); - let nym_topology = nym_topology_from_bonds(good_topology.mixnodes, good_topology.gateways); - if nym_topology.mixes().len() != 3 { - panic!("topology has different than 3 number of layers") - } - if nym_topology.gateways().is_empty() { - panic!("topology does not include a gateway") - } - - nym_topology -} diff --git a/network-monitor/src/tested_network/mod.rs b/network-monitor/src/tested_network/mod.rs deleted file mode 100644 index 85cac62a66..0000000000 --- a/network-monitor/src/tested_network/mod.rs +++ /dev/null @@ -1,72 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use crate::test_packet::IpVersion; -use log::*; -use topology::{gateway, mix, NymTopology}; - -pub(crate) mod good_topology; - -#[derive(Clone)] -pub(crate) struct TestedNetwork { - system_version: String, - good_v4_topology: NymTopology, - good_v6_topology: NymTopology, -} - -impl TestedNetwork { - pub(crate) fn new_good(good_v4_topology: NymTopology, good_v6_topology: NymTopology) -> Self { - TestedNetwork { - system_version: good_v4_topology.mixes()[&1][0].version.clone(), - good_v4_topology, - good_v6_topology, - } - } - - pub(crate) fn main_v4_gateway(&self) -> &gateway::Node { - if self.good_v4_topology.gateways().len() > 1 { - warn!("we have more than a single 'good' gateway and in few places we made assumptions that only a single one existed!") - } - - self.good_v4_topology - .gateways() - .get(0) - .expect("our good v4 topology does not have any gateway specified!") - } - - pub(crate) fn system_version(&self) -> &str { - &self.system_version - } - - pub(crate) fn substitute_mix(&self, node: mix::Node, ip_version: IpVersion) -> NymTopology { - let mut good_topology = match ip_version { - IpVersion::V4 => self.good_v4_topology.clone(), - IpVersion::V6 => self.good_v6_topology.clone(), - }; - - good_topology.set_mixes_in_layer(node.layer as u8, vec![node]); - good_topology - } - - pub(crate) fn substitute_gateway( - &self, - gateway: gateway::Node, - ip_version: IpVersion, - ) -> NymTopology { - let mut good_topology = match ip_version { - IpVersion::V4 => self.good_v4_topology.clone(), - IpVersion::V6 => self.good_v6_topology.clone(), - }; - - good_topology.set_gateways(vec![gateway]); - good_topology - } - - pub(crate) fn v4_topology(&self) -> &NymTopology { - &self.good_v4_topology - } - - pub(crate) fn v6_topology(&self) -> &NymTopology { - &self.good_v6_topology - } -}