Merge branch 'develop' into feature/base58

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",

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. 
 

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<topology::CocoNode> for CocoPresence {
-    fn into(self) -> topology::CocoNode {
-        topology::CocoNode {
+impl Into<topology::coco::Node> 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<topology::CocoNode> for CocoPresence {
     }
 }
 
-impl From<topology::CocoNode> for CocoPresence {
-    fn from(cn: CocoNode) -> Self {
+impl From<topology::coco::Node> 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<topology::MixNode> for MixNodePresence {
+impl TryInto<topology::mix::Node> for MixNodePresence {
     type Error = io::Error;
 
-    fn try_into(self) -> Result<MixNode, Self::Error> {
+    fn try_into(self) -> Result<topology::mix::Node, Self::Error> {
         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<topology::MixNode> 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<topology::MixNode> for MixNodePresence {
     }
 }
 
-impl From<topology::MixNode> for MixNodePresence {
-    fn from(mn: MixNode) -> Self {
+impl From<topology::mix::Node> 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<topology::MixProviderNode> 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<topology::provider::Node> 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<topology::MixProviderNode> for MixProviderPresence {
     }
 }
 
-impl From<topology::MixProviderNode> for MixProviderPresence {
-    fn from(mpn: MixProviderNode) -> Self {
+impl From<topology::provider::Node> 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<topology::MixProviderClient> for MixProviderClient {
-    fn into(self) -> topology::MixProviderClient {
-        topology::MixProviderClient {
+impl Into<topology::provider::Client> for MixProviderClient {
+    fn into(self) -> topology::provider::Client {
+        topology::provider::Client {
             pub_key: self.pub_key,
         }
     }
 }
 
-impl From<topology::MixProviderClient> for MixProviderClient {
-    fn from(mpc: topology::MixProviderClient) -> Self {
+impl From<topology::provider::Client> for MixProviderClient {
+    fn from(mpc: topology::provider::Client) -> Self {
         MixProviderClient {
             pub_key: mpc.pub_key,
         }
     }
 }
 
+impl PresenceEq for Vec<CocoPresence> {
+    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<MixNodePresence> {
+    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<MixProviderPresence> {
+    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::<Vec<_>>()
+                        .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::<Vec<_>>()
+                        .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<MixProviderPresence>,
 }
 
+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<MixNode>,
-        mix_provider_nodes: Vec<MixProviderNode>,
-        coco_nodes: Vec<CocoNode>,
+        mix_nodes: Vec<mix::Node>,
+        mix_provider_nodes: Vec<provider::Node>,
+        coco_nodes: Vec<coco::Node>,
     ) -> 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<topology::MixNode> {
+    fn mix_nodes(&self) -> Vec<mix::Node> {
         self.mix_nodes
             .iter()
             .filter_map(|x| x.clone().try_into().ok())
             .collect()
     }
 
-    fn get_mix_provider_nodes(&self) -> Vec<topology::MixProviderNode> {
+    fn providers(&self) -> Vec<provider::Node> {
         self.mix_provider_nodes
             .iter()
             .map(|x| x.clone().into())
             .collect()
     }
 
-    fn get_coco_nodes(&self) -> Vec<topology::CocoNode> {
+    fn coco_nodes(&self) -> Vec<topology::coco::Node> {
         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<topology::MixNode, io::Error> = mix_presence.try_into();
+        let result: Result<mix::Node, io::Error> = 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<topology::MixNode, io::Error> = mix_presence.try_into();
+        let result: Result<topology::mix::Node, io::Error> = mix_presence.try_into();
         assert!(result.is_ok())
     }
 }

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"}
 

common/clients/mix-client/src/packet.rs

@@ -28,7 +28,7 @@ pub fn encapsulate_message<T: NymTopology>(
     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();

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" }

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]

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<MixProviderNode>,
+        providers: Vec<provider::Node>,
         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)

common/healthcheck/src/result.rs

@@ -28,8 +28,8 @@ impl HealthCheckResult {
 
     fn zero_score<T: NymTopology>(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 {

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),

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"}

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()
+    }
+}

common/topology/src/filter.rs

@@ -0,0 +1,18 @@
+pub trait Versioned: Clone {
+    fn version(&self) -> String;
+}
+
+pub trait VersionFilterable<T> {
+    fn filter_by_version(&self, expected_version: &str) -> Self;
+}
+
+impl<T: Versioned> VersionFilterable<T> for Vec<T> {
+    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()
+    }
+}

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<SphinxNode> 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<MixProviderClient>,
-    pub last_seen: u64,
-    pub version: String,
-}
-
-impl Into<SphinxNode> 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<u64>),
-}
-
-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<MixNode>,
-        mix_provider_nodes: Vec<MixProviderNode>,
-        coco_nodes: Vec<CocoNode>,
+        mix_nodes: Vec<mix::Node>,
+        mix_provider_nodes: Vec<provider::Node>,
+        coco_nodes: Vec<coco::Node>,
     ) -> Self;
-    fn get_mix_nodes(&self) -> Vec<MixNode>;
-    fn get_mix_provider_nodes(&self) -> Vec<MixProviderNode>;
-    fn get_coco_nodes(&self) -> Vec<CocoNode>;
-    fn make_layered_topology(&self) -> Result<HashMap<u64, Vec<MixNode>>, NymTopologyError> {
-        let mut layered_topology: HashMap<u64, Vec<MixNode>> = HashMap::new();
+    fn mix_nodes(&self) -> Vec<mix::Node>;
+    fn providers(&self) -> Vec<provider::Node>;
+    fn coco_nodes(&self) -> Vec<coco::Node>;
+    fn make_layered_topology(&self) -> Result<HashMap<u64, Vec<mix::Node>>, NymTopologyError> {
+        let mut layered_topology: HashMap<u64, Vec<mix::Node>> = 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<Vec<SphinxNode>, 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<Vec<SphinxNode>, NymTopologyError> {
         Ok(self
             .mix_route()?
@@ -148,7 +76,7 @@ pub trait NymTopology: Sized {
 
     fn all_paths(&self) -> Result<Vec<Vec<SphinxNode>>, NymTopologyError> {
         let mut layered_topology = self.make_layered_topology()?;
-        let providers = self.get_mix_provider_nodes();
+        let providers = self.providers();
 
         let sorted_layers: Vec<Vec<SphinxNode>> = (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<u64>),
+}

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<SphinxNode> 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)
+    }
+}

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<Client>,
+    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<SphinxNode> 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)
+    }
+}

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"));
     }
 }

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"

mixnode/src/node/mod.rs

@@ -39,6 +39,7 @@ impl Config {
 pub enum MixProcessingError {
     SphinxRecoveryError,
     ReceivedFinalHopError,
+    SphinxProcessingError,
 }
 
 impl From<sphinx::ProcessingError> 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);

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" }

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<T>(
+pub(crate) async fn start_loop_cover_traffic_stream<T: NymTopology>(
     tx: mpsc::UnboundedSender<MixMessage>,
     our_info: Destination,
-    topology: T,
-) where
-    T: NymTopology,
-{
+    topology_ctrl_ref: TopologyInnerRef<T>,
+) {
     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

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<u8>);
 
@@ -77,50 +73,16 @@ impl NymClient {
         }
     }
 
-    // TODO: this will be moved into module responsible for refreshing topology
-    async fn get_compatible_topology(&self) -> Result<Topology, TopologyError> {
-        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<T: NymTopology>(
+        &self,
+        topology_ctrl_ref: TopologyInnerRef<T>,
+    ) -> 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<dyn std::error::Error>> {
@@ -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::<Topology>::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()
             );
         });
 

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<T: NymTopology> {
     delay: time::Delay,
     mix_tx: mpsc::UnboundedSender<MixMessage>,
     input_rx: mpsc::UnboundedReceiver<InputMessage>,
     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<T>,
 }
 
-impl Stream for OutQueueControl {
-    type Item = (SocketAddr, SphinxPacket);
+pub(crate) enum StreamMessage {
+    Cover,
+    Real(InputMessage),
+}
+
+impl<T: NymTopology> Stream for OutQueueControl<T> {
+    type Item = StreamMessage;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         // 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<T: NymTopology> OutQueueControl<T> {
     pub(crate) fn new(
         mix_tx: mpsc::UnboundedSender<MixMessage>,
         input_rx: mpsc::UnboundedReceiver<InputMessage>,
         our_info: Destination,
-        topology: Topology,
+        topology: TopologyInnerRef<T>,
     ) -> 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();
         }
     }

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<T> = Arc<FRwLock<Inner<T>>>;
+
+pub(crate) struct TopologyControl<T: NymTopology> {
+    directory_server: String,
+    refresh_rate: f64,
+    inner: Arc<FRwLock<Inner<T>>>,
+}
+
+#[derive(Debug)]
+enum TopologyError {
+    HealthCheckError,
+    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
+        {
+            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<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;
+
+        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<FRwLock<Inner<T>>> {
+        self.inner.clone()
+    }
+
+    async fn update_global_topology(&mut self, new_topology: Option<T>) {
+        // 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<T>) -> 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<T: NymTopology> {
+    pub topology: Option<T>,
+}
+
+impl<T: NymTopology> Inner<T> {
+    fn new(topology: Option<T>) -> Self {
+        Inner { topology }
+    }
+}

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<InputMessage>,
     ) -> 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<T: NymTopology>(topology: &TopologyInnerRef<T>) -> 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<T: NymTopology>(
     data: &[u8],
-    request_handling_data: RequestHandlingData,
+    request_handling_data: RequestHandlingData<T>,
 ) -> Result<ServerResponse, TCPSocketError> {
     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<T: NymTopology> {
     msg_input: mpsc::UnboundedSender<InputMessage>,
     msg_query: mpsc::UnboundedSender<BufferResponse>,
     self_address: DestinationAddressBytes,
-    topology: Arc<Topology>,
+    topology: TopologyInnerRef<T>,
 }
 
-async fn accept_connection(
+async fn accept_connection<T: 'static + NymTopology>(
     mut socket: tokio::net::TcpStream,
     msg_input: mpsc::UnboundedSender<InputMessage>,
     msg_query: mpsc::UnboundedSender<BufferResponse>,
     self_address: DestinationAddressBytes,
-    topology: Topology,
+    topology: TopologyInnerRef<T>,
 ) {
     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<T: 'static + NymTopology>(
     address: SocketAddr,
     message_tx: mpsc::UnboundedSender<InputMessage>,
     received_messages_query_tx: mpsc::UnboundedSender<BufferResponse>,
     self_address: DestinationAddressBytes,
-    topology: Topology,
+    topology: TopologyInnerRef<T>,
 ) -> Result<(), TCPSocketError> {
     let mut listener = tokio::net::TcpListener::bind(address).await?;
 

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<T: NymTopology> {
     address: SocketAddr,
     msg_input: mpsc::UnboundedSender<InputMessage>,
     msg_query: mpsc::UnboundedSender<BufferResponse>,
     rx: UnboundedReceiver<Message>,
     self_address: DestinationAddressBytes,
-    topology: Topology,
+    topology: TopologyInnerRef<T>,
     tx: UnboundedSender<Message>,
 }
 
-impl Connection {
+impl<T: NymTopology> Connection<T> {
     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<InputMessage>,
     ) -> 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<T: NymTopology>(topology: &TopologyInnerRef<T>) -> 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<Message> for ServerResponse {
     }
 }
 
-async fn accept_connection(
+async fn accept_connection<T: 'static + NymTopology>(
     stream: tokio::net::TcpStream,
     msg_input: mpsc::UnboundedSender<InputMessage>,
     msg_query: mpsc::UnboundedSender<BufferResponse>,
     self_address: DestinationAddressBytes,
-    topology: Topology,
+    topology: TopologyInnerRef<T>,
 ) {
-    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<T: 'static + NymTopology>(
     address: SocketAddr,
     message_tx: mpsc::UnboundedSender<InputMessage>,
     received_messages_query_tx: mpsc::UnboundedSender<BufferResponse>,
     self_address: DestinationAddressBytes,
-    topology: Topology,
+    topology: TopologyInnerRef<T>,
 ) -> Result<(), WebSocketError> {
     let mut listener = tokio::net::TcpListener::bind(address).await?;
 

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"

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" }

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<sphinx::ProcessingError> 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,