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

Removed standalone network-monitor (#669)

Browse Source 
It has been integrated into the validator-api
This commit is contained in:
Jędrzej Stuczyński
2021-07-02 11:31:29 +01:00
committed by GitHub
parent 3b7315d1e2
commit e6460208d3
16 changed files with 0 additions and 2926 deletions
Download Patch File Download Diff File Expand all files Collapse all files
network-monitor/Cargo.toml
-35
View File
@@ -1,35 +0,0 @@
# Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
# SPDX-License-Identifier: Apache-2.0
[package]
name = "nym-validator-api"
version = "0.10.1"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
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]
network-monitor/src/chunker.rs
-75
View File
@@ -1,75 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// 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<OsRng>,
}
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<u8>,
topology: &NymTopology,
packet_sender: Recipient,
) -> Vec<MixPacket> {
// I really dislike how we have to overwrite the parameter of the `MessagePreparer` on each run
// but without some significant API changes in the `MessagePreparer` this was the easiest
// way to being able to have variable sender address.
self.message_preparer.set_sender_address(packet_sender);
self.prepare_packets(message, topology, packet_sender).await
}
async fn prepare_packets(
&mut self,
message: Vec<u8>,
topology: &NymTopology,
packet_sender: Recipient,
) -> Vec<MixPacket> {
let ack_key: AckKey = AckKey::new(&mut self.rng);
let (split_message, _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
}
}
network-monitor/src/gateways_reader.rs
-183
View File
@@ -1,183 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<Vec<u8>>) and I definitely
// had a reason for doing so, but right now I can't remember what that was...
pub(crate) type GatewayMessages = Vec<Vec<u8>>;
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<Vec<u8>>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
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<GatewayChannel>,
channels: Vec<GatewayChannel>,
waker: Option<Waker>,
}
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<Poll<Option<GatewayMessages>>> {
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<Option<Self::Item>> {
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
}
}
network-monitor/src/main.rs
-356
View File
@@ -1,356 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// 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<usize> = 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::<Vec<_>>())
.unwrap_or_else(|| DEFAULT_VALIDATORS.to_vec());
let validators_rest_uris = validators_rest_uris_borrowed
.into_iter()
.map(|uri| uri.to_string())
.collect::<Vec<_>>();
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<identity::KeyPair>,
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<encryption::KeyPair>,
) -> 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<String>,
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<S: Into<String>>(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
)
}
}
}
network-monitor/src/monitor/mod.rs
-220
View File
@@ -1,220 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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);
}
}
}
}
}
network-monitor/src/monitor/preparer.rs
-477
View File
@@ -1,477 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<S1, S2>(identity: S1, owner: S2) -> Self
where
S1: Into<String>,
S2: Into<String>,
{
TestedNode {
identity: identity.into(),
owner: owner.into(),
node_type: NodeType::Mixnode,
}
}
pub(crate) fn from_raw_gateway<S1, S2>(identity: S1, owner: S2) -> Self
where
S1: Into<String>,
S2: Into<String>,
{
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<GatewayPackets>,
/// 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<TestedNode>,
/// 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<InvalidNode>,
}
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<MixNodeBond>, Vec<GatewayBond>), 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::<mix::Node>::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::<gateway::Node>::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<PreparedNode> {
nodes
.iter()
.map(|mix| self.mix_into_prepared_node(nonce, mix))
.collect()
}
fn prepare_gateways(&self, nonce: u64, nodes: &[GatewayBond]) -> Vec<PreparedNode> {
nodes
.iter()
.map(|gateway| self.gateway_into_prepared_node(nonce, gateway))
.collect()
}
fn tested_nodes(&self, nodes: &[PreparedNode]) -> Vec<TestedNode> {
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<PreparedNode>,
invalid: &mut Vec<InvalidNode>,
) -> Vec<MixPacket> {
// 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<PreparedNode>,
invalid: &mut Vec<InvalidNode>,
) -> Vec<GatewayPackets> {
// 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<PreparedPackets, PacketPreparerError> {
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,
})
}
}
network-monitor/src/monitor/processor.rs
-227
View File
@@ -1,227 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<GatewayMessages>;
pub(crate) type ReceivedProcessorReceiver = mpsc::UnboundedReceiver<GatewayMessages>;
#[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<u64>,
/// Channel for receiving packets/messages from the gateway clients
packets_receiver: ReceivedProcessorReceiver,
// TODO: right now it's identical for each gateway we send through, but should it?
/// Encryption key of the clients sending through the gateways.
client_encryption_keypair: Arc<encryption::KeyPair>,
/// Structure responsible for decrypting and recovering plaintext message from received ciphertexts.
message_receiver: MessageReceiver,
/// Vector containing all received (and decrypted) packets in the current test run.
received_packets: Vec<TestPacket>,
}
impl ReceivedProcessorInner {
fn on_message(&mut self, message: Vec<u8>) -> Result<(), ProcessingError> {
// if the nonce is none it means the packet was received during the 'waiting' for the
// next test run
if self.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<TestPacket> {
self.nonce = None;
mem::take(&mut self.received_packets)
}
}
pub(crate) struct ReceivedProcessor {
permit_changer: mpsc::Sender<LockPermit>,
inner: Arc<Mutex<ReceivedProcessorInner>>,
}
impl ReceivedProcessor {
pub(crate) fn new(
packets_receiver: ReceivedProcessorReceiver,
client_encryption_keypair: Arc<encryption::KeyPair>,
) -> Self {
let inner: Arc<Mutex<ReceivedProcessorInner>> =
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<Mutex<ReceivedProcessorInner>>,
mut permit_change: mpsc::Receiver<LockPermit>,
) {
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<LockPermit>,
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<LockPermit>,
inner: &'a Mutex<ReceivedProcessorInner>,
) -> 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<TestPacket> {
// 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
}
}
network-monitor/src/monitor/receiver.rs
-69
View File
@@ -1,69 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<GatewayClientUpdate>;
pub(crate) type GatewayClientUpdateReceiver = mpsc::UnboundedReceiver<GatewayClientUpdate>;
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()),
}
}
}
}
network-monitor/src/monitor/sender.rs
-448
View File
@@ -1,448 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<MixPacket>,
}
impl GatewayPackets {
pub(crate) fn new(
clients_address: String,
pub_key: identity::PublicKey,
packets: Vec<MixPacket>,
) -> Self {
GatewayPackets {
clients_address,
pub_key,
packets,
}
}
pub(super) fn push_packets(&mut self, mut packets: Vec<MixPacket>) {
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<identity::KeyPair>,
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<FreshGatewayClientData>,
max_concurrent_clients: Option<usize>,
max_sending_rate: usize,
}
impl PacketSender {
pub(crate) fn new(
gateways_status_updater: GatewayClientUpdateSender,
local_identity: Arc<identity::KeyPair>,
gateway_response_timeout: Duration,
max_concurrent_clients: Option<usize>,
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<MixPacket>,
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<MixPacket>, at: usize) -> Option<Vec<MixPacket>> {
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<FreshGatewayClientData>,
client: Option<GatewayClient>,
max_sending_rate: usize,
) -> Option<GatewayClient> {
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<GatewayPackets>) {
// 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<RwLock<HashMap<key, Mutex<GatewayClient>>>>
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<St, Fut, F> {
#[pin]
stream: Option<St>,
f: F,
futures: FuturesUnordered<Fut>,
limit: Option<NonZeroUsize>,
result: Vec<Option<GatewayClient>>,
}
impl<St, Fut, F> ForEachConcurrentClientUse<St, Fut, F>
where
St: Stream,
F: FnMut(St::Item) -> Fut,
Fut: Future<Output = Option<GatewayClient>>,
{
pub(super) fn new(stream: St, limit: Option<usize>, 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<St, Fut, F> Future for ForEachConcurrentClientUse<St, Fut, F>
where
St: Stream,
F: FnMut(St::Item) -> Fut,
Fut: Future<Output = Option<GatewayClient>>,
{
type Output = Vec<Option<GatewayClient>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
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;
}
}
}
}
network-monitor/src/monitor/summary_producer.rs
-270
View File
@@ -1,270 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<MixStatus> {
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<GatewayStatus> {
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<InvalidNode>,
// below are only populated if we're going to be printing the report
pub(crate) only_ipv4_compatible_mixes: Vec<TestedNode>, // can't speak v6, but can speak v4
pub(crate) only_ipv6_compatible_mixes: Vec<TestedNode>, // can't speak v4, but can speak v6
pub(crate) completely_unroutable_mixes: Vec<TestedNode>, // can't speak either v4 or v6
pub(crate) fully_working_mixes: Vec<TestedNode>,
pub(crate) only_ipv4_compatible_gateways: Vec<TestedNode>, // can't speak v6, but can speak v4
pub(crate) only_ipv6_compatible_gateways: Vec<TestedNode>, // can't speak v4, but can speak v6
pub(crate) completely_unroutable_gateways: Vec<TestedNode>, // can't speak either v4 or v6
pub(crate) fully_working_gateways: Vec<TestedNode>,
}
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<TestedNode, NodeResult>) {
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<TestedNode>,
received_packets: Vec<TestPacket>,
invalid_nodes: Vec<InvalidNode>,
) -> 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<TestedNode, NodeResult> = 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,
}
}
}
network-monitor/src/node_status_api/client.rs
-110
View File
@@ -1,110 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<S: Into<String>>(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()
)))
}
}
}
network-monitor/src/node_status_api/mod.rs
-73
View File
@@ -1,73 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<reqwest::Error> for NodeStatusApiClientError {
fn from(err: reqwest::Error) -> Self {
NodeStatusApiClientError::ReqwestClientError(err)
}
}
impl From<ErrorResponses> 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::<String>()
)
}
}
}
}
}
network-monitor/src/node_status_api/models.rs
-68
View File
@@ -1,68 +0,0 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<MixStatus>,
}
#[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<GatewayStatus>,
}
#[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),
}
network-monitor/src/test_packet.rs
-214
View File
@@ -1,214 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// 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<u8> for NodeType {
type Error = TestPacketError;
fn try_from(value: u8) -> Result<Self, Self::Error> {
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<identity::KeyRecoveryError> for TestPacketError {
fn from(_: identity::KeyRecoveryError) -> Self {
TestPacketError::InvalidNodeKey
}
}
impl From<Utf8Error> 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<u8> for IpVersion {
type Error = TestPacketError;
fn try_from(value: u8) -> Result<Self, Self::Error> {
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<IpVersion> 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<H: Hasher>(&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<u8> {
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<Self, TestPacketError> {
// nonce size
let n = mem::size_of::<u64>();
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<TestPacket> for TestedNode {
fn from(packet: TestPacket) -> Self {
TestedNode {
identity: packet.pub_key.to_base58_string(),
owner: packet.owner,
node_type: packet.node_type,
}
}
}
network-monitor/src/tested_network/good_topology.rs
-29
View File
@@ -1,29 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// 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<MixNodeBond>,
gateways: Vec<GatewayBond>,
}
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::<GoodTopology>(&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
}
network-monitor/src/tested_network/mod.rs
-72
View File
@@ -1,72 +0,0 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// 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
}
}