From 52aede3d9986b83ec452b8ee925f4d561ef7e27b Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Mon, 27 Jan 2020 17:10:53 +0000 Subject: [PATCH] More generic nym-client with regards to identity keypair --- nym-client/src/client/mod.rs | 54 ++++++++++----- nym-client/src/client/topology_control.rs | 80 +++++++++++++++-------- nym-client/src/commands/tcpsocket.rs | 7 +- nym-client/src/commands/websocket.rs | 7 +- 4 files changed, 94 insertions(+), 54 deletions(-) diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index c2435acf0d..4d4e0da408 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -3,12 +3,14 @@ use crate::client::received_buffer::ReceivedMessagesBuffer; use crate::client::topology_control::TopologyInnerRef; use crate::sockets::tcp; use crate::sockets::ws; +use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPrivateKey, MixnetIdentityPublicKey}; use directory_client::presence::Topology; use futures::channel::mpsc; use futures::join; use log::*; use sfw_provider_requests::AuthToken; -use sphinx::route::{Destination, DestinationAddressBytes}; +use sphinx::route::Destination; +use std::marker::PhantomData; use std::net::SocketAddr; use tokio::runtime::Runtime; use topology::NymTopology; @@ -35,9 +37,13 @@ pub enum SocketType { None, } -pub struct NymClient { - // to be replaced by something else I guess - address: DestinationAddressBytes, +pub struct NymClient +where + IDPair: MixnetIdentityKeyPair + Send + Sync, + Priv: MixnetIdentityPrivateKey + Send + Sync, + Pub: MixnetIdentityPublicKey + Send + Sync, +{ + keypair: IDPair, // to be used by "send" function or socket, etc pub input_tx: mpsc::UnboundedSender, @@ -47,14 +53,22 @@ pub struct NymClient { directory: String, auth_token: Option, socket_type: SocketType, + + _phantom_private: PhantomData, + _phantom_public: PhantomData, } #[derive(Debug)] pub struct InputMessage(pub Destination, pub Vec); -impl NymClient { +impl NymClient +where + IDPair: MixnetIdentityKeyPair + Send + Sync, + Priv: MixnetIdentityPrivateKey + Send + Sync, + Pub: MixnetIdentityPublicKey + Send + Sync, +{ pub fn new( - address: DestinationAddressBytes, + keypair: IDPair, socket_listening_address: SocketAddr, directory: String, auth_token: Option, @@ -63,13 +77,15 @@ impl NymClient { let (input_tx, input_rx) = mpsc::unbounded::(); NymClient { - address, + keypair, input_tx, input_rx, socket_listening_address, directory, auth_token, socket_type, + _phantom_private: PhantomData, + _phantom_public: PhantomData, } } @@ -106,11 +122,18 @@ impl NymClient { let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) = mpsc::unbounded(); + let self_address = self.keypair.public_key().derive_address(); + + // generate same type of keys we have as our identity + let healthcheck_keys = IDPair::new(); + // TODO: when we switch to our graph topology, we need to remember to change 'Topology' type - let topology_controller = rt.block_on(topology_control::TopologyControl::::new( - self.directory.clone(), - TOPOLOGY_REFRESH_RATE, - )); + let topology_controller = + rt.block_on(topology_control::TopologyControl::::new( + self.directory.clone(), + TOPOLOGY_REFRESH_RATE, + healthcheck_keys, + )); let provider_client_listener_address = rt.block_on(self.get_provider_socket_address(topology_controller.get_inner_ref())); @@ -118,7 +141,7 @@ impl NymClient { let mut provider_poller = provider_poller::ProviderPoller::new( poller_input_tx, provider_client_listener_address, - self.address, + self_address, self.auth_token, ); @@ -144,12 +167,11 @@ impl NymClient { let loop_cover_traffic_future = rt.spawn(cover_traffic_stream::start_loop_cover_traffic_stream( mix_tx.clone(), - Destination::new(self.address, Default::default()), + Destination::new(self_address, Default::default()), topology_controller.get_inner_ref(), )); // cloning arguments required by OutQueueControl; required due to move - let self_address = self.address; let input_rx = self.input_rx; let topology_ref = topology_controller.get_inner_ref(); @@ -180,7 +202,7 @@ impl NymClient { self.socket_listening_address, self.input_tx, received_messages_buffer_output_tx, - self.address, + self_address, topology_controller.get_inner_ref(), )); } @@ -189,7 +211,7 @@ impl NymClient { self.socket_listening_address, self.input_tx, received_messages_buffer_output_tx, - self.address, + self_address, topology_controller.get_inner_ref(), )); } diff --git a/nym-client/src/client/topology_control.rs b/nym-client/src/client/topology_control.rs index 426c6be821..9daafc0572 100644 --- a/nym-client/src/client/topology_control.rs +++ b/nym-client/src/client/topology_control.rs @@ -1,4 +1,6 @@ use crate::built_info; +use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPrivateKey, MixnetIdentityPublicKey}; +use healthcheck::HealthChecker; use log::{error, info, trace, warn}; use std::sync::Arc; use std::time::Duration; @@ -10,10 +12,17 @@ const NODE_HEALTH_THRESHOLD: f64 = 0.0; // auxiliary type for ease of use pub type TopologyInnerRef = Arc>>; -pub(crate) struct TopologyControl { +pub(crate) struct TopologyControl +where + T: NymTopology, + IDPair: MixnetIdentityKeyPair, + Priv: MixnetIdentityPrivateKey, + Pub: MixnetIdentityPublicKey, +{ directory_server: String, - refresh_rate: f64, inner: Arc>>, + health_checker: HealthChecker, + refresh_rate: f64, } #[derive(Debug)] @@ -22,40 +31,56 @@ enum TopologyError { NoValidPathsError, } -impl TopologyControl { - pub(crate) async fn new(directory_server: String, refresh_rate: f64) -> Self { - let initial_topology = match Self::get_current_compatible_topology(directory_server.clone()) - .await - { +impl TopologyControl +where + T: NymTopology, + IDPair: MixnetIdentityKeyPair, + Priv: MixnetIdentityPrivateKey, + Pub: MixnetIdentityPublicKey, +{ + pub(crate) async fn new( + directory_server: String, + refresh_rate: f64, + identity_keypair: IDPair, + ) -> Self { + // topology control run a healthcheck to determine healthy-ish nodes: + // this is a temporary solution as the healthcheck will eventually be moved to validators + + let healthcheck_config = healthcheck::config::HealthCheck { + directory_server: directory_server.clone(), + // those are literally irrelevant when running single check + interval: 100000.0, + resolution_timeout: 5.0, + num_test_packets: 2, + }; + let health_checker = healthcheck::HealthChecker::new(healthcheck_config, identity_keypair); + + let mut topology_control = TopologyControl { + directory_server, + refresh_rate, + inner: Arc::new(FRwLock::new(Inner::new(None))), + health_checker, + }; + + // best effort approach to try to get a valid topology after call to 'new' + let initial_topology = match topology_control.get_current_compatible_topology().await { Ok(topology) => Some(topology), Err(err) => { error!("Initial topology is invalid - {:?}. Right now it will be impossible to send any packets through the mixnet!", err); None } }; + topology_control + .update_global_topology(initial_topology) + .await; - TopologyControl { - directory_server, - refresh_rate, - inner: Arc::new(FRwLock::new(Inner::new(initial_topology))), - } + topology_control } - async fn get_current_compatible_topology(directory_server: String) -> Result { - let full_topology = T::new(directory_server.clone()); - - // run a healthcheck to determine healthy-ish nodes: - // this is a temporary solution as the healthcheck will eventually be moved to validators - let healthcheck_config = healthcheck::config::HealthCheck { - directory_server, - // those are literally irrelevant when running single check - interval: 100000.0, - resolution_timeout: 5.0, - num_test_packets: 2, - }; - let healthcheck = healthcheck::HealthChecker::new(healthcheck_config); - let healthcheck_result = healthcheck.do_check().await; + async fn get_current_compatible_topology(&self) -> Result { + let full_topology = T::new(self.directory_server.clone()); + let healthcheck_result = self.health_checker.do_check().await; let healthcheck_scores = match healthcheck_result { Err(err) => { error!("Error while performing the healthcheck: {:?}", err); @@ -96,8 +121,7 @@ impl TopologyControl { let delay_duration = Duration::from_secs_f64(self.refresh_rate); loop { trace!("Refreshing the topology"); - let new_topology_res = - Self::get_current_compatible_topology(self.directory_server.clone()).await; + let new_topology_res = self.get_current_compatible_topology().await; let new_topology = match new_topology_res { Ok(topology) => Some(topology), diff --git a/nym-client/src/commands/tcpsocket.rs b/nym-client/src/commands/tcpsocket.rs index a050e78832..7a98a91681 100644 --- a/nym-client/src/commands/tcpsocket.rs +++ b/nym-client/src/commands/tcpsocket.rs @@ -1,7 +1,7 @@ use crate::client::{NymClient, SocketType}; use crate::config::persistance::pathfinder::ClientPathfinder; use clap::ArgMatches; -use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey}; +use crypto::identity::DummyMixIdentityKeyPair; use pemstore::pemstore::PemStore; use std::net::ToSocketAddrs; @@ -34,12 +34,9 @@ pub fn execute(matches: &ArgMatches) { println!("Public key: {}", keypair.public_key.to_base58_string()); - let mut temporary_address = [0u8; 32]; - let public_key_bytes = keypair.public_key().to_bytes(); - temporary_address.copy_from_slice(&public_key_bytes[..]); let auth_token = None; let client = NymClient::new( - temporary_address, + keypair, socket_address.clone(), directory_server, auth_token, diff --git a/nym-client/src/commands/websocket.rs b/nym-client/src/commands/websocket.rs index 82437514a2..3193c1ffde 100644 --- a/nym-client/src/commands/websocket.rs +++ b/nym-client/src/commands/websocket.rs @@ -1,7 +1,7 @@ use crate::client::{NymClient, SocketType}; use crate::config::persistance::pathfinder::ClientPathfinder; use clap::ArgMatches; -use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey}; +use crypto::identity::DummyMixIdentityKeyPair; use pemstore::pemstore::PemStore; use std::net::ToSocketAddrs; @@ -35,12 +35,9 @@ pub fn execute(matches: &ArgMatches) { println!("Public key: {}", keypair.public_key.to_base58_string()); - let mut temporary_address = [0u8; 32]; - let public_key_bytes = keypair.public_key().to_bytes(); - temporary_address.copy_from_slice(&public_key_bytes[..]); let auth_token = None; let client = NymClient::new( - temporary_address, + keypair, socket_address, directory_server, auth_token,