diff --git a/Cargo.lock b/Cargo.lock index 65d62946b3..c333bb8925 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1798,31 +1798,6 @@ dependencies = [ "version-checker", ] -[[package]] -name = "nym-network-monitor" -version = "0.10.1" -dependencies = [ - "clap", - "crypto", - "dotenv", - "futures", - "gateway-client", - "gateway-requests", - "log", - "mixnet-contract", - "nymsphinx", - "pin-project", - "pretty_env_logger", - "rand 0.7.3", - "reqwest", - "serde", - "serde_json", - "tokio", - "topology", - "validator-client", - "version-checker", -] - [[package]] name = "nym-network-requester" version = "0.10.1" @@ -1874,6 +1849,31 @@ dependencies = [ "version-checker", ] +[[package]] +name = "nym-validator-api" +version = "0.10.1" +dependencies = [ + "clap", + "crypto", + "dotenv", + "futures", + "gateway-client", + "gateway-requests", + "log", + "mixnet-contract", + "nymsphinx", + "pin-project", + "pretty_env_logger", + "rand 0.7.3", + "reqwest", + "serde", + "serde_json", + "tokio", + "topology", + "validator-client", + "version-checker", +] + [[package]] name = "nymsphinx" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 2fc54859cd..1f8a0a5e77 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,7 +42,7 @@ members = [ "gateway", "gateway/gateway-requests", "mixnode", - "network-monitor", + "validator-api", "service-providers/network-requester", ] @@ -53,7 +53,7 @@ default-members = [ "gateway", "service-providers/network-requester", "mixnode", - "network-monitor", + "validator-api", ] exclude = ["explorer", "contracts"] diff --git a/network-monitor/Cargo.toml b/network-monitor/Cargo.toml index 704b05f295..ef7b52c305 100644 --- a/network-monitor/Cargo.toml +++ b/network-monitor/Cargo.toml @@ -2,7 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 [package] -name = "nym-network-monitor" +name = "nym-validator-api" version = "0.10.1" authors = ["Dave Hrycyszyn ", "Jędrzej Stuczyński "] edition = "2018" diff --git a/validator-api/Cargo.toml b/validator-api/Cargo.toml new file mode 100644 index 0000000000..ef7b52c305 --- /dev/null +++ b/validator-api/Cargo.toml @@ -0,0 +1,35 @@ +# 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/validator-api/src/chunker.rs b/validator-api/src/chunker.rs new file mode 100644 index 0000000000..b2bc7c1736 --- /dev/null +++ b/validator-api/src/chunker.rs @@ -0,0 +1,75 @@ +// 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/validator-api/src/gateways_reader.rs b/validator-api/src/gateways_reader.rs new file mode 100644 index 0000000000..e4288caec4 --- /dev/null +++ b/validator-api/src/gateways_reader.rs @@ -0,0 +1,183 @@ +// 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/validator-api/src/main.rs b/validator-api/src/main.rs new file mode 100644 index 0000000000..1813ebecfa --- /dev/null +++ b/validator-api/src/main.rs @@ -0,0 +1,356 @@ +// 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/validator-api/src/monitor/mod.rs b/validator-api/src/monitor/mod.rs new file mode 100644 index 0000000000..a3c1a6ca96 --- /dev/null +++ b/validator-api/src/monitor/mod.rs @@ -0,0 +1,220 @@ +// 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/validator-api/src/monitor/preparer.rs b/validator-api/src/monitor/preparer.rs new file mode 100644 index 0000000000..a1aba7b924 --- /dev/null +++ b/validator-api/src/monitor/preparer.rs @@ -0,0 +1,477 @@ +// 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/validator-api/src/monitor/processor.rs b/validator-api/src/monitor/processor.rs new file mode 100644 index 0000000000..f2e9da15ee --- /dev/null +++ b/validator-api/src/monitor/processor.rs @@ -0,0 +1,227 @@ +// 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/validator-api/src/monitor/receiver.rs b/validator-api/src/monitor/receiver.rs new file mode 100644 index 0000000000..230f4e7989 --- /dev/null +++ b/validator-api/src/monitor/receiver.rs @@ -0,0 +1,69 @@ +// 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/validator-api/src/monitor/sender.rs b/validator-api/src/monitor/sender.rs new file mode 100644 index 0000000000..8c97b3a993 --- /dev/null +++ b/validator-api/src/monitor/sender.rs @@ -0,0 +1,448 @@ +// 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/validator-api/src/monitor/summary_producer.rs b/validator-api/src/monitor/summary_producer.rs new file mode 100644 index 0000000000..ca4a4c575e --- /dev/null +++ b/validator-api/src/monitor/summary_producer.rs @@ -0,0 +1,270 @@ +// 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/validator-api/src/node_status_api/client.rs b/validator-api/src/node_status_api/client.rs new file mode 100644 index 0000000000..b7698f6c07 --- /dev/null +++ b/validator-api/src/node_status_api/client.rs @@ -0,0 +1,110 @@ +// 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/validator-api/src/node_status_api/mod.rs b/validator-api/src/node_status_api/mod.rs new file mode 100644 index 0000000000..6570a3ebc1 --- /dev/null +++ b/validator-api/src/node_status_api/mod.rs @@ -0,0 +1,73 @@ +// 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/validator-api/src/node_status_api/models.rs b/validator-api/src/node_status_api/models.rs new file mode 100644 index 0000000000..93e39ec470 --- /dev/null +++ b/validator-api/src/node_status_api/models.rs @@ -0,0 +1,68 @@ +// 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/validator-api/src/test_packet.rs b/validator-api/src/test_packet.rs new file mode 100644 index 0000000000..08ef1bcad7 --- /dev/null +++ b/validator-api/src/test_packet.rs @@ -0,0 +1,214 @@ +// 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/validator-api/src/tested_network/good_topology.rs b/validator-api/src/tested_network/good_topology.rs new file mode 100644 index 0000000000..3fa91c8e0b --- /dev/null +++ b/validator-api/src/tested_network/good_topology.rs @@ -0,0 +1,29 @@ +// 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/validator-api/src/tested_network/mod.rs b/validator-api/src/tested_network/mod.rs new file mode 100644 index 0000000000..abcf9026ba --- /dev/null +++ b/validator-api/src/tested_network/mod.rs @@ -0,0 +1,72 @@ +// 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: "0.10.0".to_string(), + 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 + } +} diff --git a/validator-api/v4.json b/validator-api/v4.json new file mode 100644 index 0000000000..4a67eecacd --- /dev/null +++ b/validator-api/v4.json @@ -0,0 +1,87 @@ +{ + "mixnodes": [ + { + "amount": [ + { + "denom": "uhal", + "amount": "100000000" + } + ], + "owner": "hal1c8t5cj5sgjt474y3wr04m6h3648yg7se0dkff2", + "mix_node": { + "host": "178.79.174.233", + "mix_port": 1789, + "verloc_port": 1790, + "http_api_port": 8000, + "clients_port": 9000, + "layer": 1, + "location": "London, UK", + "sphinx_key": "73V9Hdo79YxAUwUgAN6qgkhzcDJGymcGnGF6tFseaWFT", + "identity_key": "824WyExLUWvLE2mpSHBatN4AoByuLzfnHFeHWiBYzg4z", + "version": "0.10.0" + } + }, + { + "amount": [ + { + "denom": "uhal", + "amount": "100000000" + } + ], + "owner": "hal1yz3ac9382la00l4srczdtma0vrg5p7a72kky6t", + "mix_node": { + "host": "178.79.172.191", + "mix_port": 1789, + "verloc_port": 1790, + "http_api_port": 8000, + "clients_port": 9000, + "layer": 2, + "location": "London, UK", + "sphinx_key": "7eAfV7CgzGzbKU9QKYyUFKmvsCgPC1NMUF2U16Guezsn", + "identity_key": "7NBLS5M7QxFqUZkBZn2RWe3kkM8uH3MT2mtK4UPP2q7d", + "version": "0.10.0" + } + }, + { + "amount": [ + { + "denom": "uhal", + "amount": "100000000" + } + ], + "owner": "hal1rz7sqd90pjj3a6r540vyjsvw5ac3l26ddgzn5p", + "mix_node": { + "host": "139.162.211.224", + "mix_port": 1789, + "verloc_port": 1790, + "http_api_port": 8000, + "clients_port": 9000, + "layer": 3, + "location": "London, UK", + "sphinx_key": "EvYDvouNZQTmYbY2hgmdY55AxNmZCv8R8x3Kjrg7F8q7", + "identity_key": "J1xMf1BYEpT2W7HGk8HUCAXhhDubC6U61AzkU9hGDjHc", + "version": "0.10.0" + } + } + ], + "gateways": [ + { + "amount": [ + { + "denom": "uhal", + "amount": "100000000" + } + ], + "owner": "hal1r2w0y5v93sawac3yfhx0fkls86cp3mxn44jucp", + "gateway": { + "host": "176.58.120.67", + "mix_port": 1789, + "clients_port": 9000, + "location": "London, UK", + "sphinx_key": "4zdhKkDQ2ZuyqCA2Mm5e2ZxwbrnFTCfdMP1FmDKhB9Mp", + "identity_key": "AmoRv85ak8UrYkqd43NZpQJFQjn8rtgMfViBgAFaPDRh", + "version": "0.10.0" + } + } + ] +} diff --git a/validator-api/v6.json b/validator-api/v6.json new file mode 100644 index 0000000000..3c28fed12c --- /dev/null +++ b/validator-api/v6.json @@ -0,0 +1,87 @@ +{ + "mixnodes": [ + { + "amount": [ + { + "denom": "uhal", + "amount": "100000000" + } + ], + "owner": "hal1c8t5cj5sgjt474y3wr04m6h3648yg7se0dkff2", + "mix_node": { + "host": "2a01:7e00::f03c:92ff:fe2f:5eee", + "mix_port": 1789, + "verloc_port": 1790, + "http_api_port": 8000, + "clients_port": 9000, + "layer": 1, + "location": "London, UK", + "sphinx_key": "73V9Hdo79YxAUwUgAN6qgkhzcDJGymcGnGF6tFseaWFT", + "identity_key": "824WyExLUWvLE2mpSHBatN4AoByuLzfnHFeHWiBYzg4z", + "version": "0.10.0" + } + }, + { + "amount": [ + { + "denom": "uhal", + "amount": "100000000" + } + ], + "owner": "hal1yz3ac9382la00l4srczdtma0vrg5p7a72kky6t", + "mix_node": { + "host": "2a01:7e00::f03c:92ff:fe2f:d749", + "mix_port": 1789, + "verloc_port": 1790, + "http_api_port": 8000, + "clients_port": 9000, + "layer": 2, + "location": "London, UK", + "sphinx_key": "7eAfV7CgzGzbKU9QKYyUFKmvsCgPC1NMUF2U16Guezsn", + "identity_key": "7NBLS5M7QxFqUZkBZn2RWe3kkM8uH3MT2mtK4UPP2q7d", + "version": "0.10.0" + } + }, + { + "amount": [ + { + "denom": "uhal", + "amount": "100000000" + } + ], + "owner": "hal1rz7sqd90pjj3a6r540vyjsvw5ac3l26ddgzn5p", + "mix_node": { + "host": "2a01:7e00::f03c:92ff:fe2f:6b03", + "layer": 3, + "mix_port": 1789, + "verloc_port": 1790, + "http_api_port": 8000, + "clients_port": 9000, + "location": "London, UK", + "sphinx_key": "EvYDvouNZQTmYbY2hgmdY55AxNmZCv8R8x3Kjrg7F8q7", + "identity_key": "J1xMf1BYEpT2W7HGk8HUCAXhhDubC6U61AzkU9hGDjHc", + "version": "0.10.0" + } + } + ], + "gateways": [ + { + "amount": [ + { + "denom": "uhal", + "amount": "100000000" + } + ], + "owner": "hal1r2w0y5v93sawac3yfhx0fkls86cp3mxn44jucp", + "gateway": { + "host": "2a01:7e00::f03c:92ff:fe2f:6b08", + "mix_port": 1789, + "clients_port": 9000, + "location": "London, UK", + "sphinx_key": "4zdhKkDQ2ZuyqCA2Mm5e2ZxwbrnFTCfdMP1FmDKhB9Mp", + "identity_key": "AmoRv85ak8UrYkqd43NZpQJFQjn8rtgMfViBgAFaPDRh", + "version": "0.10.0" + } + } + ] +}