diff --git a/Cargo.lock b/Cargo.lock index 86b41fd545..ba55ec418a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2104,7 +2104,7 @@ dependencies = [ [[package]] name = "sphinx" version = "0.1.0" -source = "git+https://github.com/nymtech/sphinx?rev=1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f#1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" +source = "git+https://github.com/nymtech/sphinx?rev=c3ba1447d61d6eb900fcdea0c20b3070a22bf11a#c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" dependencies = [ "aes-ctr", "arrayref", @@ -2116,6 +2116,7 @@ dependencies = [ "hkdf", "hmac", "lioness", + "log", "rand 0.7.2", "rand_distr", "rand_os", diff --git a/README.md b/README.md index d43dd40a96..82767c0049 100644 --- a/README.md +++ b/README.md @@ -4,8 +4,8 @@ This repository contains the full Nym platform, written in Rust. The platform is composed of multiple Rust crates. Top-level executable binary crates include: -* client - an executable which you can build into your own applications. Use it for interacting with Nym nodes. -* mixnode - the mixnode crate. +* mixnode - shuffles [Sphinx](https://github.com/nymtech/sphinx) packets together to provide privacy against network-level attackers. +* nym-client - an executable which you can build into your own applications. Use it for interacting with Nym nodes. * sfw-provider - a store-and-forward service provider. The provider acts sort of like a mailbox for mixnet messages. * validator - currently just starting development. Handles consensus ordering of transactions, mixmining, and coconut credential generation and validation. diff --git a/common/clients/directory-client/src/presence.rs b/common/clients/directory-client/src/presence.rs index 8be3c73b9e..f56305f172 100644 --- a/common/clients/directory-client/src/presence.rs +++ b/common/clients/directory-client/src/presence.rs @@ -2,10 +2,20 @@ use crate::requests::presence_topology_get::PresenceTopologyGetRequester; use crate::{Client, Config, DirectoryClient}; use log::*; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; use std::convert::TryInto; use std::io; use std::net::ToSocketAddrs; -use topology::{CocoNode, MixNode, MixProviderNode, NymTopology}; +use topology::coco; +use topology::mix; +use topology::provider; +use topology::NymTopology; + +// special version of 'PartialEq' that does not care about last_seen field (where applicable) +// or order of elements in vectors +trait PresenceEq { + fn presence_eq(&self, other: &Self) -> bool; +} #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] @@ -16,9 +26,9 @@ pub struct CocoPresence { pub version: String, } -impl Into for CocoPresence { - fn into(self) -> topology::CocoNode { - topology::CocoNode { +impl Into for CocoPresence { + fn into(self) -> topology::coco::Node { + topology::coco::Node { host: self.host, pub_key: self.pub_key, last_seen: self.last_seen, @@ -27,8 +37,8 @@ impl Into for CocoPresence { } } -impl From for CocoPresence { - fn from(cn: CocoNode) -> Self { +impl From for CocoPresence { + fn from(cn: coco::Node) -> Self { CocoPresence { host: cn.host, pub_key: cn.pub_key, @@ -48,10 +58,10 @@ pub struct MixNodePresence { pub version: String, } -impl TryInto for MixNodePresence { +impl TryInto for MixNodePresence { type Error = io::Error; - fn try_into(self) -> Result { + fn try_into(self) -> Result { let resolved_hostname = self.host.to_socket_addrs()?.next(); if resolved_hostname.is_none() { return Err(io::Error::new( @@ -60,7 +70,7 @@ impl TryInto for MixNodePresence { )); } - Ok(topology::MixNode { + Ok(topology::mix::Node { host: resolved_hostname.unwrap(), pub_key: self.pub_key, layer: self.layer, @@ -70,8 +80,8 @@ impl TryInto for MixNodePresence { } } -impl From for MixNodePresence { - fn from(mn: MixNode) -> Self { +impl From for MixNodePresence { + fn from(mn: mix::Node) -> Self { MixNodePresence { host: mn.host.to_string(), pub_key: mn.pub_key, @@ -93,9 +103,35 @@ pub struct MixProviderPresence { pub version: String, } -impl Into for MixProviderPresence { - fn into(self) -> topology::MixProviderNode { - topology::MixProviderNode { +impl PresenceEq for MixProviderPresence { + fn presence_eq(&self, other: &Self) -> bool { + if self.registered_clients.len() != other.registered_clients.len() { + return false; + } + + if self.client_listener != other.client_listener + || self.mixnet_listener != other.mixnet_listener + || self.pub_key != other.pub_key + || self.version != other.version + { + return false; + } + + let clients_self_set: HashSet<_> = + self.registered_clients.iter().map(|c| &c.pub_key).collect(); + let clients_other_set: HashSet<_> = other + .registered_clients + .iter() + .map(|c| &c.pub_key) + .collect(); + + clients_self_set == clients_other_set + } +} + +impl Into for MixProviderPresence { + fn into(self) -> topology::provider::Node { + topology::provider::Node { client_listener: self.client_listener.parse().unwrap(), mixnet_listener: self.mixnet_listener.parse().unwrap(), pub_key: self.pub_key, @@ -110,8 +146,8 @@ impl Into for MixProviderPresence { } } -impl From for MixProviderPresence { - fn from(mpn: MixProviderNode) -> Self { +impl From for MixProviderPresence { + fn from(mpn: provider::Node) -> Self { MixProviderPresence { client_listener: mpn.client_listener.to_string(), mixnet_listener: mpn.mixnet_listener.to_string(), @@ -133,22 +169,116 @@ pub struct MixProviderClient { pub pub_key: String, } -impl Into for MixProviderClient { - fn into(self) -> topology::MixProviderClient { - topology::MixProviderClient { +impl Into for MixProviderClient { + fn into(self) -> topology::provider::Client { + topology::provider::Client { pub_key: self.pub_key, } } } -impl From for MixProviderClient { - fn from(mpc: topology::MixProviderClient) -> Self { +impl From for MixProviderClient { + fn from(mpc: topology::provider::Client) -> Self { MixProviderClient { pub_key: mpc.pub_key, } } } +impl PresenceEq for Vec { + fn presence_eq(&self, other: &Self) -> bool { + if self.len() != other.len() { + return false; + } + + // we can't take the whole thing into set as it does not implement 'Eq' and we can't + // derive it as we don't want to take 'last_seen' into consideration + let self_set: HashSet<_> = self + .iter() + .map(|c| (&c.host, &c.pub_key, &c.version)) + .collect(); + let other_set: HashSet<_> = other + .iter() + .map(|c| (&c.host, &c.pub_key, &c.version)) + .collect(); + + self_set == other_set + } +} + +impl PresenceEq for Vec { + fn presence_eq(&self, other: &Self) -> bool { + if self.len() != other.len() { + return false; + } + + // we can't take the whole thing into set as it does not implement 'Eq' and we can't + // derive it as we don't want to take 'last_seen' into consideration + let self_set: HashSet<_> = self + .iter() + .map(|m| (&m.host, &m.pub_key, &m.version, &m.layer)) + .collect(); + let other_set: HashSet<_> = other + .iter() + .map(|m| (&m.host, &m.pub_key, &m.version, &m.layer)) + .collect(); + + self_set == other_set + } +} + +impl PresenceEq for Vec { + fn presence_eq(&self, other: &Self) -> bool { + if self.len() != other.len() { + return false; + } + + // we can't take the whole thing into set as it does not implement 'Eq' and we can't + // derive it as we don't want to take 'last_seen' into consideration. + // We also don't care about order of registered_clients + + // since we're going to be getting rid of this very soon anyway, just clone registered + // clients vector and sort it + + let self_set: HashSet<_> = self + .iter() + .map(|p| { + ( + &p.client_listener, + &p.mixnet_listener, + &p.pub_key, + &p.version, + p.registered_clients + .iter() + .cloned() + .map(|c| c.pub_key) + .collect::>() + .sort(), + ) + }) + .collect(); + let other_set: HashSet<_> = other + .iter() + .map(|p| { + ( + &p.client_listener, + &p.mixnet_listener, + &p.pub_key, + &p.version, + p.registered_clients + .iter() + .cloned() + .map(|c| c.pub_key) + .collect::>() + .sort(), + ) + }) + .collect(); + + self_set == other_set + } +} + // Topology shows us the current state of the overall Nym network #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] @@ -158,6 +288,23 @@ pub struct Topology { pub mix_provider_nodes: Vec, } +impl PartialEq for Topology { + // we need a custom implementation as the order of nodes does not matter + // also we do not care about 'last_seen' when comparing topologies + fn eq(&self, other: &Self) -> bool { + if !self.coco_nodes.presence_eq(&other.coco_nodes) + || !self.mix_nodes.presence_eq(&other.mix_nodes) + || !self + .mix_provider_nodes + .presence_eq(&other.mix_provider_nodes) + { + return false; + } + + true + } +} + impl NymTopology for Topology { fn new(directory_server: String) -> Self { debug!("Using directory server: {:?}", directory_server); @@ -174,9 +321,9 @@ impl NymTopology for Topology { } fn new_from_nodes( - mix_nodes: Vec, - mix_provider_nodes: Vec, - coco_nodes: Vec, + mix_nodes: Vec, + mix_provider_nodes: Vec, + coco_nodes: Vec, ) -> Self { Topology { coco_nodes: coco_nodes.into_iter().map(|node| node.into()).collect(), @@ -188,21 +335,21 @@ impl NymTopology for Topology { } } - fn get_mix_nodes(&self) -> Vec { + fn mix_nodes(&self) -> Vec { self.mix_nodes .iter() .filter_map(|x| x.clone().try_into().ok()) .collect() } - fn get_mix_provider_nodes(&self) -> Vec { + fn providers(&self) -> Vec { self.mix_provider_nodes .iter() .map(|x| x.clone().into()) .collect() } - fn get_coco_nodes(&self) -> Vec { + fn coco_nodes(&self) -> Vec { self.coco_nodes.iter().map(|x| x.clone().into()).collect() } } @@ -223,7 +370,7 @@ mod converting_mixnode_presence_into_topology_mixnode { version: "".to_string(), }; - let result: Result = mix_presence.try_into(); + let result: Result = mix_presence.try_into(); assert!(result.is_err()) } @@ -239,7 +386,7 @@ mod converting_mixnode_presence_into_topology_mixnode { version: "".to_string(), }; - let result: Result = mix_presence.try_into(); + let result: Result = mix_presence.try_into(); assert!(result.is_ok()) } } diff --git a/common/clients/mix-client/Cargo.toml b/common/clients/mix-client/Cargo.toml index a7ef6e35c8..dcd16d33d5 100644 --- a/common/clients/mix-client/Cargo.toml +++ b/common/clients/mix-client/Cargo.toml @@ -18,5 +18,6 @@ addressing = {path = "../../addressing"} topology = {path = "../../topology"} ## will be moved to proper dependencies once released -sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" } +sphinx = { git = "https://github.com/nymtech/sphinx", rev="c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" } +# sphinx = { path = "../../../../sphinx"} diff --git a/common/clients/mix-client/src/packet.rs b/common/clients/mix-client/src/packet.rs index cc1a027cb6..7999d1e620 100644 --- a/common/clients/mix-client/src/packet.rs +++ b/common/clients/mix-client/src/packet.rs @@ -28,7 +28,7 @@ pub fn encapsulate_message( topology: &T, average_delay: f64, ) -> (SocketAddr, SphinxPacket) { - let mut providers = topology.get_mix_provider_nodes(); + let mut providers = topology.providers(); let provider = providers.pop().unwrap().into(); let route = topology.route_to(provider).unwrap(); diff --git a/common/clients/provider-client/Cargo.toml b/common/clients/provider-client/Cargo.toml index 450a76cb2b..1c1d87b7b1 100644 --- a/common/clients/provider-client/Cargo.toml +++ b/common/clients/provider-client/Cargo.toml @@ -16,4 +16,4 @@ tokio = { version = "0.2", features = ["full"] } sfw-provider-requests = { path = "../../../sfw-provider/sfw-provider-requests" } ## will be moved to proper dependencies once released -sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" } +sphinx = { git = "https://github.com/nymtech/sphinx", rev="c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" } diff --git a/common/healthcheck/Cargo.toml b/common/healthcheck/Cargo.toml index 918efa0df9..aaacb88d66 100644 --- a/common/healthcheck/Cargo.toml +++ b/common/healthcheck/Cargo.toml @@ -25,6 +25,7 @@ sfw-provider-requests = { path = "../../sfw-provider/sfw-provider-requests" } topology = {path = "../topology" } ## will be moved to proper dependencies once released -sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" } +sphinx = { git = "https://github.com/nymtech/sphinx", rev="c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" } +# sphinx = { path = "../../../sphinx"} [dev-dependencies] diff --git a/common/healthcheck/src/path_check.rs b/common/healthcheck/src/path_check.rs index 62b94a86b1..f2d8b4879b 100644 --- a/common/healthcheck/src/path_check.rs +++ b/common/healthcheck/src/path_check.rs @@ -1,12 +1,12 @@ use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey}; use itertools::Itertools; -use log::{debug, error, trace, warn}; +use log::{debug, error, info, trace, warn}; use mix_client::MixClient; use provider_client::ProviderClient; use sphinx::header::delays::Delay; use sphinx::route::{Destination, Node as SphinxNode}; use std::collections::HashMap; -use topology::MixProviderNode; +use topology::provider; #[derive(Debug, PartialEq, Clone)] pub enum PathStatus { @@ -27,7 +27,7 @@ pub(crate) struct PathChecker { impl PathChecker { pub(crate) async fn new( - providers: Vec, + providers: Vec, ephemeral_keys: DummyMixIdentityKeyPair, ) -> Self { let mut provider_clients = HashMap::new(); @@ -223,7 +223,7 @@ impl PathChecker { debug!("sending test packet to {}", first_node_address); match first_node_client.send(packet, first_node_address).await { Err(err) => { - warn!("failed to send packet to {} - {}", first_node_address, err); + info!("failed to send packet to {} - {}", first_node_address, err); if self .paths_status .insert(path_identifier, PathStatus::Unhealthy) diff --git a/common/healthcheck/src/result.rs b/common/healthcheck/src/result.rs index 79ec6a291f..b7aae8d33c 100644 --- a/common/healthcheck/src/result.rs +++ b/common/healthcheck/src/result.rs @@ -28,8 +28,8 @@ impl HealthCheckResult { fn zero_score(topology: T) -> Self { warn!("The network is unhealthy, could not send any packets - returning zero score!"); - let mixes = topology.get_mix_nodes(); - let providers = topology.get_mix_provider_nodes(); + let mixes = topology.mix_nodes(); + let providers = topology.providers(); let health = mixes .into_iter() @@ -58,7 +58,7 @@ impl HealthCheckResult { score_threshold: f64, ) -> T { let filtered_mix_nodes = topology - .get_mix_nodes() + .mix_nodes() .into_iter() .filter(|node| { match self.node_score(NodeAddressBytes::from_b64_string(node.pub_key.clone())) { @@ -72,7 +72,7 @@ impl HealthCheckResult { .collect(); let filtered_provider_nodes = topology - .get_mix_provider_nodes() + .providers() .into_iter() .filter(|node| { match self.node_score(NodeAddressBytes::from_b64_string(node.pub_key.clone())) { @@ -85,7 +85,7 @@ impl HealthCheckResult { }) .collect(); // coco nodes remain unchanged as no healthcheck is being run on them or time being - let filtered_coco_nodes = topology.get_coco_nodes(); + let filtered_coco_nodes = topology.coco_nodes(); T::new_from_nodes( filtered_mix_nodes, @@ -110,19 +110,16 @@ impl HealthCheckResult { // create entries for all nodes let mut score_map = HashMap::new(); - topology.get_mix_nodes().into_iter().for_each(|node| { + topology.mix_nodes().into_iter().for_each(|node| { score_map.insert(node.get_pub_key_bytes(), NodeScore::from_mixnode(node)); }); - topology - .get_mix_provider_nodes() - .into_iter() - .for_each(|node| { - score_map.insert(node.get_pub_key_bytes(), NodeScore::from_provider(node)); - }); + topology.providers().into_iter().for_each(|node| { + score_map.insert(node.get_pub_key_bytes(), NodeScore::from_provider(node)); + }); let ephemeral_keys = DummyMixIdentityKeyPair::new(); - let providers = topology.get_mix_provider_nodes(); + let providers = topology.providers(); let mut path_checker = PathChecker::new(providers, ephemeral_keys).await; for i in 0..iterations { diff --git a/common/healthcheck/src/score.rs b/common/healthcheck/src/score.rs index 8f97cf96a8..4d29e8f5b6 100644 --- a/common/healthcheck/src/score.rs +++ b/common/healthcheck/src/score.rs @@ -4,7 +4,8 @@ use std::cmp::Ordering; use std::fmt::Error; use std::fmt::Formatter; use std::net::SocketAddr; -use topology::{MixNode, MixProviderNode}; +use topology::mix; +use topology::provider; // TODO: should 'nodetype' really be part of healthcheck::score @@ -91,7 +92,7 @@ impl PartialEq for NodeScore { } impl NodeScore { - pub(crate) fn from_mixnode(node: MixNode) -> Self { + pub(crate) fn from_mixnode(node: mix::Node) -> Self { NodeScore { typ: NodeType::Mix, pub_key: NodeAddressBytes::from_b64_string(node.pub_key), @@ -103,7 +104,7 @@ impl NodeScore { } } - pub(crate) fn from_provider(node: MixProviderNode) -> Self { + pub(crate) fn from_provider(node: provider::Node) -> Self { NodeScore { typ: NodeType::MixProvider, pub_key: NodeAddressBytes::from_b64_string(node.pub_key), diff --git a/common/topology/Cargo.toml b/common/topology/Cargo.toml index e9e8022c91..905df9f2da 100644 --- a/common/topology/Cargo.toml +++ b/common/topology/Cargo.toml @@ -20,4 +20,5 @@ addressing = {path = "../addressing"} version-checker = {path = "../version-checker" } ## will be moved to proper dependencies once released -sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" } +sphinx = { git = "https://github.com/nymtech/sphinx", rev="c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" } +# sphinx = { path = "../../../sphinx"} diff --git a/common/topology/src/coco.rs b/common/topology/src/coco.rs new file mode 100644 index 0000000000..bfc20a093e --- /dev/null +++ b/common/topology/src/coco.rs @@ -0,0 +1,15 @@ +use crate::filter; + +#[derive(Debug, Clone)] +pub struct Node { + pub host: String, + pub pub_key: String, + pub last_seen: u64, + pub version: String, +} + +impl filter::Versioned for Node { + fn version(&self) -> String { + self.version.clone() + } +} diff --git a/common/topology/src/filter.rs b/common/topology/src/filter.rs new file mode 100644 index 0000000000..68242b8b92 --- /dev/null +++ b/common/topology/src/filter.rs @@ -0,0 +1,18 @@ +pub trait Versioned: Clone { + fn version(&self) -> String; +} + +pub trait VersionFilterable { + fn filter_by_version(&self, expected_version: &str) -> Self; +} + +impl VersionFilterable for Vec { + fn filter_by_version(&self, expected_version: &str) -> Self { + self.iter() + .filter(|node| { + version_checker::is_minor_version_compatible(&node.version(), expected_version) + }) + .cloned() + .collect() + } +} diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 8fadc5c2ab..94ac5a74b4 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -1,103 +1,29 @@ -use addressing; -use curve25519_dalek::montgomery::MontgomeryPoint; +use crate::filter::VersionFilterable; use itertools::Itertools; use rand::seq::IteratorRandom; -use sphinx::route::{Node as SphinxNode, NodeAddressBytes}; +use sphinx::route::Node as SphinxNode; use std::cmp::max; use std::collections::HashMap; -use std::net::SocketAddr; -use version_checker; -#[derive(Debug, Clone)] -pub struct MixNode { - pub host: SocketAddr, - pub pub_key: String, - pub layer: u64, - pub last_seen: u64, - pub version: String, -} +pub mod coco; +mod filter; +pub mod mix; +pub mod provider; -impl Into for MixNode { - fn into(self) -> SphinxNode { - let address_bytes = addressing::encoded_bytes_from_socket_address(self.host); - let key_bytes = self.get_pub_key_bytes(); - let key = MontgomeryPoint(key_bytes); - - SphinxNode::new(NodeAddressBytes::from_bytes(address_bytes), key) - } -} - -impl MixNode { - pub fn get_pub_key_bytes(&self) -> [u8; 32] { - let decoded_key_bytes = base64::decode_config(&self.pub_key, base64::URL_SAFE).unwrap(); - let mut key_bytes = [0; 32]; - key_bytes.copy_from_slice(&decoded_key_bytes[..]); - key_bytes - } -} - -#[derive(Debug, Clone)] -pub struct MixProviderClient { - pub pub_key: String, -} - -#[derive(Debug, Clone)] -pub struct MixProviderNode { - pub client_listener: SocketAddr, - pub mixnet_listener: SocketAddr, - pub pub_key: String, - pub registered_clients: Vec, - pub last_seen: u64, - pub version: String, -} - -impl Into for MixProviderNode { - fn into(self) -> SphinxNode { - let address_bytes = addressing::encoded_bytes_from_socket_address(self.mixnet_listener); - let key_bytes = self.get_pub_key_bytes(); - let key = MontgomeryPoint(key_bytes); - - SphinxNode::new(NodeAddressBytes::from_bytes(address_bytes), key) - } -} - -impl MixProviderNode { - pub fn get_pub_key_bytes(&self) -> [u8; 32] { - let decoded_key_bytes = base64::decode_config(&self.pub_key, base64::URL_SAFE).unwrap(); - let mut key_bytes = [0; 32]; - key_bytes.copy_from_slice(&decoded_key_bytes[..]); - key_bytes - } -} - -#[derive(Debug, Clone)] -pub struct CocoNode { - pub host: String, - pub pub_key: String, - pub last_seen: u64, - pub version: String, -} - -#[derive(Debug)] -pub enum NymTopologyError { - InvalidMixLayerError, - MissingLayerError(Vec), -} - -pub trait NymTopology: Sized { +pub trait NymTopology: Sized + PartialEq + std::fmt::Debug + Send + Sync { fn new(directory_server: String) -> Self; fn new_from_nodes( - mix_nodes: Vec, - mix_provider_nodes: Vec, - coco_nodes: Vec, + mix_nodes: Vec, + mix_provider_nodes: Vec, + coco_nodes: Vec, ) -> Self; - fn get_mix_nodes(&self) -> Vec; - fn get_mix_provider_nodes(&self) -> Vec; - fn get_coco_nodes(&self) -> Vec; - fn make_layered_topology(&self) -> Result>, NymTopologyError> { - let mut layered_topology: HashMap> = HashMap::new(); + fn mix_nodes(&self) -> Vec; + fn providers(&self) -> Vec; + fn coco_nodes(&self) -> Vec; + fn make_layered_topology(&self) -> Result>, NymTopologyError> { + let mut layered_topology: HashMap> = HashMap::new(); let mut highest_layer = 0; - for mix in self.get_mix_nodes() { + for mix in self.mix_nodes() { // we need to have extra space for provider if mix.layer > sphinx::constants::MAX_PATH_LENGTH as u64 { return Err(NymTopologyError::InvalidMixLayerError); @@ -125,6 +51,8 @@ pub trait NymTopology: Sized { Ok(layered_topology) } + + // Tries to get a route through the mix network fn mix_route(&self) -> Result, NymTopologyError> { let mut layered_topology = self.make_layered_topology()?; let num_layers = layered_topology.len(); @@ -137,7 +65,7 @@ pub trait NymTopology: Sized { Ok(route) } - // sets a route to specific provider + // Sets up a route to a specific provider fn route_to(&self, provider_node: SphinxNode) -> Result, NymTopologyError> { Ok(self .mix_route()? @@ -148,7 +76,7 @@ pub trait NymTopology: Sized { fn all_paths(&self) -> Result>, NymTopologyError> { let mut layered_topology = self.make_layered_topology()?; - let providers = self.get_mix_provider_nodes(); + let providers = self.providers(); let sorted_layers: Vec> = (1..=layered_topology.len() as u64) .map(|layer| layered_topology.remove(&layer).unwrap()) // get all nodes per layer @@ -172,32 +100,13 @@ pub trait NymTopology: Sized { expected_provider_version: &str, expected_coco_version: &str, ) -> Self { - let filtered_mixes = self - .get_mix_nodes() - .iter() - .filter(|mix_node| { - version_checker::is_compatible(&mix_node.version, expected_mix_version) - }) - .cloned() - .collect(); - let filtered_providers = self - .get_mix_provider_nodes() - .iter() - .filter(|provider_node| { - version_checker::is_compatible(&provider_node.version, expected_provider_version) - }) - .cloned() - .collect(); - let filtered_coco_nodes = self - .get_coco_nodes() - .iter() - .filter(|coco_node| { - version_checker::is_compatible(&coco_node.version, expected_coco_version) - }) - .cloned() - .collect(); + let mixes = self.mix_nodes().filter_by_version(expected_mix_version); + let providers = self + .providers() + .filter_by_version(expected_provider_version); + let cocos = self.coco_nodes().filter_by_version(expected_coco_version); - Self::new_from_nodes(filtered_mixes, filtered_providers, filtered_coco_nodes) + Self::new_from_nodes(mixes, providers, cocos) } fn can_construct_path_through(&self) -> bool { @@ -208,4 +117,8 @@ pub trait NymTopology: Sized { } } -// TODO: tests... +#[derive(Debug)] +pub enum NymTopologyError { + InvalidMixLayerError, + MissingLayerError(Vec), +} diff --git a/common/topology/src/mix.rs b/common/topology/src/mix.rs new file mode 100644 index 0000000000..5236e191fa --- /dev/null +++ b/common/topology/src/mix.rs @@ -0,0 +1,38 @@ +use crate::filter; +use sphinx::route::Node as SphinxNode; +use sphinx::route::NodeAddressBytes; +use std::net::SocketAddr; + +#[derive(Debug, Clone)] +pub struct Node { + pub host: SocketAddr, + pub pub_key: String, + pub layer: u64, + pub last_seen: u64, + pub version: String, +} + +impl Node { + pub fn get_pub_key_bytes(&self) -> [u8; 32] { + let decoded_key_bytes = base64::decode_config(&self.pub_key, base64::URL_SAFE).unwrap(); + let mut key_bytes = [0; 32]; + key_bytes.copy_from_slice(&decoded_key_bytes[..]); + key_bytes + } +} + +impl filter::Versioned for Node { + fn version(&self) -> String { + self.version.clone() + } +} + +impl Into for Node { + fn into(self) -> SphinxNode { + let address_bytes = addressing::encoded_bytes_from_socket_address(self.host); + let key_bytes = self.get_pub_key_bytes(); + let key = sphinx::key::new(key_bytes); + + SphinxNode::new(NodeAddressBytes::from_bytes(address_bytes), key) + } +} diff --git a/common/topology/src/provider.rs b/common/topology/src/provider.rs new file mode 100644 index 0000000000..e9d4869e0e --- /dev/null +++ b/common/topology/src/provider.rs @@ -0,0 +1,44 @@ +use crate::filter; +use sphinx::route::Node as SphinxNode; +use sphinx::route::NodeAddressBytes; +use std::net::SocketAddr; + +#[derive(Debug, Clone)] +pub struct Client { + pub pub_key: String, +} + +#[derive(Debug, Clone)] +pub struct Node { + pub client_listener: SocketAddr, + pub mixnet_listener: SocketAddr, + pub pub_key: String, + pub registered_clients: Vec, + pub last_seen: u64, + pub version: String, +} + +impl Node { + pub fn get_pub_key_bytes(&self) -> [u8; 32] { + let decoded_key_bytes = base64::decode_config(&self.pub_key, base64::URL_SAFE).unwrap(); + let mut key_bytes = [0; 32]; + key_bytes.copy_from_slice(&decoded_key_bytes[..]); + key_bytes + } +} + +impl filter::Versioned for Node { + fn version(&self) -> String { + self.version.clone() + } +} + +impl Into for Node { + fn into(self) -> SphinxNode { + let address_bytes = addressing::encoded_bytes_from_socket_address(self.mixnet_listener); + let key_bytes = self.get_pub_key_bytes(); + let key = sphinx::key::new(key_bytes); + + SphinxNode::new(NodeAddressBytes::from_bytes(address_bytes), key) + } +} diff --git a/common/version-checker/src/lib.rs b/common/version-checker/src/lib.rs index d6ae38d19e..55232a9885 100644 --- a/common/version-checker/src/lib.rs +++ b/common/version-checker/src/lib.rs @@ -1,16 +1,22 @@ use semver::Version; use semver::VersionReq; -/// Checks whether given `version` is compatible with a given semantic -/// version requirement `req` according to major-minor semver rules. -/// The semantic version requirement can be passed as a full, concrete version -/// number, because that's what we'll have in our Cargo.toml files (e.g. 0.3.2). -/// The patch number in the requirement will be dropped and replaced with a wildcard (0.3.*) as all minor versions should be compatible with each other. -pub fn is_compatible(version: &str, req: &str) -> bool { - let version = Version::parse(version).unwrap(); - let tmp = Version::parse(req).unwrap(); +/// Checks whether given `version` is compatible with a given semantic version requirement `req` +/// according to major-minor semver rules. The semantic version requirement can be passed as a full, +/// concrete version number, because that's what we'll have in our Cargo.toml files (e.g. 0.3.2). +/// The patch number in the requirement gets dropped and replaced with a wildcard (0.3.*) as all +/// minor versions should be compatible with each other. +pub fn is_minor_version_compatible(version: &str, req: &str) -> bool { + let version = match Version::parse(version) { + Ok(v) => v, + Err(_) => return false, + }; + let tmp = match Version::parse(req) { + Ok(v) => v, + Err(_) => return false, + }; let wildcard = format!("{}.{}.*", tmp.major, tmp.minor).to_string(); - let semver_requirement = VersionReq::parse(&wildcard).unwrap(); + let semver_requirement = VersionReq::parse(&wildcard).expect("panicked on semver requirement parsing. This should never happen as inputs should already have been sanitized."); semver_requirement.matches(&version) } @@ -21,31 +27,41 @@ mod tests { #[test] fn version_0_3_0_is_compatible_with_requirement_0_3_x() { - assert!(is_compatible("0.3.0", "0.3.2")); + assert!(is_minor_version_compatible("0.3.0", "0.3.2")); } #[test] fn version_0_3_1_is_compatible_with_minimum_requirement_0_3_x() { - assert!(is_compatible("0.3.1", "0.3.2")); + assert!(is_minor_version_compatible("0.3.1", "0.3.2")); } #[test] fn version_0_3_2_is_compatible_with_minimum_requirement_0_3_x() { - assert!(is_compatible("0.3.2", "0.3.0")); + assert!(is_minor_version_compatible("0.3.2", "0.3.0")); } #[test] fn version_0_2_0_is_not_compatible_with_requirement_0_3_x() { - assert!(!is_compatible("0.2.0", "0.3.2")); + assert!(!is_minor_version_compatible("0.2.0", "0.3.2")); } #[test] fn version_0_4_0_is_not_compatible_with_requirement_0_3_x() { - assert!(!is_compatible("0.4.0", "0.3.2")); + assert!(!is_minor_version_compatible("0.4.0", "0.3.2")); } #[test] fn version_1_3_2_is_not_compatible_with_requirement_0_3_x() { - assert!(!is_compatible("1.3.2", "0.3.2")); + assert!(!is_minor_version_compatible("1.3.2", "0.3.2")); + } + + #[test] + fn returns_false_on_foo_version() { + assert!(!is_minor_version_compatible("foo", "0.3.2")); + } + + #[test] + fn returns_false_on_bar_version() { + assert!(!is_minor_version_compatible("0.3.2", "bar")); } } diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml index e538c329d9..52db3fa9b6 100644 --- a/mixnode/Cargo.toml +++ b/mixnode/Cargo.toml @@ -22,7 +22,7 @@ addressing = {path = "../common/addressing" } directory-client = { path = "../common/clients/directory-client" } ## will be moved to proper dependencies once released -sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" } +sphinx = { git = "https://github.com/nymtech/sphinx", rev="c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" } [build-dependencies] built = "0.3.2" diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 1ae4b6e65d..c8ce9ae783 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -39,6 +39,7 @@ impl Config { pub enum MixProcessingError { SphinxRecoveryError, ReceivedFinalHopError, + SphinxProcessingError, } impl From for MixProcessingError { @@ -115,10 +116,14 @@ impl PacketProcessor { let packet = SphinxPacket::from_bytes(packet_data.to_vec())?; let (next_packet, next_hop_address, delay) = match packet.process(processing_data.secret_key) { - ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay) => { + Ok(ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay)) => { (packet, address, delay) } - _ => return Err(MixProcessingError::ReceivedFinalHopError), + Ok(_) => return Err(MixProcessingError::ReceivedFinalHopError), + Err(e) => { + warn!("Failed to unwrap Sphinx packet: {:?}", e); + return Err(MixProcessingError::SphinxProcessingError); + } }; let next_mix = MixPeer::new(next_hop_address); diff --git a/nym-client/Cargo.toml b/nym-client/Cargo.toml index 51730b0dd1..eade92dd4b 100644 --- a/nym-client/Cargo.toml +++ b/nym-client/Cargo.toml @@ -40,7 +40,8 @@ sfw-provider-requests = { path = "../sfw-provider/sfw-provider-requests" } topology = {path = "../common/topology" } ## will be moved to proper dependencies once released -sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" } +sphinx = { git = "https://github.com/nymtech/sphinx", rev="c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" } +# sphinx = { path = "../../sphinx"} # putting this explicitly below everything and most likely, the next time we look into it, it will already have a proper release tokio-tungstenite = { git = "https://github.com/snapview/tokio-tungstenite", rev="308d9680c0e59dd1e8651659a775c05df937934e" } diff --git a/nym-client/src/client/cover_traffic_stream.rs b/nym-client/src/client/cover_traffic_stream.rs index 82f904eec7..f1caedc740 100644 --- a/nym-client/src/client/cover_traffic_stream.rs +++ b/nym-client/src/client/cover_traffic_stream.rs @@ -1,29 +1,35 @@ use crate::client::mix_traffic::MixMessage; +use crate::client::topology_control::TopologyInnerRef; use crate::client::LOOP_COVER_AVERAGE_DELAY; use futures::channel::mpsc; -use log::{info, trace}; +use log::{info, trace, warn}; use sphinx::route::Destination; use std::time::Duration; use topology::NymTopology; -pub(crate) async fn start_loop_cover_traffic_stream( +pub(crate) async fn start_loop_cover_traffic_stream( tx: mpsc::UnboundedSender, our_info: Destination, - topology: T, -) where - T: NymTopology, -{ + topology_ctrl_ref: TopologyInnerRef, +) { info!("Starting loop cover traffic stream"); loop { trace!("next cover message!"); let delay = mix_client::poisson::sample(LOOP_COVER_AVERAGE_DELAY); let delay_duration = Duration::from_secs_f64(delay); tokio::time::delay_for(delay_duration).await; - let cover_message = mix_client::packet::loop_cover_message( - our_info.address, - our_info.identifier, - &topology, - ); + + let read_lock = topology_ctrl_ref.read().await; + let topology = read_lock.topology.as_ref(); + if topology.is_none() { + warn!("No valid topology detected - won't send any loop cover message this time"); + continue; + } + + let topology = topology.unwrap(); + + let cover_message = + mix_client::packet::loop_cover_message(our_info.address, our_info.identifier, topology); // if this one fails, there's no retrying because it means that either: // - we run out of memory diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index 8359716f71..c2435acf0d 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -1,6 +1,6 @@ -use crate::built_info; use crate::client::mix_traffic::MixTrafficController; use crate::client::received_buffer::ReceivedMessagesBuffer; +use crate::client::topology_control::TopologyInnerRef; use crate::sockets::tcp; use crate::sockets::ws; use directory_client::presence::Topology; @@ -18,6 +18,7 @@ mod mix_traffic; mod provider_poller; mod real_traffic_stream; pub mod received_buffer; +pub mod topology_control; // TODO: all of those constants should probably be moved to config file const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5; @@ -26,6 +27,8 @@ const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 0.5; // seconds; const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds; +const TOPOLOGY_REFRESH_RATE: f64 = 10.0; // seconds + pub enum SocketType { TCP, WebSocket, @@ -46,13 +49,6 @@ pub struct NymClient { socket_type: SocketType, } -// TODO: this will be moved into module responsible for refreshing topology -#[derive(Debug)] -enum TopologyError { - HealthCheckError, - NoValidPathsError, -} - #[derive(Debug)] pub struct InputMessage(pub Destination, pub Vec); @@ -77,50 +73,16 @@ impl NymClient { } } - // TODO: this will be moved into module responsible for refreshing topology - async fn get_compatible_topology(&self) -> Result { - let score_threshold = 0.0; - info!("Trying to obtain valid, healthy, topology"); - - let full_topology = Topology::new(self.directory.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: self.directory.clone(), - // those are literally unrelevant 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; - - let healthcheck_scores = match healthcheck_result { - Err(err) => { - error!("Error while performing the healtcheck: {:?}", err); - return Err(TopologyError::HealthCheckError); - } - Ok(scores) => scores, - }; - - let healthy_topology = - healthcheck_scores.filter_topology_by_score(&full_topology, score_threshold); - - // for time being assume same versioning, i.e. if client is running X.Y.Z, - // we're expecting mixes, providers and coconodes to also be running X.Y.Z - let versioned_healthy_topology = healthy_topology.filter_node_versions( - built_info::PKG_VERSION, - built_info::PKG_VERSION, - built_info::PKG_VERSION, - ); - - // make sure you can still send a packet through the network: - if !versioned_healthy_topology.can_construct_path_through() { - return Err(TopologyError::NoValidPathsError); - } - - Ok(versioned_healthy_topology) + async fn get_provider_socket_address( + &self, + topology_ctrl_ref: TopologyInnerRef, + ) -> SocketAddr { + // this is temporary and assumes there exists only a single provider. + topology_ctrl_ref.read().await.topology.as_ref().unwrap() + .providers() + .first() + .expect("Could not get a provider from the initial network topology, are you using the right directory server?") + .client_listener } pub fn start(self) -> Result<(), Box> { @@ -144,20 +106,14 @@ impl NymClient { let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) = mpsc::unbounded(); - // get initial topology; already filtered by health and version - let initial_topology = match rt.block_on(self.get_compatible_topology()) { - Ok(topology) => topology, - Err(err) => { - panic!("Failed to obtain initial network topology: {:?}", err); - } - }; + // 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, + )); - // this is temporary and assumes there exists only a single provider. - let provider_client_listener_address: SocketAddr = initial_topology - .get_mix_provider_nodes() - .first() - .expect("Could not get a provider from the supplied network topology, are you using the right directory server?") - .client_listener; + let provider_client_listener_address = + rt.block_on(self.get_provider_socket_address(topology_controller.get_inner_ref())); let mut provider_poller = provider_poller::ProviderPoller::new( poller_input_tx, @@ -189,13 +145,13 @@ impl NymClient { rt.spawn(cover_traffic_stream::start_loop_cover_traffic_stream( mix_tx.clone(), Destination::new(self.address, Default::default()), - initial_topology.clone(), + topology_controller.get_inner_ref(), )); // cloning arguments required by OutQueueControl; required due to move - let topology_clone = initial_topology.clone(); let self_address = self.address; let input_rx = self.input_rx; + let topology_ref = topology_controller.get_inner_ref(); // future constantly pumping traffic at some specified average rate // if a real message is available on 'input_rx' that might have been received from say @@ -206,7 +162,7 @@ impl NymClient { mix_tx, input_rx, Destination::new(self_address, Default::default()), - topology_clone, + topology_ref, ) .run_out_queue_control() .await @@ -225,7 +181,7 @@ impl NymClient { self.input_tx, received_messages_buffer_output_tx, self.address, - initial_topology, + topology_controller.get_inner_ref(), )); } SocketType::TCP => { @@ -234,12 +190,16 @@ impl NymClient { self.input_tx, received_messages_buffer_output_tx, self.address, - initial_topology, + topology_controller.get_inner_ref(), )); } SocketType::None => (), } + // future responsible for periodically polling directory server and updating + // the current global view of topology + let topology_refresher_future = rt.spawn(topology_controller.run_refresher()); + rt.block_on(async { let future_results = join!( received_messages_buffer_controllers_future, @@ -247,6 +207,7 @@ impl NymClient { loop_cover_traffic_future, out_queue_control_future, provider_polling_future, + topology_refresher_future, ); assert!( @@ -255,6 +216,7 @@ impl NymClient { && future_results.2.is_ok() && future_results.3.is_ok() && future_results.4.is_ok() + && future_results.5.is_ok() ); }); diff --git a/nym-client/src/client/real_traffic_stream.rs b/nym-client/src/client/real_traffic_stream.rs index f89257747b..9d1ee055ab 100644 --- a/nym-client/src/client/real_traffic_stream.rs +++ b/nym-client/src/client/real_traffic_stream.rs @@ -1,38 +1,38 @@ use crate::client::mix_traffic::MixMessage; +use crate::client::topology_control::TopologyInnerRef; use crate::client::{InputMessage, MESSAGE_SENDING_AVERAGE_DELAY}; -use directory_client::presence::Topology; use futures::channel::mpsc; use futures::task::{Context, Poll}; use futures::{Future, Stream, StreamExt}; -use log::{debug, info, trace}; +use log::{info, trace, warn}; use sphinx::route::Destination; -use sphinx::SphinxPacket; -use std::net::SocketAddr; use std::pin::Pin; use std::time::Duration; use tokio::time; +use topology::NymTopology; // have a rather low value for test sake const AVERAGE_PACKET_DELAY: f64 = 0.1; -pub(crate) struct OutQueueControl { +pub(crate) struct OutQueueControl { delay: time::Delay, mix_tx: mpsc::UnboundedSender, input_rx: mpsc::UnboundedReceiver, our_info: Destination, - - // due to pinning, DerefMut trait, futures, etc its way easier to - // just have concrete implementation here rather than generic NymTopology - // considering that it will be replaced with refreshing topology within few days anyway - topology: Topology, + topology_ctrl_ref: TopologyInnerRef, } -impl Stream for OutQueueControl { - type Item = (SocketAddr, SphinxPacket); +pub(crate) enum StreamMessage { + Cover, + Real(InputMessage), +} + +impl Stream for OutQueueControl { + type Item = StreamMessage; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { // it is not yet time to return a message - if Future::poll(Pin::new(&mut self.delay), cx).is_pending() { + if Pin::new(&mut self.delay).poll(cx).is_pending() { return Poll::Pending; }; @@ -49,41 +49,26 @@ impl Stream for OutQueueControl { self.delay.reset(next); // decide what kind of message to send - match Stream::poll_next(Pin::new(&mut self.input_rx), cx) { + match Pin::new(&mut self.input_rx).poll_next(cx) { // in the case our real message channel stream was closed, we should also indicate we are closed // (and whoever is using the stream should panic) Poll::Ready(None) => Poll::Ready(None), // if there's an actual message - return it - Poll::Ready(Some(real_message)) => { - trace!("real message"); - Poll::Ready(Some(mix_client::packet::encapsulate_message( - real_message.0, - real_message.1, - &self.topology, - AVERAGE_PACKET_DELAY, - ))) - } + Poll::Ready(Some(real_message)) => Poll::Ready(Some(StreamMessage::Real(real_message))), // otherwise construct a dummy one - _ => { - trace!("loop cover message"); - Poll::Ready(Some(mix_client::packet::loop_cover_message( - self.our_info.address, - self.our_info.identifier, - &self.topology, - ))) - } + Poll::Pending => Poll::Ready(Some(StreamMessage::Cover)), } } } -impl OutQueueControl { +impl OutQueueControl { pub(crate) fn new( mix_tx: mpsc::UnboundedSender, input_rx: mpsc::UnboundedReceiver, our_info: Destination, - topology: Topology, + topology: TopologyInnerRef, ) -> Self { let initial_delay = time::delay_for(Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY)); OutQueueControl { @@ -91,20 +76,44 @@ impl OutQueueControl { mix_tx, input_rx, our_info, - topology, + topology_ctrl_ref: topology, } } pub(crate) async fn run_out_queue_control(mut self) { info!("starting out queue controller"); while let Some(next_message) = self.next().await { - debug!("created new message"); + trace!("created new message"); + let read_lock = self.topology_ctrl_ref.read().await; + let topology = read_lock.topology.as_ref(); + + if topology.is_none() { + warn!("No valid topology detected - won't send any loop cover or real message this time"); + continue; + } + + let topology = topology.unwrap(); + + let next_packet = match next_message { + StreamMessage::Cover => mix_client::packet::loop_cover_message( + self.our_info.address, + self.our_info.identifier, + topology, + ), + StreamMessage::Real(real_message) => mix_client::packet::encapsulate_message( + real_message.0, + real_message.1, + topology, + AVERAGE_PACKET_DELAY, + ), + }; + // if this one fails, there's no retrying because it means that either: // - we run out of memory // - the receiver channel is closed // in either case there's no recovery and we can only panic self.mix_tx - .unbounded_send(MixMessage::new(next_message.0, next_message.1)) + .unbounded_send(MixMessage::new(next_packet.0, next_packet.1)) .unwrap(); } } diff --git a/nym-client/src/client/topology_control.rs b/nym-client/src/client/topology_control.rs new file mode 100644 index 0000000000..6f749682dd --- /dev/null +++ b/nym-client/src/client/topology_control.rs @@ -0,0 +1,141 @@ +use crate::built_info; +use log::{error, info, trace, warn}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::RwLock as FRwLock; +use topology::NymTopology; + +const NODE_HEALTH_THRESHOLD: f64 = 0.0; + +// auxiliary type for ease of use +pub type TopologyInnerRef = Arc>>; + +pub(crate) struct TopologyControl { + directory_server: String, + refresh_rate: f64, + inner: Arc>>, +} + +#[derive(Debug)] +enum TopologyError { + HealthCheckError, + 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 + { + 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 + } + }; + + TopologyControl { + directory_server, + refresh_rate, + inner: Arc::new(FRwLock::new(Inner::new(initial_topology))), + } + } + + 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; + + let healthcheck_scores = match healthcheck_result { + Err(err) => { + error!("Error while performing the healtcheck: {:?}", err); + return Err(TopologyError::HealthCheckError); + } + Ok(scores) => scores, + }; + + let healthy_topology = + healthcheck_scores.filter_topology_by_score(&full_topology, NODE_HEALTH_THRESHOLD); + + let versioned_healthy_topology = healthy_topology.filter_node_versions( + built_info::PKG_VERSION, + built_info::PKG_VERSION, + built_info::PKG_VERSION, + ); + + // make sure you can still send a packet through the network: + if !versioned_healthy_topology.can_construct_path_through() { + return Err(TopologyError::NoValidPathsError); + } + + Ok(versioned_healthy_topology) + } + + pub(crate) fn get_inner_ref(&self) -> Arc>> { + self.inner.clone() + } + + async fn update_global_topology(&mut self, new_topology: Option) { + // acquire write lock + let mut write_lock = self.inner.write().await; + write_lock.topology = new_topology; + } + + async fn should_update_topology(&mut self, new_topology: &Option) -> bool { + let read_lock = self.inner.read().await; + match new_topology { + // if new topology is invalid, we MUST update to it as it is impossible to send packets through + None => true, + Some(new_topology) => match &read_lock.topology { + None => true, + Some(old_topology) => new_topology != old_topology, + }, + } + } + + pub(crate) async fn run_refresher(mut self) { + info!("Starting topology refresher"); + 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 = match new_topology_res { + Ok(topology) => Some(topology), + Err(err) => { + warn!("the obtained topology seems to be invalid - {:?}, it will be impossible to send packets through", err); + None + } + }; + + if self.should_update_topology(&new_topology).await { + info!("Detected changes in topology - updating global view!"); + + self.update_global_topology(new_topology).await; + } + + tokio::time::delay_for(delay_duration).await; + } + } +} + +pub struct Inner { + pub topology: Option, +} + +impl Inner { + fn new(topology: Option) -> Self { + Inner { topology } + } +} diff --git a/nym-client/src/sockets/tcp.rs b/nym-client/src/sockets/tcp.rs index 709181a684..8f49707c60 100644 --- a/nym-client/src/sockets/tcp.rs +++ b/nym-client/src/sockets/tcp.rs @@ -1,18 +1,17 @@ use crate::client::received_buffer::BufferResponse; +use crate::client::topology_control::TopologyInnerRef; use crate::client::InputMessage; -use directory_client::presence::Topology; use futures::channel::{mpsc, oneshot}; use futures::future::FutureExt; use futures::io::Error; use futures::SinkExt; use log::*; use sphinx::route::{Destination, DestinationAddressBytes}; -use std::borrow::Borrow; use std::convert::TryFrom; use std::io; use std::net::SocketAddr; -use std::sync::Arc; use tokio::prelude::*; +use topology::NymTopology; const SEND_REQUEST_PREFIX: u8 = 1; const FETCH_REQUEST_PREFIX: u8 = 2; @@ -96,6 +95,16 @@ impl ClientRequest { mut input_tx: mpsc::UnboundedSender, ) -> ServerResponse { trace!("sending to: {:?}, msg: {:?}", recipient_address, msg); + if msg.len() > sphinx::constants::MAXIMUM_PLAINTEXT_LENGTH { + return ServerResponse::Error { + message: format!( + "too long message. Sent {} bytes while the maximum is {}", + msg.len(), + sphinx::constants::MAXIMUM_PLAINTEXT_LENGTH + ) + .to_string(), + }; + } let dummy_surb = [0; 16]; let input_msg = InputMessage(Destination::new(recipient_address, dummy_surb), msg); input_tx.send(input_msg).await.unwrap(); @@ -124,14 +133,24 @@ impl ClientRequest { ServerResponse::Fetch { messages } } - async fn handle_get_clients(topology: &Topology) -> ServerResponse { - let clients = topology - .mix_provider_nodes - .iter() - .flat_map(|provider| provider.registered_clients.iter()) - .map(|client| base64::decode_config(&client.pub_key, base64::URL_SAFE).unwrap()) // TODO: this can potentially throw an error - .collect(); - ServerResponse::GetClients { clients } + async fn handle_get_clients(topology: &TopologyInnerRef) -> ServerResponse { + let topology_data = &topology.read().await.topology; + match topology_data { + Some(topology) => { + let clients = topology + .providers() + .iter() + .flat_map(|provider| provider.registered_clients.iter()) + .filter_map(|client| { + base64::decode_config(&client.pub_key, base64::URL_SAFE).ok() + }) + .collect(); + ServerResponse::GetClients { clients } + } + None => ServerResponse::Error { + message: "Invalid network topology".to_string(), + }, + } } async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse { @@ -196,9 +215,9 @@ impl ServerResponse { } } -async fn handle_connection( +async fn handle_connection( data: &[u8], - request_handling_data: RequestHandlingData, + request_handling_data: RequestHandlingData, ) -> Result { let request = ClientRequest::try_from(data)?; let response = match request { @@ -211,7 +230,7 @@ async fn handle_connection( } ClientRequest::Fetch => ClientRequest::handle_fetch(request_handling_data.msg_query).await, ClientRequest::GetClients => { - ClientRequest::handle_get_clients(request_handling_data.topology.borrow()).await + ClientRequest::handle_get_clients(&request_handling_data.topology).await } ClientRequest::OwnDetails => { ClientRequest::handle_own_details(request_handling_data.self_address).await @@ -221,27 +240,25 @@ async fn handle_connection( Ok(response) } -struct RequestHandlingData { +struct RequestHandlingData { msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Arc, + topology: TopologyInnerRef, } -async fn accept_connection( +async fn accept_connection( mut socket: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, ) { let address = socket .peer_addr() .expect("connected streams should have a peer address"); debug!("Peer address: {}", address); - let topology = Arc::new(topology); - let mut buf = [0u8; 2048]; // In a loop, read data from the socket and write the data back. @@ -280,12 +297,12 @@ async fn accept_connection( } } -pub async fn start_tcpsocket( +pub async fn start_tcpsocket( address: SocketAddr, message_tx: mpsc::UnboundedSender, received_messages_query_tx: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, ) -> Result<(), TCPSocketError> { let mut listener = tokio::net::TcpListener::bind(address).await?; diff --git a/nym-client/src/sockets/ws.rs b/nym-client/src/sockets/ws.rs index 621c9ffe80..28d8b0e196 100644 --- a/nym-client/src/sockets/ws.rs +++ b/nym-client/src/sockets/ws.rs @@ -1,6 +1,6 @@ use crate::client::received_buffer::BufferResponse; +use crate::client::topology_control::TopologyInnerRef; use crate::client::InputMessage; -use directory_client::presence::Topology; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; use futures::channel::{mpsc, oneshot}; use futures::future::FutureExt; @@ -12,20 +12,21 @@ use sphinx::route::{Destination, DestinationAddressBytes}; use std::convert::TryFrom; use std::io; use std::net::SocketAddr; +use topology::NymTopology; use tungstenite::protocol::frame::coding::CloseCode; use tungstenite::protocol::{CloseFrame, Message}; -struct Connection { +struct Connection { address: SocketAddr, msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, rx: UnboundedReceiver, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, tx: UnboundedSender, } -impl Connection { +impl Connection { async fn handle_text_message(&self, msg: String) -> ServerResponse { debug!("Handling text message request"); trace!("Content: {:?}", msg.clone()); @@ -48,9 +49,7 @@ impl Connection { ClientRequest::handle_send(message, recipient_address, self.msg_input.clone()).await } ClientRequest::Fetch => ClientRequest::handle_fetch(self.msg_query.clone()).await, - ClientRequest::GetClients => { - ClientRequest::handle_get_clients(self.topology.clone()).await - } + ClientRequest::GetClients => ClientRequest::handle_get_clients(&self.topology).await, ClientRequest::OwnDetails => ClientRequest::handle_own_details(self.self_address).await, } } @@ -189,20 +188,12 @@ impl ClientRequest { mut input_tx: mpsc::UnboundedSender, ) -> ServerResponse { let message_bytes = msg.into_bytes(); - // TODO: wait until 0.4.0 release to replace those constants with newly exposed - // sphinx::constants::MAXIMUM_PLAINTEXT_LENGTH - // we can't do it now for compatibility reasons as most recent sphinx revision - // has breaking changes due to packet format changes - let maximum_plaintext_length = sphinx::constants::PAYLOAD_SIZE - - sphinx::constants::SECURITY_PARAMETER - - sphinx::constants::DESTINATION_ADDRESS_LENGTH - - 1; - if message_bytes.len() > maximum_plaintext_length { + if message_bytes.len() > sphinx::constants::MAXIMUM_PLAINTEXT_LENGTH { return ServerResponse::Error { message: format!( - "too long message. Sent {} bytes while the maximum is {}", + "message too long. Sent {} bytes, but the maximum is {}", message_bytes.len(), - maximum_plaintext_length + sphinx::constants::MAXIMUM_PLAINTEXT_LENGTH ) .to_string(), }; @@ -212,7 +203,7 @@ impl ClientRequest { Err(e) => { return ServerResponse::Error { message: e.to_string(), - } + }; } Ok(hex) => hex, }; @@ -269,14 +260,22 @@ impl ClientRequest { ServerResponse::Fetch { messages } } - async fn handle_get_clients(topology: Topology) -> ServerResponse { - let clients = topology - .mix_provider_nodes - .into_iter() - .flat_map(|provider| provider.registered_clients.into_iter()) - .map(|client| client.pub_key) - .collect(); - ServerResponse::GetClients { clients } + async fn handle_get_clients(topology: &TopologyInnerRef) -> ServerResponse { + let topology_data = &topology.read().await.topology; + match topology_data { + Some(topology) => { + let clients = topology + .providers() + .iter() + .flat_map(|provider| provider.registered_clients.iter()) + .map(|client| client.pub_key.clone()) + .collect(); + ServerResponse::GetClients { clients } + } + None => ServerResponse::Error { + message: "Invalid network topology".to_string(), + }, + } } async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse { @@ -306,14 +305,13 @@ impl Into for ServerResponse { } } -async fn accept_connection( +async fn accept_connection( stream: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender, msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, ) { - warn!("accept_connection"); let address = stream .peer_addr() .expect("connected streams should have a peer address"); @@ -337,6 +335,8 @@ async fn accept_connection( msg_query, self_address, }; + + // TODO: make sure this actually doesn't leak memory... tokio::spawn(conn.handle()); while let Some(message) = ws_stream.next().await { @@ -385,12 +385,12 @@ async fn accept_connection( } } -pub async fn start_websocket( +pub async fn start_websocket( address: SocketAddr, message_tx: mpsc::UnboundedSender, received_messages_query_tx: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: Topology, + topology: TopologyInnerRef, ) -> Result<(), WebSocketError> { let mut listener = tokio::net::TcpListener::bind(address).await?; diff --git a/sfw-provider/Cargo.toml b/sfw-provider/Cargo.toml index 4c0fc7c132..7bea6173d0 100644 --- a/sfw-provider/Cargo.toml +++ b/sfw-provider/Cargo.toml @@ -29,7 +29,7 @@ directory-client = { path = "../common/clients/directory-client" } sfw-provider-requests = { path = "./sfw-provider-requests" } ## will be moved to proper dependencies once released -sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" } +sphinx = { git = "https://github.com/nymtech/sphinx", rev="c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" } [build-dependencies] built = "0.3.2" diff --git a/sfw-provider/sfw-provider-requests/Cargo.toml b/sfw-provider/sfw-provider-requests/Cargo.toml index 97763acb6c..9b207791e3 100644 --- a/sfw-provider/sfw-provider-requests/Cargo.toml +++ b/sfw-provider/sfw-provider-requests/Cargo.toml @@ -7,4 +7,4 @@ edition = "2018" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" } +sphinx = { git = "https://github.com/nymtech/sphinx", rev="c3ba1447d61d6eb900fcdea0c20b3070a22bf11a" } diff --git a/sfw-provider/src/provider/mix_handling/mod.rs b/sfw-provider/src/provider/mix_handling/mod.rs index 1670311dd1..c7a4e8cc02 100644 --- a/sfw-provider/src/provider/mix_handling/mod.rs +++ b/sfw-provider/src/provider/mix_handling/mod.rs @@ -1,5 +1,6 @@ use crate::provider::storage::StoreData; use crypto::identity::DummyMixIdentityPrivateKey; +use log::warn; use sphinx::{ProcessedPacket, SphinxPacket}; use std::path::PathBuf; use std::sync::{Arc, RwLock}; @@ -8,11 +9,12 @@ use std::sync::{Arc, RwLock}; // DUPLICATE WITH MIXNODE CODE!!! #[derive(Debug)] pub enum MixProcessingError { - SphinxRecoveryError, - ReceivedForwardHopError, + FileIOFailure, InvalidPayload, NonMatchingRecipient, - FileIOFailure, + ReceivedForwardHopError, + SphinxRecoveryError, + SphinxProcessingError, } impl From for MixProcessingError { @@ -63,10 +65,14 @@ impl MixPacketProcessor { let read_processing_data = processing_data.read().unwrap(); let (client_address, client_surb_id, payload) = match packet.process(read_processing_data.secret_key.as_scalar()) { - ProcessedPacket::ProcessedPacketFinalHop(client_address, surb_id, payload) => { + Ok(ProcessedPacket::ProcessedPacketFinalHop(client_address, surb_id, payload)) => { (client_address, surb_id, payload) } - _ => return Err(MixProcessingError::ReceivedForwardHopError), + Ok(_) => return Err(MixProcessingError::ReceivedForwardHopError), + Err(e) => { + warn!("Error unwrapping Sphinx packet: {:?}", e); + return Err(MixProcessingError::SphinxProcessingError); + } }; // TODO: should provider try to be recovering plaintext? this would potentially make client retrieve messages of non-constant length,