Network monitor using validator client

network-monitor/Cargo.toml

@@ -16,10 +16,10 @@ tokio = { version = "0.2", features = ["signal", "rt-threaded", "macros"] }
 
 ## internal
 crypto = { path = "../common/crypto" }
-directory-client = { path = "../common/client-libs/directory-client" }
 gateway-client = { path = "../common/client-libs/gateway-client" }
 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/main.rs

@@ -16,7 +16,6 @@ use crate::monitor::MixnetReceiver;
 use crate::run_info::{TestRunUpdateReceiver, TestRunUpdateSender};
 use crate::tested_network::{good_topology, TestedNetwork};
 use crypto::asymmetric::{encryption, identity};
-use directory_client::DirectoryClient;
 use futures::channel::mpsc;
 use gateway_client::GatewayClient;
 use monitor::{AckSender, MixnetSender, Monitor};
@@ -54,8 +53,8 @@ async fn main() {
     setup_logging();
 
     // Set up topology
-    let directory_uri = "https://qa-directory.nymtech.net";
-    println!("* directory server: {}", directory_uri);
+    let validator_rest_uri = "https://qa-directory.nymtech.net";
+    println!("* validator server: {}", validator_rest_uri);
 
     // TODO: this might change if it turns out we need both v4 and v6 gateway clients
     let gateway = tested_network::v4_gateway();
@@ -78,13 +77,13 @@ async fn main() {
         gateway.identity_key,
     );
 
-    let directory_client = new_directory_client(directory_uri);
+    let validator_client = new_validator_client(validator_rest_uri);
 
     let mut network_monitor = Monitor::new();
 
     let notifier = new_notifier(
         encryption_keypair,
-        Arc::clone(&directory_client),
+        Arc::clone(&validator_client),
         mixnet_receiver,
         test_run_receiver,
     );
@@ -93,7 +92,7 @@ async fn main() {
     let tested_network = new_tested_network(gateway_client).await;
 
     let packet_sender = new_packet_sender(
-        directory_client,
+        validator_client,
         tested_network,
         self_address,
         test_run_sender,
@@ -110,13 +109,13 @@ async fn new_tested_network(gateway_client: GatewayClient) -> TestedNetwork {
 }
 
 fn new_packet_sender(
-    directory_client: Arc<directory_client::Client>,
+    validator_client: Arc<validator_client::Client>,
     tested_network: TestedNetwork,
     self_address: Recipient,
     test_run_sender: TestRunUpdateSender,
 ) -> PacketSender {
     PacketSender::new(
-        directory_client,
+        validator_client,
         tested_network,
         self_address,
         test_run_sender,
@@ -144,21 +143,21 @@ pub fn new_gateway_client(
     )
 }
 
-fn new_directory_client(directory_uri: &str) -> Arc<directory_client::Client> {
-    let config = directory_client::Config::new(directory_uri.to_string());
-    Arc::new(DirectoryClient::new(config))
+fn new_validator_client(validator_rest_uri: &str) -> Arc<validator_client::Client> {
+    let config = validator_client::Config::new(validator_rest_uri.to_string());
+    Arc::new(validator_client::Client::new(config))
 }
 
 fn new_notifier(
     encryption_keypair: encryption::KeyPair,
-    directory_client: Arc<directory_client::Client>,
+    validator_client: Arc<validator_client::Client>,
     mixnet_receiver: MixnetReceiver,
     test_run_receiver: TestRunUpdateReceiver,
 ) -> Notifier {
     Notifier::new(
         mixnet_receiver,
         encryption_keypair,
-        directory_client,
+        validator_client,
         test_run_receiver,
     )
 }

network-monitor/src/notifications/mod.rs

@@ -19,18 +19,19 @@ use crate::notifications::test_timeout::TestTimeout;
 use crate::run_info::{RunInfo, TestRunUpdate, TestRunUpdateReceiver};
 use crate::PRINT_DETAILED_REPORT;
 use crypto::asymmetric::encryption::KeyPair;
-use directory_client::mixmining::BatchMixStatus;
 use futures::StreamExt;
 use log::*;
 use nymsphinx::receiver::MessageReceiver;
 use std::sync::Arc;
+use validator_client::models::mixmining::BatchMixStatus;
+use validator_client::ValidatorClientError;
 
 mod test_run;
 mod test_timeout;
 
 #[derive(Debug)]
 enum NotifierError {
-    DirectoryError(String),
+    ValidatorError(ValidatorClientError),
     MalformedPacketReceived,
     NonTestPacketReceived,
 }
@@ -39,7 +40,7 @@ pub(crate) struct Notifier {
     client_encryption_keypair: KeyPair,
     message_receiver: MessageReceiver,
     mixnet_receiver: MixnetReceiver,
-    directory_client: Arc<directory_client::Client>,
+    validator_client: Arc<validator_client::Client>,
     test_run_receiver: TestRunUpdateReceiver,
     test_run_nonce: u64,
     current_test_run: TestRun,
@@ -50,7 +51,7 @@ impl Notifier {
     pub(crate) fn new(
         mixnet_receiver: MixnetReceiver,
         client_encryption_keypair: KeyPair,
-        directory_client: Arc<directory_client::Client>,
+        validator_client: Arc<validator_client::Client>,
         test_run_receiver: TestRunUpdateReceiver,
     ) -> Notifier {
         let message_receiver = MessageReceiver::new();
@@ -62,7 +63,7 @@ impl Notifier {
             client_encryption_keypair,
             message_receiver,
             mixnet_receiver,
-            directory_client,
+            validator_client,
             test_run_receiver,
             test_run_nonce: 0,
             current_test_run,
@@ -145,10 +146,10 @@ impl Notifier {
     }
 
     async fn notify_validator(&self, status: BatchMixStatus) -> Result<(), NotifierError> {
-        self.directory_client
+        self.validator_client
             .post_batch_mixmining_status(status)
             .await
-            .map_err(|err| NotifierError::DirectoryError(err.to_string()))?;
+            .map_err(|err| NotifierError::ValidatorError(err))?;
         Ok(())
     }
 }

network-monitor/src/notifications/test_run.rs

@@ -14,10 +14,10 @@
 
 use crate::run_info::RunInfo;
 use crate::test_packet::TestPacket;
-use directory_client::mixmining::{BatchMixStatus, MixStatus};
 use log::*;
 use std::collections::{HashMap, HashSet};
 use std::mem;
+use validator_client::models::mixmining::{BatchMixStatus, MixStatus};
 
 pub(super) struct TestRun {
     print_report: bool,

network-monitor/src/packet_sender.rs

@@ -16,18 +16,19 @@ use crate::chunker::Chunker;
 use crate::run_info::{RunInfo, TestRunUpdate, TestRunUpdateSender};
 use crate::test_packet::{IpVersion, TestPacket};
 use crate::tested_network::{TestMix, TestedNetwork};
-use directory_client::presence::mixnodes::MixNodePresence;
 use gateway_client::error::GatewayClientError;
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::forwarding::packet::MixPacket;
 use std::convert::TryInto;
 use std::sync::Arc;
 use topology::mix;
+use validator_client::models::mixnode::RegisteredMix;
+use validator_client::ValidatorClientError;
 
 #[derive(Debug)]
 pub(crate) enum PacketSenderError {
-    DirectoryError(String),
+    ValidatorError(ValidatorClientError),
     GatewayError(GatewayClientError),
 }
 
@@ -39,34 +40,34 @@ impl From<GatewayClientError> for PacketSenderError {
 
 pub struct PacketSender {
     chunker: Chunker,
-    directory_client: Arc<directory_client::Client>,
+    validator_client: Arc<validator_client::Client>,
     tested_network: TestedNetwork,
     test_run_sender: TestRunUpdateSender,
     nonce: u64,
 }
 
 impl PacketSender {
     pub(crate) fn new(
-        directory_client: Arc<directory_client::Client>,
+        validator_client: Arc<validator_client::Client>,
         tested_network: TestedNetwork,
         self_address: Recipient,
         test_run_sender: TestRunUpdateSender,
     ) -> Self {
         PacketSender {
             chunker: Chunker::new(self_address),
-            directory_client,
+            validator_client,
             tested_network,
             test_run_sender,
             nonce: 0,
         }
     }
 
-    fn make_test_mix(&self, presence: MixNodePresence) -> TestMix {
+    fn make_test_mix(&self, mix: RegisteredMix) -> TestMix {
         // the reason for that conversion is that I want to operate on concrete types
         // rather than on "String" everywhere and also this way we remove obviously wrong
         // mixnodes where somebody is sending bullshit presence data.
-        let mix_id = presence.pub_key.clone();
-        let mix: Result<mix::Node, _> = presence.try_into();
+        let mix_id = mix.identity().clone();
+        let mix: Result<mix::Node, _> = mix.try_into();
         match mix {
             Err(err) => {
                 error!("mix {} is malformed - {:?}", mix_id, err);
@@ -77,8 +78,10 @@ impl PacketSender {
                     &mix.version,
                     self.tested_network.system_version(),
                 ) {
-                    let v4_test_packet = TestPacket::new(mix.pub_key, IpVersion::V4, self.nonce);
-                    let v6_test_packet = TestPacket::new(mix.pub_key, IpVersion::V6, self.nonce);
+                    let v4_test_packet =
+                        TestPacket::new(mix.identity_key, IpVersion::V4, self.nonce);
+                    let v6_test_packet =
+                        TestPacket::new(mix.identity_key, IpVersion::V6, self.nonce);
 
                     TestMix::ValidMix(mix, [v4_test_packet, v6_test_packet])
                 } else {
@@ -90,13 +93,13 @@ impl PacketSender {
 
     async fn get_test_mixes(&self) -> Result<Vec<TestMix>, PacketSenderError> {
         Ok(self
-            .directory_client
+            .validator_client
             .get_topology()
             .await
-            .map_err(|err| PacketSenderError::DirectoryError(err.to_string()))?
+            .map_err(|err| PacketSenderError::ValidatorError(err))?
             .mix_nodes
             .into_iter()
-            .map(|presence| self.make_test_mix(presence))
+            .map(|mix| self.make_test_mix(mix))
             .collect())
     }
 
@@ -113,9 +116,8 @@ impl PacketSender {
                     test_packets.push(mix_test_packets[1]);
                 }
                 TestMix::MalformedMix(pub_key) => malformed_mixes.push(pub_key.clone()),
-                TestMix::IncompatibleMix(mix) => {
-                    incompatible_mixes.push((mix.pub_key.to_base58_string(), mix.version.clone()))
-                }
+                TestMix::IncompatibleMix(mix) => incompatible_mixes
+                    .push((mix.identity_key.to_base58_string(), mix.version.clone())),
             }
         }
         RunInfo {

network-monitor/src/test_packet.rs

@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crypto::asymmetric::encryption;
-use crypto::asymmetric::encryption::KeyRecoveryError;
-use directory_client::mixmining::MixStatus;
+use crypto::asymmetric::{encryption, identity};
 use std::convert::{TryFrom, TryInto};
 use std::fmt::{self, Display, Formatter};
+use std::hash::{Hash, Hasher};
 use std::mem;
+use validator_client::models::mixmining::MixStatus;
 
 #[derive(Debug)]
 pub(crate) enum TestPacketError {
@@ -26,8 +26,8 @@ pub(crate) enum TestPacketError {
     InvalidNodeKey,
 }
 
-impl From<encryption::KeyRecoveryError> for TestPacketError {
-    fn from(_: KeyRecoveryError) -> Self {
+impl From<identity::KeyRecoveryError> for TestPacketError {
+    fn from(_: identity::KeyRecoveryError) -> Self {
         TestPacketError::InvalidNodeKey
     }
 }
@@ -69,11 +69,11 @@ impl Display for IpVersion {
     }
 }
 
-#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)]
+#[derive(PartialEq, Eq, Copy, Clone, Debug)]
 pub(crate) struct TestPacket {
     ip_version: IpVersion,
     nonce: u64,
-    pub_key: encryption::PublicKey, // TODO: eventually this will get replaced with identity::PublicKey
+    pub_key: identity::PublicKey,
 }
 
 impl Display for TestPacket {
@@ -88,8 +88,16 @@ impl Display for TestPacket {
     }
 }
 
+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);
+    }
+}
+
 impl TestPacket {
-    pub(crate) fn new(pub_key: encryption::PublicKey, ip_version: IpVersion, nonce: u64) -> Self {
+    pub(crate) fn new(pub_key: identity::PublicKey, ip_version: IpVersion, nonce: u64) -> Self {
         TestPacket {
             pub_key,
             ip_version,
@@ -130,7 +138,7 @@ impl TestPacket {
         // 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 ip_version = IpVersion::try_from(b[n])?;
-        let pub_key = encryption::PublicKey::from_bytes(&b[n + 1..])?;
+        let pub_key = identity::PublicKey::from_bytes(&b[n + 1..])?;
 
         Ok(TestPacket {
             ip_version,

network-monitor/src/tested_network/good_topology.rs

@@ -20,68 +20,88 @@ use std::collections::HashMap;
 use topology::{gateway, mix, NymTopology};
 
 pub(crate) fn v4_mixnodes() -> Vec<mix::Node> {
+    todo!("this needs to be updated during 0.9.0 release when we deploy our 0.9.0 nodes");
+
     let goodnode1 = mix::Node {
         location: "London".to_string(),
         host: "213.52.129.218:1789".parse().unwrap(),
-        pub_key: PublicKey::from_base58_string("EJHwrLafqygqctkBCntVZfUkMSDErGUStJjZniQoRoJr")
-            .unwrap(),
-        last_seen: 1600276206950298819,
+        identity_key: identity::PublicKey::from_base58_string("TODO").unwrap(),
+        sphinx_key: encryption::PublicKey::from_base58_string(
+            "EJHwrLafqygqctkBCntVZfUkMSDErGUStJjZniQoRoJr",
+        )
+        .unwrap(),
+        registration_time: 1600276206950298819,
         layer: 1,
-        version: "0.8.0".to_string(),
+        version: "0.9.0".to_string(),
+        reputation: 100,
     };
 
     let goodnode2 = mix::Node {
         location: "Frankfurt".to_string(),
         host: "172.104.244.117:1789".parse().unwrap(),
-        pub_key: PublicKey::from_base58_string("BW7xskYvZyHt8rGFzsmG5bEQ9ViCYYxpFsEWDcNtSYvX")
+        identity_key: identity::PublicKey::from_base58_string("TODO").unwrap(),
+        sphinx_key: PublicKey::from_base58_string("BW7xskYvZyHt8rGFzsmG5bEQ9ViCYYxpFsEWDcNtSYvX")
             .unwrap(),
-        last_seen: 1600276206950298819,
+        registration_time: 1600276206950298819,
         layer: 2,
-        version: "0.8.0".to_string(),
+        version: "0.9.0".to_string(),
+        reputation: 100,
     };
 
     let goodnode3 = mix::Node {
         location: "London".to_string(),
         host: "178.79.136.231:1789".parse().unwrap(),
-        pub_key: PublicKey::from_base58_string("BqBGpP4YDH5fRDVKB97Ru7aq2Wbarb3SNfZL5LGaH83e")
+        identity_key: identity::PublicKey::from_base58_string("TODO").unwrap(),
+        sphinx_key: PublicKey::from_base58_string("BqBGpP4YDH5fRDVKB97Ru7aq2Wbarb3SNfZL5LGaH83e")
             .unwrap(),
         layer: 3,
-        last_seen: 1600276206950298819,
-        version: "0.8.0".to_string(),
+        registration_time: 1600276206950298819,
+        version: "0.9.0".to_string(),
+        reputation: 100,
     };
 
     vec![goodnode1, goodnode2, goodnode3]
 }
 
 pub(crate) fn v6_mixnodes() -> Vec<mix::Node> {
+    todo!("this needs to be updated during 0.9.0 release when we deploy our 0.9.0 nodes");
+
     let goodnode1 = mix::Node {
         location: "London".to_string(),
         host: "[2a01:7e00::f03c:92ff:fe16:3dc2]:1789".parse().unwrap(),
-        pub_key: PublicKey::from_base58_string("EJHwrLafqygqctkBCntVZfUkMSDErGUStJjZniQoRoJr")
-            .unwrap(),
-        last_seen: 1600276206950298819,
+        identity_key: identity::PublicKey::from_base58_string("TODO").unwrap(),
+        sphinx_key: encryption::PublicKey::from_base58_string(
+            "EJHwrLafqygqctkBCntVZfUkMSDErGUStJjZniQoRoJr",
+        )
+        .unwrap(),
+        registration_time: 1600276206950298819,
         layer: 1,
-        version: "0.8.0".to_string(),
+        version: "0.9.0".to_string(),
+        reputation: 100,
     };
 
     let goodnode2 = mix::Node {
         location: "Frankfurt".to_string(),
         host: "[2a01:7e01::f03c:92ff:fe16:3d11]:1789".parse().unwrap(),
-        pub_key: PublicKey::from_base58_string("BW7xskYvZyHt8rGFzsmG5bEQ9ViCYYxpFsEWDcNtSYvX")
+        identity_key: identity::PublicKey::from_base58_string("TODO").unwrap(),
+        sphinx_key: PublicKey::from_base58_string("BW7xskYvZyHt8rGFzsmG5bEQ9ViCYYxpFsEWDcNtSYvX")
             .unwrap(),
-        last_seen: 1600276206950298819,
+        registration_time: 1600276206950298819,
         layer: 2,
-        version: "0.8.0".to_string(),
+        version: "0.9.0".to_string(),
+        reputation: 100,
     };
 
     let goodnode3 = mix::Node {
         location: "London".to_string(),
         host: "[2a01:7e00::f03c:92ff:fe16:3d7b]:1789".parse().unwrap(),
-        pub_key: PublicKey::from_base58_string("BqBGpP4YDH5fRDVKB97Ru7aq2Wbarb3SNfZL5LGaH83e")
+        identity_key: identity::PublicKey::from_base58_string("TODO").unwrap(),
+        sphinx_key: PublicKey::from_base58_string("BqBGpP4YDH5fRDVKB97Ru7aq2Wbarb3SNfZL5LGaH83e")
             .unwrap(),
         layer: 3,
-        last_seen: 1600276206950298819,
-        version: "0.8.0".to_string(),
+        registration_time: 1600276206950298819,
+        version: "0.9.0".to_string(),
+        reputation: 100,
     };
 
     vec![goodnode1, goodnode2, goodnode3]
@@ -100,8 +120,9 @@ pub(crate) fn v4_gateway() -> gateway::Node {
             "6snGVMCatcTnvjGPaf8Ye7kCnVn6ThEDdCs4TZ7DbDVj",
         )
         .unwrap(),
-        last_seen: 1600424297774836793,
-        version: "0.8.0".to_string(),
+        registration_time: 1600424297774836793,
+        version: "0.9.0".to_string(),
+        reputation: 100,
     }
 }
 
@@ -118,8 +139,9 @@ pub(crate) fn v6_gateway() -> gateway::Node {
             "6snGVMCatcTnvjGPaf8Ye7kCnVn6ThEDdCs4TZ7DbDVj",
         )
         .unwrap(),
-        last_seen: 1600424297774836793,
-        version: "0.8.0".to_string(),
+        registration_time: 1600424297774836793,
+        version: "0.9.0".to_string(),
+        reputation: 100,
     }
 }
 
@@ -131,7 +153,7 @@ pub(crate) fn new_v4() -> NymTopology {
         layered_mixes.insert((i + 1) as u8, vec![node]);
     }
 
-    NymTopology::new(Vec::new(), layered_mixes, vec![v4_gateway()])
+    NymTopology::new(layered_mixes, vec![v4_gateway()])
 }
 
 /// Returns a new ipv6 NymTopology composed of known good nodes
@@ -142,7 +164,7 @@ pub(crate) fn new_v6() -> NymTopology {
         layered_mixes.insert((i + 1) as u8, vec![node]);
     }
 
-    NymTopology::new(Vec::new(), layered_mixes, vec![v6_gateway()])
+    NymTopology::new(layered_mixes, vec![v6_gateway()])
 }
 
 // Returns a new topology of known good nodes, with one good node replaced with a test node
@@ -163,20 +185,23 @@ mod good_topology_test {
         fn returns_good_topology_with_test_node_in_desired_layer() {
             let topology = expected_topology_with_test_node();
             let expected_gateway_key = topology.gateways().first().unwrap().identity_key;
-            let expected_layer_1_mixnode_pubkey =
-                topology.mixes_in_layer(1)[0].pub_key.to_base58_string();
-            let expected_layer_2_mixnode_pubkey =
-                topology.mixes_in_layer(2)[0].pub_key.to_base58_string();
-            let expected_layer_3_mixnode_pubkey =
-                topology.mixes_in_layer(3)[0].pub_key.to_base58_string();
+            let expected_layer_1_mixnode_pubkey = topology.mixes_in_layer(1)[0]
+                .identity_key
+                .to_base58_string();
+            let expected_layer_2_mixnode_pubkey = topology.mixes_in_layer(2)[0]
+                .identity_key
+                .to_base58_string();
+            let expected_layer_3_mixnode_pubkey = topology.mixes_in_layer(3)[0]
+                .identity_key
+                .to_base58_string();
             let result = new_v4_with_node(test_node());
             let actual_gateway_key = result.gateways().first().unwrap().identity_key;
             let actual_layer_1_mixnode_pubkey =
-                result.mixes_in_layer(1)[0].pub_key.to_base58_string();
+                result.mixes_in_layer(1)[0].identity_key.to_base58_string();
             let actual_layer_2_mixnode_pubkey =
-                result.mixes_in_layer(2)[0].pub_key.to_base58_string();
+                result.mixes_in_layer(2)[0].identity_key.to_base58_string();
             let actual_layer_3_mixnode_pubkey =
-                result.mixes_in_layer(3)[0].pub_key.to_base58_string();
+                result.mixes_in_layer(3)[0].identity_key.to_base58_string();
 
             assert_eq!(expected_gateway_key, actual_gateway_key);
             assert_eq!(
@@ -204,20 +229,25 @@ mod good_topology_test {
         mixes.insert(1, vec![mix1]);
         mixes.insert(2, vec![mix2]);
         mixes.insert(3, vec![mix3]);
-        NymTopology::new(vec![], mixes, vec![v4_gateway()])
+        NymTopology::new(mixes, vec![v4_gateway()])
     }
 
     fn test_node() -> mix::Node {
         mix::Node {
             location: "Thunder Bay".to_string(),
             host: "1.2.3.4:1234".parse().unwrap(),
-            pub_key: encryption::PublicKey::from_base58_string(
+            identity_key: identity::PublicKey::from_base58_string(
+                "3ebjp1Fb9hdcS1AR6AZihgeJiMHkB5jjJUsvqNnfQwU7",
+            )
+            .unwrap(),
+            sphinx_key: encryption::PublicKey::from_base58_string(
                 "9fX1rMaQdBEzjuv6kT7oyPfEabt73QTM5cfuQ9kaxrRQ",
             )
             .unwrap(),
             layer: 1,
-            last_seen: 1234,
-            version: "0.8.1".to_string(),
+            registration_time: 1234,
+            version: "0.9.0".to_string(),
+            reputation: 42,
         }
     }
 }