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

Validator API server (#665)

Browse Source 
* Rocket main stub

* Add anyhow

* Stub cache reads and writes

* Finalize stubs

* Add generic Rocket.toml

* Put back targets

* Have cache own its validator client

* allow dead code

* Update rocket.toml for 0.5
This commit is contained in:
Drazen Urch
2021-07-01 14:08:04 +02:00
committed by GitHub
parent 1660e0796d
commit 343c55f981
16 changed files with 568 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files
validator-api/src/cache.rs
+41
View File
@@ -0,0 +1,41 @@
use anyhow::Result;
use mixnet_contract::MixNodeBond;
use std::time::Instant;
use validator_client::Client;
pub struct MixNodeCache {
value: Vec<MixNodeBond>,
#[allow(dead_code)]
as_at: Instant,
validator_client: Client,
}
impl MixNodeCache {
pub fn init(
value: Vec<MixNodeBond>,
validators_rest_uris: Vec<String>,
mixnet_contract: String,
) -> Self {
let config = validator_client::Config::new(validators_rest_uris, mixnet_contract);
let validator_client = validator_client::Client::new(config);
MixNodeCache {
value,
as_at: Instant::now(),
validator_client,
}
}
fn set_value(&mut self, value: Vec<MixNodeBond>) {
self.value = value;
self.as_at = Instant::now()
}
pub fn value(&self) -> Vec<MixNodeBond> {
self.value.clone()
}
pub async fn cache(&mut self) -> Result<()> {
let mixnodes = self.validator_client.get_mix_nodes().await?;
self.set_value(mixnodes);
Ok(())
}
}
validator-api/src/gateways_reader.rs
-1
View File
@@ -5,7 +5,6 @@ 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};
validator-api/src/main.rs
+192 -112
View File
@@ -1,6 +1,14 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
#[macro_use]
extern crate rocket;
use rocket::http::Method;
use rocket::serde::json::Json;
use rocket::State;
use rocket_cors::{AllowedHeaders, AllowedOrigins};
use crate::monitor::preparer::PacketPreparer;
use crate::monitor::processor::{
ReceivedProcessor, ReceivedProcessorReceiver, ReceivedProcessorSender,
@@ -12,15 +20,21 @@ 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 anyhow::Result;
use cache::MixNodeCache;
use clap::{App, Arg, ArgMatches};
use crypto::asymmetric::{encryption, identity};
use futures::channel::mpsc;
use log::*;
use log::info;
use mixnet_contract::MixNodeBond;
use nymsphinx::addressing::clients::Recipient;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tokio::time;
use topology::NymTopology;
mod cache;
mod chunker;
pub(crate) mod gateways_reader;
mod monitor;
@@ -35,6 +49,7 @@ 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 CACHE_INTERVAL_ARG: &str = "cache-interval";
const DEFAULT_VALIDATORS: &[&str] = &[
// "http://testnet-finney-validator.nymtech.net:1317",
@@ -45,6 +60,7 @@ const DEFAULT_VALIDATORS: &[&str] = &[
const DEFAULT_NODE_STATUS_API: &str = "http://localhost:8081";
const DEFAULT_GATEWAY_SENDING_RATE: usize = 500;
const DEFAULT_MIXNET_CONTRACT: &str = "hal1k0jntykt7e4g3y88ltc60czgjuqdy4c9c6gv94";
const DEFAULT_CACHE_INTERVAL_ARG: u64 = 60;
pub(crate) const TIME_CHUNK_SIZE: Duration = Duration::from_millis(50);
pub(crate) const PENALISE_OUTDATED: bool = false;
@@ -99,120 +115,13 @@ fn parse_args<'a>() -> ArgMatches<'a> {
.long(GATEWAY_SENDING_RATE_ARG)
.short("r")
)
.arg(Arg::with_name(CACHE_INTERVAL_ARG)
.help("Specified rate at which cache will be refreshed, global for all cache")
.takes_value(true)
.long(CACHE_INTERVAL_ARG))
.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!(
@@ -354,3 +263,174 @@ fn check_if_up_to_date(v4_topology: &NymTopology, v6_topology: &NymTopology) {
}
}
}
#[get("/mixnodes")]
async fn get_mixnodes(mixnode_cache: &State<Arc<RwLock<MixNodeCache>>>) -> Json<Vec<MixNodeBond>> {
let mixnodes = mixnode_cache.read().await;
Json(mixnodes.value())
}
#[tokio::main]
async fn main() -> Result<()> {
info!("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)
.to_string();
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);
let cache_interval_arg = matches
.value_of(CACHE_INTERVAL_ARG)
.map(|v| v.parse().unwrap())
.unwrap_or_else(|| DEFAULT_CACHE_INTERVAL_ARG);
check_if_up_to_date(&v4_topology, &v6_topology);
setup_logging();
info!("* validator servers: {:?}", validators_rest_uris);
info!("* node status api server: {}", node_status_api_uri);
info!("* mixnet contract: {}", mixnet_contract);
info!("* detailed report printing: {}", detailed_report);
info!("* 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();
info!(
"* 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.clone(), &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,
);
let mixnode_cache = Arc::new(RwLock::new(MixNodeCache::init(
vec![],
validators_rest_uris,
mixnet_contract,
)));
let write_mixnode_cache = Arc::clone(&mixnode_cache);
tokio::spawn(async move {
let mut interval = time::interval(Duration::from_secs(cache_interval_arg));
loop {
interval.tick().await;
{
match write_mixnode_cache.try_write() {
Ok(mut w) => w.cache().await.unwrap(),
// If we don't get the write lock skip a tick
Err(e) => error!("Could not aquire write lock on cache: {}", e),
}
}
}
});
tokio::spawn(async move { packet_receiver.run().await });
tokio::spawn(async move { monitor.run().await });
let allowed_origins = AllowedOrigins::all();
// You can also deserialize this
let cors = rocket_cors::CorsOptions {
allowed_origins,
allowed_methods: vec![Method::Post, Method::Get]
.into_iter()
.map(From::from)
.collect(),
allowed_headers: AllowedHeaders::all(),
allow_credentials: true,
..Default::default()
}
.to_cors()?;
rocket::build()
.attach(cors)
.mount("/", routes![get_mixnodes])
.manage(mixnode_cache)
.ignite()
.await?
.launch()
.await?;
wait_for_interrupt().await;
Ok(())
}
validator-api/src/monitor/mod.rs
+8 -4
View File
@@ -9,7 +9,7 @@ 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 log::{debug, info};
use tokio::time::{sleep, Duration, Instant};
pub(crate) mod preparer;
@@ -139,7 +139,7 @@ impl Monitor {
async fn test_run(&mut self) {
info!(target: "Monitor", "Starting test run no. {}", self.nonce);
debug!(target: "Monitor", "preparing mix packets to all nodes...");
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) => {
@@ -151,12 +151,16 @@ impl Monitor {
self.received_processor.set_new_expected(self.nonce).await;
info!(target: "Monitor", "starting to send all the packets...");
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);
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;
validator-api/src/monitor/preparer.rs
+7 -5
View File
@@ -6,7 +6,7 @@ use crate::monitor::sender::GatewayPackets;
use crate::test_packet::{NodeType, TestPacket};
use crate::tested_network::TestedNetwork;
use crypto::asymmetric::{encryption, identity};
use log::*;
use log::{info, warn};
use mixnet_contract::{GatewayBond, MixNodeBond};
use nymsphinx::addressing::clients::Recipient;
use nymsphinx::forwarding::packet::MixPacket;
@@ -248,8 +248,9 @@ impl PacketPreparer {
PreparedNode::TestedMix(mix, [v4_packet, v6_packet])
}
Err(err) => {
warn!(target: "bad node",
"mix {} is malformed - {}",
warn!(
target: "Bad node",
"Mix {} is malformed - {}",
mixnode_bond.mix_node().identity_key,
err
);
@@ -286,8 +287,9 @@ impl PacketPreparer {
PreparedNode::TestedGateway(gateway, [v4_packet, v6_packet])
}
Err(err) => {
warn!(target: "bad node",
"gateway {} is malformed - {:?}",
warn!(
target: "Bad node",
"gateway {} is malformed - {:?}",
gateway_bond.gateway().identity_key,
err
);
validator-api/src/monitor/processor.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use crypto::asymmetric::encryption;
use futures::channel::mpsc;
use futures::lock::{Mutex, MutexGuard};
use futures::{SinkExt, StreamExt};
use log::*;
use log::warn;
use nymsphinx::receiver::MessageReceiver;
use std::fmt::{self, Display, Formatter};
use std::mem;
validator-api/src/monitor/sender.rs
+13 -7
View File
@@ -10,7 +10,7 @@ use futures::task::Context;
use futures::{Future, Stream};
use gateway_client::error::GatewayClientError;
use gateway_client::{AcknowledgementReceiver, GatewayClient, MixnetMessageReceiver};
use log::*;
use log::{debug, info, warn};
use nymsphinx::forwarding::packet::MixPacket;
use pin_project::pin_project;
use std::collections::HashMap;
@@ -129,10 +129,15 @@ impl PacketSender {
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);
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.");
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 =
@@ -140,9 +145,10 @@ impl PacketSender {
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
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>> {
@@ -160,7 +166,7 @@ impl PacketSender {
// 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());
debug!(target: "MessageSender","Sending {} packets...", mix_packets.len());
if mix_packets.len() == 1 {
client.send_mix_packet(mix_packets.pop().unwrap()).await?;
validator-api/src/monitor/summary_producer.rs
+39 -22
View File
@@ -7,7 +7,6 @@ use crate::node_status_api::models::{
};
use crate::test_packet::{NodeType, TestPacket};
use crate::PENALISE_OUTDATED;
use log::*;
use std::collections::HashMap;
#[derive(Default)]
@@ -74,56 +73,74 @@ pub(crate) struct TestReport {
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!("Sent total of {} packets", self.total_sent);
info!("Received total of {} packets", self.total_received);
info!("{} 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!(
"{} mixnodes speak ONLY IPv4 (NO IPv6 connectivity)",
self.only_ipv4_compatible_mixes.len()
);
info!(
"{} mixnodes speak ONLY IPv6 (NO IPv4 connectivity)",
self.only_ipv6_compatible_mixes.len()
);
info!(
"{} mixnodes are totally unroutable!",
self.completely_unroutable_mixes.len()
);
info!("{} 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());
info!(
"{} gateways speak ONLY IPv4 (NO IPv6 connectivity)",
self.only_ipv4_compatible_gateways.len()
);
info!(
"{} gateways speak ONLY IPv6 (NO IPv4 connectivity)",
self.only_ipv6_compatible_gateways.len()
);
info!(
"{} gateways are totally unroutable!",
self.completely_unroutable_gateways.len()
);
info!("{} gateways work fine!", self.fully_working_gateways.len());
if detailed {
info!(target: "Detailed report", "full summary:");
info!("full summary:");
for malformed in self.malformed.iter() {
info!(target: "Invalid node", "{}", malformed)
info!("{}", malformed)
}
for v4_node in self.only_ipv4_compatible_mixes.iter() {
info!(target: "IPv4-only mixnode", "{}", v4_node)
info!("{}", v4_node)
}
for v6_node in self.only_ipv6_compatible_mixes.iter() {
info!(target: "IPv6-only mixnode", "{}", v6_node)
info!("{}", v6_node)
}
for unroutable in self.completely_unroutable_mixes.iter() {
info!(target: "Unroutable mixnode", "{}", unroutable)
info!("{}", unroutable)
}
for working in self.fully_working_mixes.iter() {
info!(target: "Fully working mixnode", "{}", working)
info!("{}", working)
}
for v4_node in self.only_ipv4_compatible_gateways.iter() {
info!(target: "IPv4-only gateway", "{}", v4_node)
info!("{}", v4_node)
}
for v6_node in self.only_ipv6_compatible_gateways.iter() {
info!(target: "IPv6-only gateway", "{}", v6_node)
info!("{}", v6_node)
}
for unroutable in self.completely_unroutable_gateways.iter() {
info!(target: "Unroutable gateway", "{}", unroutable)
info!("{}", unroutable)
}
for working in self.fully_working_gateways.iter() {
info!(target: "Fully working gateway", "{}", working)
info!("{}", working)
}
}
}
validator-api/src/tested_network/mod.rs
-1
View File
@@ -2,7 +2,6 @@
// SPDX-License-Identifier: Apache-2.0
use crate::test_packet::IpVersion;
use log::*;
use topology::{gateway, mix, NymTopology};
pub(crate) mod good_topology;