Feature/health checker with existing keys

common/clients/provider-client/src/lib.rs

@@ -75,7 +75,7 @@ impl ProviderClient {
         let mut response = Vec::new();
         socket.read_to_end(&mut response).await?;
         if let Err(e) = socket.shutdown(Shutdown::Read) {
-            warn!("failed to close read part of the socket; err = {:?}", e)
+            debug!("failed to close read part of the socket; err = {:?}. It was probably already closed by the provider", e)
         }
 
         Ok(response)

common/crypto/Cargo.toml

@@ -13,3 +13,6 @@ log = "0.4"
 pretty_env_logger = "0.3"
 rand = "0.7.2"
 rand_os = "0.1"
+
+## will be moved to proper dependencies once released
+sphinx = { git = "https://github.com/nymtech/sphinx", rev="8424f4b0933b4a7f64ae828b2edefc5ba43a9e79" }

common/crypto/src/identity/mod.rs

@@ -4,8 +4,9 @@ use crate::encryption::{
 use crate::{encryption, PemStorable};
 use bs58;
 use curve25519_dalek::scalar::Scalar;
+use sphinx::route::DestinationAddressBytes;
 
-pub trait MixnetIdentityKeyPair<Priv, Pub>
+pub trait MixnetIdentityKeyPair<Priv, Pub>: Clone
 where
     Priv: MixnetIdentityPrivateKey,
     Pub: MixnetIdentityPublicKey,
@@ -19,16 +20,20 @@ where
 }
 
 pub trait MixnetIdentityPublicKey:
-    Sized + PemStorable + for<'a> From<&'a <Self as MixnetIdentityPublicKey>::PrivateKeyMaterial>
+    Sized
+    + PemStorable
+    + Clone
+    + for<'a> From<&'a <Self as MixnetIdentityPublicKey>::PrivateKeyMaterial>
 {
     // we need to couple public and private keys together
     type PrivateKeyMaterial: MixnetIdentityPrivateKey<PublicKeyMaterial = Self>;
 
+    fn derive_address(&self) -> DestinationAddressBytes;
     fn to_bytes(&self) -> Vec<u8>;
     fn from_bytes(b: &[u8]) -> Self;
 }
 
-pub trait MixnetIdentityPrivateKey: Sized + PemStorable {
+pub trait MixnetIdentityPrivateKey: Sized + PemStorable + Clone {
     // we need to couple public and private keys together
     type PublicKeyMaterial: MixnetIdentityPublicKey<PrivateKeyMaterial = Self>;
 
@@ -45,7 +50,7 @@ pub trait MixnetIdentityPrivateKey: Sized + PemStorable {
 
 // for time being define a dummy identity using x25519 encryption keys (as we've done so far)
 // and replace it with proper keys, like ed25519 later on
-
+#[derive(Clone)]
 pub struct DummyMixIdentityKeyPair {
     pub private_key: DummyMixIdentityPrivateKey,
     pub public_key: DummyMixIdentityPublicKey,
@@ -84,6 +89,14 @@ pub struct DummyMixIdentityPublicKey(encryption::x25519::PublicKey);
 impl MixnetIdentityPublicKey for DummyMixIdentityPublicKey {
     type PrivateKeyMaterial = DummyMixIdentityPrivateKey;
 
+    fn derive_address(&self) -> DestinationAddressBytes {
+        let mut temporary_address = [0u8; 32];
+        let public_key_bytes = self.to_bytes();
+        temporary_address.copy_from_slice(&public_key_bytes[..]);
+
+        temporary_address
+    }
+
     fn to_bytes(&self) -> Vec<u8> {
         self.0.to_bytes()
     }

common/healthcheck/Cargo.toml

@@ -11,6 +11,8 @@ futures = "0.3.1"
 itertools = "0.8.2"
 log = "0.4.8"
 pretty_env_logger = "0.3"
+rand_os = "0.1"
+rand = "0.7.2"
 serde = "1.0.104"
 serde_derive = "1.0.104"
 tokio = { version = "0.2", features = ["full"] }

common/healthcheck/src/lib.rs

@@ -1,8 +1,10 @@
 use crate::result::HealthCheckResult;
+use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPrivateKey, MixnetIdentityPublicKey};
 use directory_client::requests::presence_topology_get::PresenceTopologyGetRequester;
 use directory_client::DirectoryClient;
 use log::{debug, error, info, trace};
 use std::fmt::{Error, Formatter};
+use std::marker::PhantomData;
 use std::time::Duration;
 use topology::NymTopologyError;
 
@@ -34,15 +36,29 @@ impl From<topology::NymTopologyError> for HealthCheckerError {
     }
 }
 
-pub struct HealthChecker {
+pub struct HealthChecker<IDPair, Priv, Pub>
+where
+    IDPair: MixnetIdentityKeyPair<Priv, Pub>,
+    Priv: MixnetIdentityPrivateKey,
+    Pub: MixnetIdentityPublicKey,
+{
     directory_client: directory_client::Client,
     interval: Duration,
     num_test_packets: usize,
     resolution_timeout: Duration,
+    identity_keypair: IDPair,
+
+    _phantom_private: PhantomData<Priv>,
+    _phantom_public: PhantomData<Pub>,
 }
 
-impl HealthChecker {
-    pub fn new(config: config::HealthCheck) -> Self {
+impl<IDPair, Priv, Pub> HealthChecker<IDPair, Priv, Pub>
+where
+    IDPair: crypto::identity::MixnetIdentityKeyPair<Priv, Pub>,
+    Priv: crypto::identity::MixnetIdentityPrivateKey,
+    Pub: crypto::identity::MixnetIdentityPublicKey,
+{
+    pub fn new(config: config::HealthCheck, identity_keypair: IDPair) -> Self {
         debug!(
             "healthcheck will be using the following directory server: {:?}",
             config.directory_server
@@ -53,11 +69,16 @@ impl HealthChecker {
             interval: Duration::from_secs_f64(config.interval),
             resolution_timeout: Duration::from_secs_f64(config.resolution_timeout),
             num_test_packets: config.num_test_packets,
+            identity_keypair,
+
+            _phantom_private: PhantomData,
+            _phantom_public: PhantomData,
         }
     }
 
     pub async fn do_check(&self) -> Result<HealthCheckResult, HealthCheckerError> {
         trace!("going to perform a healthcheck!");
+
         let current_topology = match self.directory_client.presence_topology.get() {
             Ok(topology) => topology,
             Err(err) => {
@@ -71,6 +92,7 @@ impl HealthChecker {
             current_topology,
             self.num_test_packets,
             self.resolution_timeout,
+            &self.identity_keypair,
         )
         .await;
         healthcheck_result.sort_scores();

common/healthcheck/src/path_check.rs

@@ -1,8 +1,8 @@
-use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey};
+use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPrivateKey, MixnetIdentityPublicKey};
 use itertools::Itertools;
 use log::{debug, error, info, trace, warn};
 use mix_client::MixClient;
-use provider_client::ProviderClient;
+use provider_client::{ProviderClient, ProviderClientError};
 use sphinx::header::delays::Delay;
 use sphinx::route::{Destination, Node as SphinxNode};
 use std::collections::HashMap;
@@ -23,28 +23,36 @@ pub(crate) struct PathChecker {
     layer_one_clients: HashMap<[u8; 32], Option<MixClient>>,
     paths_status: HashMap<Vec<u8>, PathStatus>,
     our_destination: Destination,
+    check_id: [u8; 16],
 }
 
 impl PathChecker {
-    pub(crate) async fn new(
+    pub(crate) async fn new<IDPair, Priv, Pub>(
         providers: Vec<provider::Node>,
-        ephemeral_keys: DummyMixIdentityKeyPair,
-    ) -> Self {
+        identity_keys: &IDPair,
+        check_id: [u8; 16],
+    ) -> Self
+    where
+        IDPair: MixnetIdentityKeyPair<Priv, Pub>,
+        Priv: MixnetIdentityPrivateKey,
+        Pub: MixnetIdentityPublicKey,
+    {
         let mut provider_clients = HashMap::new();
 
-        let mut temporary_address = [0u8; 32];
-        let public_key_bytes = ephemeral_keys.public_key().to_bytes();
-        temporary_address.copy_from_slice(&public_key_bytes[..]);
+        let address = identity_keys.public_key().derive_address();
 
         for provider in providers {
-            let mut provider_client =
-                ProviderClient::new(provider.client_listener, temporary_address, None);
+            let mut provider_client = ProviderClient::new(provider.client_listener, address, None);
             let insertion_result = match provider_client.register().await {
                 Ok(token) => {
                     debug!("registered at provider {}", provider.pub_key);
                     provider_client.update_token(token);
                     provider_clients.insert(provider.get_pub_key_bytes(), Some(provider_client))
                 }
+                Err(ProviderClientError::ClientAlreadyRegisteredError) => {
+                    info!("We were already registered");
+                    provider_clients.insert(provider.get_pub_key_bytes(), Some(provider_client))
+                }
                 Err(err) => {
                     warn!(
                         "failed to register at provider {} - {:?}",
@@ -62,15 +70,19 @@ impl PathChecker {
         PathChecker {
             provider_clients,
             layer_one_clients: HashMap::new(),
-            our_destination: Destination::new(temporary_address, Default::default()),
+            our_destination: Destination::new(address, Default::default()),
             paths_status: HashMap::new(),
+            check_id,
         }
     }
 
     // iteration is used to distinguish packets sent through the same path (as the healthcheck
     // may try to send say 10 packets through given path)
-    fn unique_path_key(path: &Vec<SphinxNode>, iteration: u8) -> Vec<u8> {
-        std::iter::once(iteration)
+    fn unique_path_key(path: &Vec<SphinxNode>, check_id: [u8; 16], iteration: u8) -> Vec<u8> {
+        check_id
+            .iter()
+            .cloned()
+            .chain(std::iter::once(iteration))
             .chain(
                 path.iter()
                     .map(|node| node.pub_key.to_bytes().to_vec())
@@ -80,10 +92,10 @@ impl PathChecker {
     }
 
     pub(crate) fn path_key_to_node_keys(path_key: Vec<u8>) -> Vec<[u8; 32]> {
-        assert_eq!(path_key.len() % 32, 1);
+        assert_eq!(path_key.len() % 32, 17);
         path_key
             .into_iter()
-            .skip(1) // remove first byte as it represents the iteration number which we do not care about now
+            .skip(16 + 1) // remove 16 + 1 bytes as it represents check_id and the iteration number which we do not care about now
             .chunks(32)
             .into_iter()
             .map(|key_chunk| {
@@ -135,6 +147,8 @@ impl PathChecker {
                         if msg == sfw_provider_requests::DUMMY_MESSAGE_CONTENT {
                             // finish iterating the loop as the messages might not be ordered
                             should_stop = true;
+                        } else if msg[..16] != self.check_id {
+                            warn!("received response from previous healthcheck")
                         } else {
                             provider_messages.push(msg);
                         }
@@ -171,7 +185,7 @@ impl PathChecker {
         }
 
         debug!("Checking path: {:?} ({})", path, iteration);
-        let path_identifier = PathChecker::unique_path_key(path, iteration);
+        let path_identifier = PathChecker::unique_path_key(path, self.check_id, iteration);
 
         // check if there is even any point in sending the packet
 

common/healthcheck/src/result.rs

@@ -1,7 +1,8 @@
 use crate::path_check::{PathChecker, PathStatus};
 use crate::score::NodeScore;
-use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair};
+use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPrivateKey, MixnetIdentityPublicKey};
 use log::{debug, error, info, warn};
+use rand_os::rand_core::RngCore;
 use sphinx::route::NodeAddressBytes;
 use std::collections::HashMap;
 use std::fmt::{Error, Formatter};
@@ -94,15 +95,31 @@ impl HealthCheckResult {
         )
     }
 
-    pub async fn calculate<T: NymTopology>(
+    fn generate_check_id() -> [u8; 16] {
+        let mut id = [0u8; 16];
+        let mut rng = rand_os::OsRng::new().unwrap();
+        rng.fill_bytes(&mut id);
+        id
+    }
+
+    pub async fn calculate<T, IDPair, Priv, Pub>(
         topology: T,
         iterations: usize,
         resolution_timeout: Duration,
-    ) -> Self {
+        identity_keys: &IDPair,
+    ) -> Self
+    where
+        T: NymTopology,
+        IDPair: MixnetIdentityKeyPair<Priv, Pub>,
+        Priv: MixnetIdentityPrivateKey,
+        Pub: MixnetIdentityPublicKey,
+    {
         // currently healthchecker supports only up to 255 iterations - if we somehow
         // find we need more, it's relatively easy change
         assert!(iterations <= 255);
 
+        let check_id = Self::generate_check_id();
+
         let all_paths = match topology.all_paths() {
             Ok(paths) => paths,
             Err(_) => return Self::zero_score(topology),
@@ -118,10 +135,9 @@ impl HealthCheckResult {
             score_map.insert(node.get_pub_key_bytes(), NodeScore::from_provider(node));
         });
 
-        let ephemeral_keys = DummyMixIdentityKeyPair::new();
         let providers = topology.providers();
 
-        let mut path_checker = PathChecker::new(providers, ephemeral_keys).await;
+        let mut path_checker = PathChecker::new(providers, identity_keys, check_id).await;
         for i in 0..iterations {
             debug!("running healthcheck iteration {} / {}", i + 1, iterations);
             for path in &all_paths {