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

More generic nym-client with regards to identity keypair

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2020-01-27 17:10:53 +00:00
parent c662dad57f
commit 52aede3d99
4 changed files with 94 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-client/src/client/mod.rs
+38 -16
View File
@@ -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<IDPair, Priv, Pub>
where
IDPair: MixnetIdentityKeyPair<Priv, Pub> + 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<InputMessage>,
@@ -47,14 +53,22 @@ pub struct NymClient {
directory: String,
auth_token: Option<AuthToken>,
socket_type: SocketType,
_phantom_private: PhantomData<Priv>,
_phantom_public: PhantomData<Pub>,
}
#[derive(Debug)]
pub struct InputMessage(pub Destination, pub Vec<u8>);
impl NymClient {
impl<IDPair: 'static, Priv: 'static, Pub: 'static> NymClient<IDPair, Priv, Pub>
where
IDPair: MixnetIdentityKeyPair<Priv, Pub> + 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<AuthToken>,
@@ -63,13 +77,15 @@ impl NymClient {
let (input_tx, input_rx) = mpsc::unbounded::<InputMessage>();
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::<Topology>::new(
self.directory.clone(),
TOPOLOGY_REFRESH_RATE,
));
let topology_controller =
rt.block_on(topology_control::TopologyControl::<Topology, _, _, _>::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(),
));
}
nym-client/src/client/topology_control.rs
+52 -28
View File
@@ -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<T> = Arc<FRwLock<Inner<T>>>;
pub(crate) struct TopologyControl<T: NymTopology> {
pub(crate) struct TopologyControl<T, IDPair, Priv, Pub>
where
T: NymTopology,
IDPair: MixnetIdentityKeyPair<Priv, Pub>,
Priv: MixnetIdentityPrivateKey,
Pub: MixnetIdentityPublicKey,
{
directory_server: String,
refresh_rate: f64,
inner: Arc<FRwLock<Inner<T>>>,
health_checker: HealthChecker<IDPair, Priv, Pub>,
refresh_rate: f64,
}
#[derive(Debug)]
@@ -22,40 +31,56 @@ enum TopologyError {
NoValidPathsError,
}
impl<T: NymTopology> TopologyControl<T> {
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<T, IDPair, Priv, Pub> TopologyControl<T, IDPair, Priv, Pub>
where
T: NymTopology,
IDPair: MixnetIdentityKeyPair<Priv, Pub>,
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<T, TopologyError> {
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<T, TopologyError> {
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<T: NymTopology> TopologyControl<T> {
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),
nym-client/src/commands/tcpsocket.rs
+2 -5
View File
@@ -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,
nym-client/src/commands/websocket.rs
+2 -5
View File
@@ -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,