Merge branch 'release/0.6.0'

This commit is contained in:
Dave Hrycyszyn
2020-04-07 17:29:55 +01:00
63 changed files with 4231 additions and 414 deletions
+86 -7
View File
@@ -1,8 +1,83 @@
# Changelog
## [Unreleased](https://github.com/nymtech/nym/tree/HEAD)
## [v0.6.0](https://github.com/nymtech/nym/tree/HEAD)
[Full Changelog](https://github.com/nymtech/nym/compare/v0.4.1...HEAD)
[Full Changelog](https://github.com/nymtech/nym/compare/v0.5.0...HEAD)
This release fixes bugs in v0.5.0. All testnet node operators are advised to upgrade from v0.5.0.
* fixed premature EOFs on socket connections by using the new multi-TCP client
* fixed a bug causing client and mixnode connection hangs for misconfigured nodes
* by default 'Debug' section of saved configs is now empty and default values are used unless explicitly overridden
* introduced packet chunking allowing clients to send messages of arbitrary length. Note that packet retransmission is not implemented yet, so for longer messages, you might not get anything
* mixnodes now periodically log stats regarding number of packets mixed
* fixed possible client hang ups when sending high rates of traffic
* preventing mixes from starting with same announce-host as an existing node
* fixed overflow multiplication if connection backoff was set to a high value
**Closed issues:**
- Periodic activity summary [\#172](https://github.com/nymtech/nym/issues/172)
- Move contents of 'common/addressing' into 'common/nymsphinx' [\#161](https://github.com/nymtech/nym/issues/161)
- Make builds simpler for node operators [\#114](https://github.com/nymtech/nym/issues/114)
- Chunking in `nym-client` \(receive\) [\#83](https://github.com/nymtech/nym/issues/83)
- Chunking in `nym-client` \(send\) [\#82](https://github.com/nymtech/nym/issues/82)
**Merged pull requests:**
- Feature/tcp client connection timeout [\#176](https://github.com/nymtech/nym/pull/176) ([jstuczyn](https://github.com/jstuczyn))
- Feature/mixing stats logging [\#175](https://github.com/nymtech/nym/pull/175) ([jstuczyn](https://github.com/jstuczyn))
- Preventing multiplication overflow for reconnection backoff [\#174](https://github.com/nymtech/nym/pull/174) ([jstuczyn](https://github.com/jstuczyn))
- Feature/non mandatory debug config [\#173](https://github.com/nymtech/nym/pull/173) ([jstuczyn](https://github.com/jstuczyn))
- Feature/addressing move [\#169](https://github.com/nymtech/nym/pull/169) ([jstuczyn](https://github.com/jstuczyn))
- Checking if any other node is already announcing the same host [\#168](https://github.com/nymtech/nym/pull/168) ([jstuczyn](https://github.com/jstuczyn))
- Bugfix/closing tcp client connections on drop [\#167](https://github.com/nymtech/nym/pull/167) ([jstuczyn](https://github.com/jstuczyn))
- Yielding tokio task upon creating loop/real traffic message [\#166](https://github.com/nymtech/nym/pull/166) ([jstuczyn](https://github.com/jstuczyn))
- Feature/minor healthchecker improvements [\#165](https://github.com/nymtech/nym/pull/165) ([jstuczyn](https://github.com/jstuczyn))
- Feature/packet chunking [\#158](https://github.com/nymtech/nym/pull/158) ([jstuczyn](https://github.com/jstuczyn))
## [v0.5.0](https://github.com/nymtech/nym/tree/v0.5.0) (2020-03-23)
[Full Changelog](https://github.com/nymtech/nym/compare/v0.5.0-rc.1...v0.5.0)
1. Introduced proper configuration options for mixnodes, clients and providers. Everything is initialised with the `init` command that creates a saved config.toml file. To run the binary you now use `nym-<binary-name> run`, for example `nym-mixnode run`. Each flag can be overwritten at any stage with the following priority: run flags, data in config.toml and finally init flags.
2. Made mixnet TCP connections persistent. When sending a Sphinx packet, it should no longer go through the lengthy process of establishing a TCP connection only to immediately tear it down after sending a single packet. This significantly boosts throughput.
3. A lot of work on code clean up and refactoring including some performance fixes.
4. Client now determines its default nym-sfw-provider at startup and should always try to connect to the same one. Note: we still can't reliably run more than a single provider on the network.
5. Logging messages now have timestamps and when running at more aggressive log mode (like debug or even trace) we should no longer be overwhelmed with messages from external crates.
6. Initial compatibility with Windows. Please let us know if you have problems.
7. More work on validator, including initial Tendermint integration in Rust, and the start of the mixmining system.
**Closed issues:**
- Introduce timestamps to log messages [\#124](https://github.com/nymtech/nym/issues/124)
**Merged pull requests:**
- removing spooky startup warning message [\#155](https://github.com/nymtech/nym/pull/155) ([futurechimp](https://github.com/futurechimp))
- Some more startup fixes [\#154](https://github.com/nymtech/nym/pull/154) ([futurechimp](https://github.com/futurechimp))
- Entering runtime context when creating mix traffic controller [\#153](https://github.com/nymtech/nym/pull/153) ([jstuczyn](https://github.com/jstuczyn))
- Friendlification of startup messages [\#151](https://github.com/nymtech/nym/pull/151) ([futurechimp](https://github.com/futurechimp))
- Entering runtime context when creating packet forwarder [\#150](https://github.com/nymtech/nym/pull/150) ([jstuczyn](https://github.com/jstuczyn))
- Feature/add topology to validator [\#149](https://github.com/nymtech/nym/pull/149) ([futurechimp](https://github.com/futurechimp))
- Making code work on windows machines [\#148](https://github.com/nymtech/nym/pull/148) ([jstuczyn](https://github.com/jstuczyn))
- validator: adding HTTP interface [\#146](https://github.com/nymtech/nym/pull/146) ([futurechimp](https://github.com/futurechimp))
- Extracting the log setup [\#145](https://github.com/nymtech/nym/pull/145) ([futurechimp](https://github.com/futurechimp))
- Feature/optional location in configs [\#144](https://github.com/nymtech/nym/pull/144) ([jstuczyn](https://github.com/jstuczyn))
- Feature/concurrent connection managers [\#142](https://github.com/nymtech/nym/pull/142) ([jstuczyn](https://github.com/jstuczyn))
- Defaulting for global 'Info' logging level if not set in .env [\#140](https://github.com/nymtech/nym/pull/140) ([jstuczyn](https://github.com/jstuczyn))
- Provider not storing loop cover messages [\#139](https://github.com/nymtech/nym/pull/139) ([jstuczyn](https://github.com/jstuczyn))
- Using log builder to include timestamps + filters [\#138](https://github.com/nymtech/nym/pull/138) ([jstuczyn](https://github.com/jstuczyn))
- Feature/client ws refactoring [\#134](https://github.com/nymtech/nym/pull/134) ([jstuczyn](https://github.com/jstuczyn))
- Bugfix/metrics presence delay fix [\#133](https://github.com/nymtech/nym/pull/133) ([jstuczyn](https://github.com/jstuczyn))
- Removed outdated and redundant sample-configs [\#131](https://github.com/nymtech/nym/pull/131) ([jstuczyn](https://github.com/jstuczyn))
- If not overridden, 'announce-host' should default to 'host' [\#130](https://github.com/nymtech/nym/pull/130) ([jstuczyn](https://github.com/jstuczyn))
- Nice to know who we're talking to at startup... [\#129](https://github.com/nymtech/nym/pull/129) ([futurechimp](https://github.com/futurechimp))
## [v0.5.0-rc.1](https://github.com/nymtech/nym/tree/v0.5.0-rc.1) (2020-03-06)
[Full Changelog](https://github.com/nymtech/nym/compare/v0.4.1...v0.5.0-rc.1)
**Closed issues:**
@@ -48,14 +123,18 @@
## [v0.4.0](https://github.com/nymtech/nym/tree/v0.4.0) (2020-01-28)
[Full Changelog](https://github.com/nymtech/nym/compare/v0.4.0-rc.2...v0.4.0)
[Full Changelog](https://github.com/nymtech/nym/compare/0.4.0-rc.2...v0.4.0)
Nym 0.4.0 Platform
In this release, we're taking a lot more care with version numbers, so that we can ensure upgrade compatibility for mixnodes, providers, clients, and validators more easily.
Nym 0.4.0 Platform
In this release, we're taking a lot more care with version numbers, so that we can ensure upgrade compatibility for mixnodes, providers, clients, and validators more easily.
This release also integrates a health-checker and network topology refresh into the Nym client, so that the client can intelligently choose paths which route around any non-functional or incompatible nodes.
## [0.4.0-rc.2](https://github.com/nymtech/nym/tree/0.4.0-rc.2) (2020-01-28)
[Full Changelog](https://github.com/nymtech/nym/compare/v0.4.0-rc.2...0.4.0-rc.2)
## [v0.4.0-rc.2](https://github.com/nymtech/nym/tree/v0.4.0-rc.2) (2020-01-28)
[Full Changelog](https://github.com/nymtech/nym/compare/v0.4.0-rc.1...v0.4.0-rc.2)
Generated
+22 -19
View File
@@ -17,14 +17,6 @@ dependencies = [
"tokio 0.1.22",
]
[[package]]
name = "addressing"
version = "0.1.0"
dependencies = [
"log 0.4.8",
"pretty_env_logger",
]
[[package]]
name = "adler32"
version = "1.0.4"
@@ -931,13 +923,13 @@ dependencies = [
name = "healthcheck"
version = "0.1.0"
dependencies = [
"addressing",
"crypto",
"directory-client",
"futures 0.3.4",
"itertools",
"log 0.4.8",
"mix-client",
"multi-tcp-client",
"nymsphinx",
"pretty_env_logger",
"provider-client",
"rand 0.7.3",
@@ -1432,8 +1424,8 @@ dependencies = [
name = "mix-client"
version = "0.1.0"
dependencies = [
"addressing",
"log 0.4.8",
"nymsphinx",
"pretty_env_logger",
"rand 0.7.3",
"rand_distr",
@@ -1535,9 +1527,8 @@ dependencies = [
[[package]]
name = "nym-client"
version = "0.5.0"
version = "0.6.0"
dependencies = [
"addressing",
"bs58",
"built",
"clap",
@@ -1552,6 +1543,7 @@ dependencies = [
"log 0.4.8",
"mix-client",
"multi-tcp-client",
"nymsphinx",
"pem",
"pemstore",
"pretty_env_logger",
@@ -1569,9 +1561,8 @@ dependencies = [
[[package]]
name = "nym-mixnode"
version = "0.5.0"
version = "0.6.0"
dependencies = [
"addressing",
"bs58",
"built",
"clap",
@@ -1584,17 +1575,19 @@ dependencies = [
"futures 0.3.4",
"log 0.4.8",
"multi-tcp-client",
"nymsphinx",
"pemstore",
"pretty_env_logger",
"serde",
"sphinx",
"tempfile",
"tokio 0.2.12",
"topology",
]
[[package]]
name = "nym-sfw-provider"
version = "0.5.0"
version = "0.6.0"
dependencies = [
"bs58",
"built",
@@ -1619,11 +1612,12 @@ dependencies = [
"sphinx",
"tempfile",
"tokio 0.2.12",
"topology",
]
[[package]]
name = "nym-validator"
version = "0.5.0"
version = "0.6.0"
dependencies = [
"abci",
"built",
@@ -1647,6 +1641,15 @@ dependencies = [
"topology",
]
[[package]]
name = "nymsphinx"
version = "0.1.0"
dependencies = [
"log 0.4.8",
"rand 0.7.3",
"sphinx",
]
[[package]]
name = "opaque-debug"
version = "0.2.3"
@@ -2497,7 +2500,7 @@ dependencies = [
[[package]]
name = "sphinx"
version = "0.1.0"
source = "git+https://github.com/nymtech/sphinx?rev=23f9c89b257ee0936e70afd682e9ed6a62e89eee#23f9c89b257ee0936e70afd682e9ed6a62e89eee"
source = "git+https://github.com/nymtech/sphinx?rev=44d8f2aece5049eaa4fe84b7948758ce82b4b80d#44d8f2aece5049eaa4fe84b7948758ce82b4b80d"
dependencies = [
"aes-ctr",
"arrayref",
@@ -2844,10 +2847,10 @@ dependencies = [
name = "topology"
version = "0.1.0"
dependencies = [
"addressing",
"bs58",
"itertools",
"log 0.4.8",
"nymsphinx",
"pretty_env_logger",
"rand 0.7.3",
"serde",
+1 -1
View File
@@ -9,10 +9,10 @@ members = [
"common/clients/multi-tcp-client",
"common/clients/provider-client",
"common/clients/validator-client",
"common/addressing",
"common/config",
"common/crypto",
"common/healthcheck",
"common/nymsphinx",
"common/pemstore",
"common/topology",
"mixnode",
+4
View File
@@ -18,3 +18,7 @@ Platform build instructions are available on [our docs site](https://nymtech.net
### Developing
There's a `.env.sample-dev` file provided which you can rename to `.env` if you want convenient logging, backtrace, or other environment variables pre-set. The `.env` file is ignored so you don't need to worry about checking it in.
### Developer chat
You can chat to us in [Keybase](https://keybase.io). Download their chat app, then click **Teams -> Join a team**. Type **nymtech.friends** into the team name and hit **continue**. For general chat, hang out in the **#general** channel. Our development takes places in the **#dev** channel. Node operators should be in the **#node-operators** channel.
-80
View File
@@ -1,80 +0,0 @@
use std::convert::{TryFrom, TryInto};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
pub enum AddressType {
V4,
V6,
}
impl Into<u8> for AddressType {
fn into(self) -> u8 {
use AddressType::*;
match self {
V4 => 4,
V6 => 6,
}
}
}
#[derive(Debug)]
pub enum AddressTypeError {
InvalidPrefixError,
}
impl TryFrom<u8> for AddressType {
type Error = AddressTypeError;
fn try_from(value: u8) -> Result<Self, Self::Error> {
use AddressType::*;
use AddressTypeError::*;
match value {
4 => Ok(V4),
6 => Ok(V6),
_ => Err(InvalidPrefixError),
}
}
}
/// FLAG || port || octets || zeropad
pub fn encoded_bytes_from_socket_address(address: SocketAddr) -> [u8; 32] {
let port_bytes = address.port().to_be_bytes();
let encoded_host: Vec<u8> = match address.ip() {
IpAddr::V4(ip) => std::iter::once(AddressType::V4.into())
.chain(port_bytes.iter().cloned())
.chain(ip.octets().iter().cloned())
.chain(std::iter::repeat(0))
.take(32)
.collect(),
IpAddr::V6(ip) => std::iter::once(AddressType::V6.into())
.chain(port_bytes.iter().cloned())
.chain(ip.octets().iter().cloned())
.chain(std::iter::repeat(0))
.take(32)
.collect(),
};
let mut address_bytes = [0u8; 32];
address_bytes.copy_from_slice(&encoded_host[..32]);
address_bytes
}
pub fn socket_address_from_encoded_bytes(b: [u8; 32]) -> Result<SocketAddr, AddressTypeError> {
let address_type: AddressType = b[0].try_into()?;
let port: u16 = u16::from_be_bytes([b[1], b[2]]);
let ip = match address_type {
AddressType::V4 => IpAddr::V4(Ipv4Addr::new(b[3], b[4], b[5], b[6])),
AddressType::V6 => {
let mut address_octets = [0u8; 16];
address_octets.copy_from_slice(&b[3..19]);
IpAddr::V6(Ipv6Addr::from(address_octets))
}
};
Ok(SocketAddr::new(ip, port))
}
@@ -86,7 +86,7 @@ mod converting_mixnode_presence_into_topology_mixnode {
version: "".to_string(),
};
let result: Result<mix::Node, std::io::Error> = mix_presence.try_into();
let _result: Result<mix::Node, std::io::Error> = mix_presence.try_into();
// assert!(result.is_err()) // This fails only for me. Why?
// ¯\_(ツ)_/¯ - works on my machine (and travis)
}
+2 -2
View File
@@ -14,10 +14,10 @@ rand_distr = "0.2.2"
tokio = { version = "0.2", features = ["full"] }
## internal
addressing = {path = "../../addressing"}
nymsphinx = {path = "../../nymsphinx"}
topology = {path = "../../topology"}
## will be moved to proper dependencies once released
sphinx = { git = "https://github.com/nymtech/sphinx", rev="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
# sphinx = { path = "../../../../sphinx"}
+8 -8
View File
@@ -1,7 +1,7 @@
use addressing;
use addressing::AddressTypeError;
use nymsphinx::addressing::nodes::{NymNodeRoutingAddress, NymNodeRoutingAddressError};
use sphinx::route::{Destination, DestinationAddressBytes, SURBIdentifier};
use sphinx::SphinxPacket;
use std::convert::TryFrom;
use std::net::SocketAddr;
use std::time;
use topology::{NymTopology, NymTopologyError};
@@ -32,8 +32,8 @@ impl From<sphinx::header::SphinxUnwrapError> for SphinxPacketEncapsulationError
}
}
impl From<AddressTypeError> for SphinxPacketEncapsulationError {
fn from(_: AddressTypeError) -> Self {
impl From<NymNodeRoutingAddressError> for SphinxPacketEncapsulationError {
fn from(_: NymNodeRoutingAddressError) -> Self {
use SphinxPacketEncapsulationError::*;
InvalidFirstMixAddress
}
@@ -96,9 +96,9 @@ pub fn encapsulate_message<T: NymTopology>(
// we know the mix route must be valid otherwise we would have already returned an error
let first_node_address =
addressing::socket_address_from_encoded_bytes(route.first().unwrap().address.to_bytes())?;
NymNodeRoutingAddress::try_from(route.first().unwrap().address.clone())?;
Ok((first_node_address, packet))
Ok((first_node_address.into(), packet))
}
pub fn encapsulate_message_route(
@@ -113,7 +113,7 @@ pub fn encapsulate_message_route(
let packet = sphinx::SphinxPacket::new(message, &route[..], &recipient, &delays)?;
let first_node_address =
addressing::socket_address_from_encoded_bytes(route.first().unwrap().address.to_bytes())?;
NymNodeRoutingAddress::try_from(route.first().unwrap().address.clone())?;
Ok((first_node_address, packet))
Ok((first_node_address.into(), packet))
}
@@ -1,6 +1,7 @@
use crate::connection_manager::reconnector::ConnectionReconnector;
use crate::connection_manager::writer::ConnectionWriter;
use futures::channel::{mpsc, oneshot};
use futures::future::{abortable, AbortHandle};
use futures::task::Poll;
use futures::{AsyncWriteExt, StreamExt};
use log::*;
@@ -34,22 +35,32 @@ pub(crate) struct ConnectionManager<'a> {
state: ConnectionState<'a>,
}
impl<'a> Drop for ConnectionManager<'a> {
fn drop(&mut self) {
debug!("Connection manager to {:?} is being dropped", self.address)
}
}
impl<'a> ConnectionManager<'static> {
pub(crate) async fn new(
address: SocketAddr,
reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
connection_timeout: Duration,
) -> ConnectionManager<'a> {
let (conn_tx, conn_rx) = mpsc::unbounded();
// based on initial connection we will either have a writer or a reconnector
let state = match tokio::net::TcpStream::connect(address).await {
Ok(conn) => ConnectionState::Writing(ConnectionWriter::new(conn)),
// the blocking call here is fine as initially we want to wait the timeout interval (at most) anyway:
let tcp_stream_res = std::net::TcpStream::connect_timeout(&address, connection_timeout);
let initial_state = match tcp_stream_res {
Ok(stream) => {
let tokio_stream = tokio::net::TcpStream::from_std(stream).unwrap();
debug!("managed to establish initial connection to {}", address);
ConnectionState::Writing(ConnectionWriter::new(tokio_stream))
}
Err(e) => {
warn!(
"failed to establish initial connection to {} ({}). Going into reconnection mode",
address, e
);
warn!("failed to establish initial connection to {} within {:?} ({}). Going into reconnection mode", address, connection_timeout, e);
ConnectionState::Reconnecting(ConnectionReconnector::new(
address,
reconnection_backoff,
@@ -64,28 +75,33 @@ impl<'a> ConnectionManager<'static> {
address,
maximum_reconnection_backoff,
reconnection_backoff,
state,
state: initial_state,
}
}
/// consumes Self and returns channel for communication
pub(crate) fn start(mut self, handle: &Handle) -> ConnectionManagerSender {
let sender_clone = self.conn_tx.clone();
handle.spawn(async move {
while let Some(msg) = self.conn_rx.next().await {
let (msg_content, res_ch) = msg;
let res = self.handle_new_message(msg_content).await;
if let Some(res_ch) = res_ch {
if let Err(e) = res_ch.send(res) {
error!(
"failed to send response on the channel to the caller! - {:?}",
e
);
}
async fn run(mut self) {
while let Some(msg) = self.conn_rx.next().await {
let (msg_content, res_ch) = msg;
let res = self.handle_new_message(msg_content).await;
if let Some(res_ch) = res_ch {
if let Err(e) = res_ch.send(res) {
error!(
"failed to send response on the channel to the caller! - {:?}",
e
);
}
}
});
sender_clone
}
}
/// consumes Self and returns channel for communication as well as an `AbortHandle`
pub(crate) fn start_abortable(self, handle: &Handle) -> (ConnectionManagerSender, AbortHandle) {
let sender_clone = self.conn_tx.clone();
let (abort_fut, abort_handle) = abortable(self.run());
handle.spawn(async move { abort_fut.await });
(sender_clone, abort_handle)
}
async fn handle_new_message(&mut self, msg: Vec<u8>) -> io::Result<()> {
@@ -60,13 +60,25 @@ impl<'a> Future for ConnectionReconnector<'a> {
self.current_retry_attempt += 1;
// we failed to re-establish connection - continue exponential backoff
// according to https://github.com/tokio-rs/tokio/issues/1953 there's an undocumented
// limit of tokio delay of about 2 years.
// let's ensure our delay is always on a sane side of being maximum 1 day.
let maximum_sane_delay = Duration::from_secs(24 * 60 * 60);
let next_delay = std::cmp::min(
self.maximum_reconnection_backoff,
2_u32.pow(self.current_retry_attempt) * self.initial_reconnection_backoff,
maximum_sane_delay,
self.initial_reconnection_backoff
.checked_mul(2_u32.pow(self.current_retry_attempt))
.unwrap_or_else(|| self.maximum_reconnection_backoff),
);
self.current_backoff_delay
.reset(tokio::time::Instant::now() + next_delay);
let now = self.current_backoff_delay.deadline();
// this can't overflow now because next_delay is limited to one day...
// ... well, unless you've been running the system for couple of decades without
// any restarts....
let next = now + next_delay;
self.current_backoff_delay.reset(next);
self.connection = tokio::net::TcpStream::connect(self.address).boxed();
+34 -11
View File
@@ -1,5 +1,6 @@
use crate::connection_manager::{ConnectionManager, ConnectionManagerSender};
use futures::channel::oneshot;
use futures::future::AbortHandle;
use log::*;
use std::collections::HashMap;
use std::io;
@@ -12,25 +13,29 @@ mod connection_manager;
pub struct Config {
initial_reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
initial_connection_timeout: Duration,
}
impl Config {
pub fn new(
initial_reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
initial_connection_timeout: Duration,
) -> Self {
Config {
initial_reconnection_backoff,
maximum_reconnection_backoff,
initial_connection_timeout,
}
}
}
pub struct Client {
runtime_handle: Handle,
connections_managers: HashMap<SocketAddr, ConnectionManagerSender>,
connections_managers: HashMap<SocketAddr, (ConnectionManagerSender, AbortHandle)>,
maximum_reconnection_backoff: Duration,
initial_reconnection_backoff: Duration,
initial_connection_timeout: Duration,
}
impl Client {
@@ -43,17 +48,24 @@ impl Client {
connections_managers: HashMap::new(),
initial_reconnection_backoff: config.maximum_reconnection_backoff,
maximum_reconnection_backoff: config.initial_reconnection_backoff,
initial_connection_timeout: config.initial_connection_timeout,
}
}
async fn start_new_connection_manager(&self, address: SocketAddr) -> ConnectionManagerSender {
ConnectionManager::new(
async fn start_new_connection_manager(
&mut self,
address: SocketAddr,
) -> (ConnectionManagerSender, AbortHandle) {
let (sender, abort_handle) = ConnectionManager::new(
address,
self.initial_reconnection_backoff,
self.maximum_reconnection_backoff,
self.initial_connection_timeout,
)
.await
.start(&self.runtime_handle)
.start_abortable(&self.runtime_handle);
(sender, abort_handle)
}
// if wait_for_response is set to true, we will get information about any possible IO errors
@@ -66,29 +78,38 @@ impl Client {
wait_for_response: bool,
) -> io::Result<()> {
if !self.connections_managers.contains_key(&address) {
info!(
debug!(
"There is no existing connection to {:?} - it will be established now",
address
);
let new_manager_sender = self.start_new_connection_manager(address).await;
let (new_manager_sender, abort_handle) =
self.start_new_connection_manager(address).await;
self.connections_managers
.insert(address, new_manager_sender);
.insert(address, (new_manager_sender, abort_handle));
}
let manager = self.connections_managers.get_mut(&address).unwrap();
if wait_for_response {
let (res_tx, res_rx) = oneshot::channel();
manager.unbounded_send((message, Some(res_tx))).unwrap();
manager.0.unbounded_send((message, Some(res_tx))).unwrap();
res_rx.await.unwrap()
} else {
manager.unbounded_send((message, None)).unwrap();
manager.0.unbounded_send((message, None)).unwrap();
Ok(())
}
}
}
impl Drop for Client {
fn drop(&mut self) {
for (_, abort_handle) in self.connections_managers.values() {
abort_handle.abort()
}
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -150,7 +171,8 @@ mod tests {
let mut rt = tokio::runtime::Runtime::new().unwrap();
let addr = "127.0.0.1:6000".parse().unwrap();
let reconnection_backoff = Duration::from_secs(1);
let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff);
let timeout = Duration::from_secs(1);
let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff, timeout);
let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()];
@@ -209,7 +231,8 @@ mod tests {
let mut rt = tokio::runtime::Runtime::new().unwrap();
let addr = "127.0.0.1:6001".parse().unwrap();
let reconnection_backoff = Duration::from_secs(2);
let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff);
let timeout = Duration::from_secs(1);
let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff, timeout);
let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()];
+1 -1
View File
@@ -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="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
+1 -1
View File
@@ -15,4 +15,4 @@ rand = "0.7.2"
rand_os = "0.1"
## will be moved to proper dependencies once released
sphinx = { git = "https://github.com/nymtech/sphinx", rev="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
+3 -3
View File
@@ -18,16 +18,16 @@ serde_derive = "1.0.104"
tokio = { version = "0.2", features = ["full"] }
## internal
addressing = {path = "../addressing" }
crypto = { path = "../crypto" }
directory-client = { path = "../clients/directory-client" }
mix-client = { path = "../clients/mix-client" }
multi-tcp-client = { path = "../clients/multi-tcp-client" }
nymsphinx = {path = "../nymsphinx" }
provider-client = { path = "../clients/provider-client" }
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="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
# sphinx = { path = "../../../sphinx"}
[dev-dependencies]
+4
View File
@@ -36,17 +36,20 @@ impl From<topology::NymTopologyError> for HealthCheckerError {
pub struct HealthChecker {
num_test_packets: usize,
resolution_timeout: Duration,
connection_timeout: Duration,
identity_keypair: MixIdentityKeyPair,
}
impl HealthChecker {
pub fn new(
resolution_timeout: Duration,
connection_timeout: Duration,
num_test_packets: usize,
identity_keypair: MixIdentityKeyPair,
) -> Self {
HealthChecker {
resolution_timeout,
connection_timeout,
num_test_packets,
identity_keypair,
}
@@ -62,6 +65,7 @@ impl HealthChecker {
current_topology,
self.num_test_packets,
self.resolution_timeout,
self.connection_timeout,
&self.identity_keypair,
)
.await;
+27 -34
View File
@@ -1,11 +1,14 @@
use crypto::identity::MixIdentityKeyPair;
use itertools::Itertools;
use log::{debug, error, info, trace, warn};
use mix_client::MixClient;
use nymsphinx::addressing::nodes::NymNodeRoutingAddress;
use provider_client::{ProviderClient, ProviderClientError};
use sphinx::header::delays::Delay;
use sphinx::route::{Destination, Node as SphinxNode};
use std::collections::HashMap;
use std::convert::TryFrom;
use std::net::SocketAddr;
use std::time::Duration;
use topology::provider;
#[derive(Debug, PartialEq, Clone)]
@@ -17,10 +20,7 @@ pub enum PathStatus {
pub(crate) struct PathChecker {
provider_clients: HashMap<[u8; 32], Option<ProviderClient>>,
// currently this is an overkill as MixClient is extremely cheap to create,
// however, once we introduce persistent connection between client and layer one mixes,
// this will be extremely helpful to have
layer_one_clients: HashMap<[u8; 32], Option<MixClient>>,
mixnet_client: multi_tcp_client::Client,
paths_status: HashMap<Vec<u8>, PathStatus>,
our_destination: Destination,
check_id: [u8; 16],
@@ -30,6 +30,7 @@ impl PathChecker {
pub(crate) async fn new(
providers: Vec<provider::Node>,
identity_keys: &MixIdentityKeyPair,
connection_timeout: Duration,
check_id: [u8; 16],
) -> Self {
let mut provider_clients = HashMap::new();
@@ -41,12 +42,12 @@ impl PathChecker {
ProviderClient::new(provider.client_listener, address.clone(), None);
let insertion_result = match provider_client.register().await {
Ok(token) => {
debug!("registered at provider {}", provider.pub_key);
debug!("[Healthcheck] registered at provider {}", provider.pub_key);
provider_client.update_token(token);
provider_clients.insert(provider.get_pub_key_bytes(), Some(provider_client))
}
Err(ProviderClientError::ClientAlreadyRegisteredError) => {
info!("We were already registered");
info!("[Healthcheck] We were already registered");
provider_clients.insert(provider.get_pub_key_bytes(), Some(provider_client))
}
Err(err) => {
@@ -63,9 +64,16 @@ impl PathChecker {
}
}
// there's no reconnection allowed - if it fails, then it fails.
let mixnet_client_config = multi_tcp_client::Config::new(
Duration::from_secs(1_000_000_000),
Duration::from_secs(1_000_000_000),
connection_timeout,
);
PathChecker {
provider_clients,
layer_one_clients: HashMap::new(),
mixnet_client: multi_tcp_client::Client::new(mixnet_client_config),
our_destination: Destination::new(address, Default::default()),
paths_status: HashMap::new(),
check_id,
@@ -180,7 +188,7 @@ impl PathChecker {
return;
}
debug!("Checking path: {:?} ({})", path, iteration);
trace!("Checking path: {:?} ({})", path, iteration);
let path_identifier = PathChecker::unique_path_key(path, self.check_id, iteration);
// check if there is even any point in sending the packet
@@ -212,32 +220,12 @@ impl PathChecker {
let layer_one_mix = path
.first()
.expect("We checked the path to contain at least one entry");
let first_node_key = layer_one_mix.pub_key.to_bytes();
// we generated the bytes data so unwrap is fine
let first_node_address =
addressing::socket_address_from_encoded_bytes(layer_one_mix.address.to_bytes())
.unwrap();
let first_node_client = self
.layer_one_clients
.entry(first_node_key)
.or_insert_with(|| Some(mix_client::MixClient::new()));
if first_node_client.is_none() {
debug!("we can ignore this path as layer one mix is inaccessible");
if self
.paths_status
.insert(path_identifier, PathStatus::Unhealthy)
.is_some()
{
panic!("Overwriting path checks!")
}
return;
}
// we already checked for 'None' case
let first_node_client = first_node_client.as_ref().unwrap();
let first_node_address: SocketAddr =
NymNodeRoutingAddress::try_from(layer_one_mix.address.clone())
.unwrap()
.into();
let delays: Vec<_> = path.iter().map(|_| Delay::new_from_nanos(0)).collect();
@@ -251,7 +239,12 @@ impl PathChecker {
.unwrap();
debug!("sending test packet to {}", first_node_address);
match first_node_client.send(packet, first_node_address).await {
match self
.mixnet_client
.send(first_node_address, packet.to_bytes(), true)
.await
{
Err(err) => {
debug!("failed to send packet to {} - {}", first_node_address, err);
if self
+3 -1
View File
@@ -102,6 +102,7 @@ impl HealthCheckResult {
topology: &T,
iterations: usize,
resolution_timeout: Duration,
connection_timeout: Duration,
identity_keys: &MixIdentityKeyPair,
) -> Self {
// currently healthchecker supports only up to 255 iterations - if we somehow
@@ -127,7 +128,8 @@ impl HealthCheckResult {
let providers = topology.providers();
let mut path_checker = PathChecker::new(providers, identity_keys, check_id).await;
let mut path_checker =
PathChecker::new(providers, identity_keys, connection_timeout, check_id).await;
for i in 0..iterations {
debug!("running healthcheck iteration {} / {}", i + 1, iterations);
for path in &all_paths {
+1 -1
View File
@@ -153,7 +153,7 @@ impl std::fmt::Display for NodeScore {
let stringified_key = self.pub_key.to_base58_string();
write!(
f,
"({})\t{}/{}\t({}%)\t|| {}\tv{} <{}> - {}",
"({})\t\t{}/{}\t({:.2}%) \t|| {}\tv{} <{}> - {}",
self.typ,
self.packets_received,
self.packets_sent,
@@ -1,5 +1,5 @@
[package]
name = "addressing"
name = "nymsphinx"
version = "0.1.0"
authors = ["Jedrzej Stuczynski <andrew@nymtech.net>"]
edition = "2018"
@@ -7,5 +7,9 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
log = "0.4"
pretty_env_logger = "0.3"
log = "0.4.8"
rand = "0.7.3"
## will be moved to proper dependencies once released
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
+1
View File
@@ -0,0 +1 @@
pub mod nodes;
+206
View File
@@ -0,0 +1,206 @@
use sphinx::constants::NODE_ADDRESS_LENGTH;
use sphinx::route::NodeAddressBytes;
use std::convert::{TryFrom, TryInto};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
/// This module is responsible for encoding and decoding node routing information, so that
/// they could be later put into an appropriate field in a sphinx header.
/// Currently, that routing information is an IP address, but in principle it can be anything
/// for as long as it's going to fit in the field.
#[derive(Debug)]
pub enum NymNodeRoutingAddressError {
InsufficientNumberOfBytesAvailableError,
InvalidIPVersion,
}
/// Current representation of Node routing information used in Nym system.
/// At this point of time it is a simple `SocketAddr`.
#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
pub struct NymNodeRoutingAddress(SocketAddr);
impl NymNodeRoutingAddress {
/// Minimum number of bytes that need to be available to represent self.
/// The value has no upper bound as when converted into bytes, it's always
/// padded with zeroes to be exactly NODE_ADDRESS_LENGTH long.
pub fn bytes_min_len(&self) -> usize {
match self.0 {
SocketAddr::V4(_) => 7,
SocketAddr::V6(_) => 19,
}
}
/// Converts self into a vector of bytes.
/// Note, this represents a generic bytes vector, not necessarily a NodeAddressBytes
/// and hence is not zero-padded.
pub fn as_bytes(&self) -> Vec<u8> {
let port_bytes = self.0.port().to_be_bytes();
let ip_octets_vec = match self.0.ip() {
IpAddr::V4(ip) => ip.octets().to_vec(),
IpAddr::V6(ip) => ip.octets().to_vec(),
};
std::iter::once(self.addr_type_as_u8())
.chain(port_bytes.iter().cloned())
.chain(ip_octets_vec.iter().cloned())
.collect()
}
/// Tries to recover `Self` from a bytes slice.
/// Does not care if it's zero-padded or not.
pub fn try_from_bytes(b: &[u8]) -> Result<Self, NymNodeRoutingAddressError> {
// the bare minimum to represent `Self` is 7 bytes (for the shorter V4 version)
if b.len() < 7 {
return Err(NymNodeRoutingAddressError::InsufficientNumberOfBytesAvailableError);
}
let ip_version = b[0];
let port: u16 = u16::from_be_bytes([b[1], b[2]]);
let ip = match ip_version {
4 => IpAddr::V4(Ipv4Addr::new(b[3], b[4], b[5], b[6])),
6 => {
if b.len() < 19 {
return Err(
NymNodeRoutingAddressError::InsufficientNumberOfBytesAvailableError,
);
}
let mut address_octets = [0u8; 16];
address_octets.copy_from_slice(&b[3..19]);
IpAddr::V6(Ipv6Addr::from(address_octets))
}
_ => return Err(NymNodeRoutingAddressError::InvalidIPVersion),
};
Ok(Self(SocketAddr::new(ip, port)))
}
/// Single byte representation of self ip version.
pub fn addr_type_as_u8(&self) -> u8 {
match self.0 {
SocketAddr::V4(_) => 4,
SocketAddr::V6(_) => 6,
}
}
}
/// Considering `NymNodeRoutingAddress` is equivalent to a `SocketAddr` at this point,
/// it makes perfect sense to allow the bilateral transformation.
impl From<SocketAddr> for NymNodeRoutingAddress {
fn from(addr: SocketAddr) -> Self {
Self(addr)
}
}
/// Considering `NymNodeRoutingAddress` is equivalent to a `SocketAddr` at this point,
/// it makes perfect sense to allow the bilateral transformation.
impl Into<SocketAddr> for NymNodeRoutingAddress {
fn into(self) -> SocketAddr {
self.0
}
}
impl TryInto<NodeAddressBytes> for NymNodeRoutingAddress {
type Error = NymNodeRoutingAddressError;
/// `NymNodeRoutingAddress` (as a `SocketAddr`) is represented the following way:
/// VersionFlag || port || octets || zeropad
/// VersionFlag is one byte representing whether self is ipv4 or ipv6 address,
/// port is 16bit big endian representation of port value
/// octets is bytes representation of octets making up the ip address of the socket address
/// (either 4 bytes for ipv4 or 16 bytes for ipv6)
/// zeropad is padding of 0 for the `NymNodeRoutingAddress` to be
/// exactly `NODE_ADDRESS_LENGTH` long.
fn try_into(self) -> Result<NodeAddressBytes, Self::Error> {
// first check if we have enough bytes to represent `self`:
if self.bytes_min_len() > NODE_ADDRESS_LENGTH {
return Err(NymNodeRoutingAddressError::InsufficientNumberOfBytesAvailableError);
}
let unpadded_address = self.as_bytes();
let padded_address: Vec<_> = unpadded_address
.into_iter()
.chain(std::iter::repeat(0))
.take(NODE_ADDRESS_LENGTH)
.collect();
let mut node_address_bytes = [0u8; 32];
node_address_bytes.copy_from_slice(&padded_address);
Ok(NodeAddressBytes::from_bytes(node_address_bytes))
}
}
impl TryFrom<NodeAddressBytes> for NymNodeRoutingAddress {
type Error = NymNodeRoutingAddressError;
fn try_from(value: NodeAddressBytes) -> Result<Self, Self::Error> {
Self::try_from_bytes(value.as_bytes())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_v4_address() {
let address = NymNodeRoutingAddress(SocketAddr::new(IpAddr::from([1, 2, 3, 4]), 42));
let address_bytes = address.as_bytes();
assert_eq!(
address,
NymNodeRoutingAddress::try_from_bytes(&address_bytes).unwrap()
)
}
#[test]
fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_v6_address() {
let address = NymNodeRoutingAddress(SocketAddr::new(
IpAddr::from([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]),
42,
));
let address_bytes = address.as_bytes();
assert_eq!(
address,
NymNodeRoutingAddress::try_from_bytes(&address_bytes).unwrap()
)
}
#[test]
fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_empty_v4_address() {
let address = NymNodeRoutingAddress(SocketAddr::new(IpAddr::from([0, 0, 0, 0]), 42));
let address_bytes = address.as_bytes();
assert_eq!(
address,
NymNodeRoutingAddress::try_from_bytes(&address_bytes).unwrap()
)
}
#[test]
fn nym_node_routing_address_can_be_converted_to_and_from_bytes_for_empty_v6_address() {
let address = NymNodeRoutingAddress(SocketAddr::new(
IpAddr::from([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]),
42,
));
let address_bytes = address.as_bytes();
assert_eq!(
address,
NymNodeRoutingAddress::try_from_bytes(&address_bytes).unwrap()
)
}
#[test]
fn nym_node_routing_address_can_be_converted_to_and_from_node_address_bytes_with_no_data_loss()
{
let address_v4 = NymNodeRoutingAddress(SocketAddr::new(IpAddr::from([1, 2, 3, 4]), 42));
let address_v6 = NymNodeRoutingAddress(SocketAddr::new(
IpAddr::from([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]),
42,
));
let node_address1: NodeAddressBytes = address_v4.try_into().unwrap();
let node_address2: NodeAddressBytes = address_v6.try_into().unwrap();
assert_eq!(address_v4, node_address1.try_into().unwrap());
assert_eq!(address_v6, node_address2.try_into().unwrap());
}
}
File diff suppressed because it is too large Load Diff
+56
View File
@@ -0,0 +1,56 @@
use crate::chunking::set::split_into_sets;
pub mod fragment;
pub mod reconstruction;
pub mod set;
/// The idea behind the process of chunking is to incur as little data overhead as possible due
/// to very computationally costly sphinx encapsulation procedure.
///
/// To achieve this, the underlying message is split into so-called "sets", which are further
/// subdivided into the base unit of "fragment" that is directly encapsulated by a Sphinx packet.
/// This allows to encapsulate messages of arbitrary length.
///
/// Each message, regardless of its size, consists of at least a single `Set` that has at least
/// a single `Fragment`.
///
/// Each `Fragment` can have variable, yet fully deterministic, length,
/// that depends on its position in the set as well as total number of sets. This is further
/// explained in `fragment.rs` file.
///
/// Similarly, each `Set` can have a variable number of `Fragment`s inside. However, that
/// value is more restrictive: if it's the last set into which the message was split
/// (or implicitly the only one), it has no lower bound on the number of `Fragment`s.
/// (Apart from the restriction of containing at least a single one). If the set is located
/// somewhere in the middle, *it must be* full. Finally, regardless of its position, it must also be
/// true that it contains no more than `u8::max_value()`, i.e. 255 `Fragment`s.
/// Again, the reasoning for this is further explained in `set.rs` file. However, you might
/// also want to look at `fragment.rs` to understand the full context behind that design choice.
///
/// Both of those concepts as well as their structures, i.e. `Set` and `Fragment`
/// are further explained in the respective files.
#[derive(PartialEq, Debug)]
pub enum ChunkingError {
InvalidPayloadLengthError,
TooBigMessageToSplit,
MalformedHeaderError,
NoValidProvidersError,
NoValidRoutesAvailableError,
InvalidTopologyError,
TooShortFragmentData,
MalformedFragmentData,
UnexpectedFragmentCount,
}
/// Takes the entire message and splits it into bytes chunks that will fit into sphinx packets
/// directly. After receiving they can be combined using `reconstruction::MessageReconstructor`
/// to obtain the original message back.
pub fn split_and_prepare_payloads(message: &[u8]) -> Vec<Vec<u8>> {
let fragmented_messages = split_into_sets(message);
fragmented_messages
.into_iter()
.flat_map(|fragment_set| fragment_set.into_iter())
.map(|fragment| fragment.into_bytes())
.collect()
}
File diff suppressed because it is too large Load Diff
+788
View File
@@ -0,0 +1,788 @@
use crate::chunking::fragment::{
Fragment, LINKED_FRAGMENTED_HEADER_LEN, LINKED_FRAGMENTED_PAYLOAD_MAX_LEN,
UNFRAGMENTED_PAYLOAD_MAX_LEN, UNLINKED_FRAGMENTED_HEADER_LEN,
UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN,
};
use rand::Rng;
/// In the simplest case of message being divided into a single set, the set has the upper bound
/// on its payload length of the maximum number of `Fragment`s multiplied by their maximum,
/// fragmented, length.
pub const MAX_UNLINKED_SET_PAYLOAD_LENGTH: usize =
u8::max_value() as usize * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN;
/// If the set is being linked to another one, by either being the very first set, or the very last,
/// one of its `Fragment`s must be changed from "unlinked" into "linked" to compensate for a tiny
/// bit extra data overhead: id of the other set.
/// Note that the "MAX" prefix only applies to if the set is the last one as it does not have
/// a lower bound on its length. If the set is one way linked and a first one, it *must have*
/// this exact payload length instead.
pub const MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH: usize = MAX_UNLINKED_SET_PAYLOAD_LENGTH
- (LINKED_FRAGMENTED_HEADER_LEN - UNLINKED_FRAGMENTED_HEADER_LEN);
/// If the set is being linked two others sets by being stuck in the middle of divided message,
/// two of its `Fragment`s (first and final one) must be changed from
/// "unlinked" into "linked" to compensate for data overhead.
/// Note that this constant no longer has a "MAX" prefix, this is because each set being stuck
/// between different sets, *must* have this exact payload length.
pub const TWO_WAY_LINKED_SET_PAYLOAD_LENGTH: usize = MAX_UNLINKED_SET_PAYLOAD_LENGTH
- 2 * (LINKED_FRAGMENTED_HEADER_LEN - UNLINKED_FRAGMENTED_HEADER_LEN);
/// `FragmentSet` is an ordered collection of 1 to 255 `Fragment`s, each with the same ID
/// that can be used to produce original message, assuming no linking took place.
///
/// Otherwise, if set linking took place, then first or last `Fragment` from the `FragmentSet`
/// is used to determine preceding or succeeding other `FragmentSet`
/// that should be used in tandem to reconstruct original message. The linking reconstruction
/// is a recursive process as a message could have been divided into an arbitrary number of
/// `FragmentSet`s with no upper bound at all.
///
/// For example if a message was divided into 300 `Fragment`s (i.e. 2 `FragmentSet`s,
/// the structures might look as follows:
///
/// Set1: [f1 {id = 12345}, f2 {id = 12345}, ... f255 {id = 12345, next_id = 54321}]
/// Set2: [f1 {id = 54321, previous_id = 12345}, f2 {id = 54321}, ... f45 {id = 54321}]
pub(crate) type FragmentSet = Vec<Fragment>;
/// Generate a pseudo-random id for a `FragmentSet`.
/// Its value is restricted to (0, i32::max_value()].
/// Note that it *excludes* 0, but *includes* i32::max_value().
/// This particular range allows for the id to be represented using 31bits, rather than
/// the full length of 32 while still providing more than enough variability to
/// distinguish different `FragmentSet`s.
/// The extra bit, as explained in `Fragment` definition is used to represents additional information,
/// indicating how further bytes should be parsed.
/// This approach saves whole byte per `Fragment`, which while may seem insignificant and
/// introduces extra complexity, quickly adds up when faced with sphinx packet encapsulation for longer
/// messages.
/// Finally, the reason 0 id is not allowed is to explicitly distinguish it from unfragmented
/// `Fragment`s thus allowing for some additional optimizations by letting it skip
/// certain procedures when reconstructing.
fn generate_set_id<R: Rng>(rng: &mut R) -> i32 {
let potential_id = rng.gen::<i32>().abs();
// make sure id is always non-zero, as we do not want to accidentally have weird
// reconstruction cases where unfragmented payload overwrites some part of set with id0
if potential_id == 0 {
generate_set_id(rng)
} else {
potential_id
}
}
/// The simplest case of splitting underlying message - when it fits into a single
/// `Fragment` thus requiring no linking or even a set id.
/// For obvious reasons the most efficient approach.
fn prepare_unfragmented_set(message: &[u8]) -> FragmentSet {
vec![Fragment::try_new_unfragmented(&message).unwrap()]
}
/// Splits underlying message into multiple `Fragment`s while all of them fit in a single
/// `Set` (number of `Fragment`s <= 255)
fn prepare_unlinked_fragmented_set(message: &[u8], id: i32) -> FragmentSet {
let pre_casted_frags =
(message.len() as f64 / UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN as f64).ceil() as usize;
debug_assert!(pre_casted_frags <= u8::max_value() as usize);
let num_fragments = pre_casted_frags as u8;
let mut fragments = Vec::with_capacity(num_fragments as usize);
for i in 1..(pre_casted_frags + 1) {
// we can't use u8 directly here as upper (NON-INCLUSIVE, so it would always fit) bound could be u8::max_value() + 1
let lb = (i as usize - 1) * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN;
let ub = usize::min(
message.len(),
i as usize * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN,
);
fragments.push(
Fragment::try_new_fragmented(&message[lb..ub], id, num_fragments, i as u8, None, None)
.unwrap(),
)
}
fragments
}
/// Similarly to `prepare_unlinked_fragmented_set`, splits part of underlying message into
/// multiple `Fragment`s. The byte slice of the message *must* fit into a single linked set, however,
/// the whole message itself is still longer than a single `Set` (number of `Fragment`s > 255).
/// During the process of splitting message, this function is called multiple times.
fn prepare_linked_fragment_set(
message: &[u8],
id: i32,
previous_link_id: Option<i32>,
next_link_id: Option<i32>,
) -> FragmentSet {
// determine number of fragments in the set:
let num_frags_usize = if next_link_id.is_some() {
u8::max_value() as usize
} else {
// we know this set is linked, if it's not post-linked then it MUST BE pre-linked
let tail_len = if message.len() >= LINKED_FRAGMENTED_PAYLOAD_MAX_LEN {
message.len() - LINKED_FRAGMENTED_PAYLOAD_MAX_LEN
} else {
0
};
let pre_casted_frags =
1 + (tail_len as f64 / UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN as f64).ceil() as usize;
if pre_casted_frags > u8::max_value() as usize {
panic!("message would produce too many fragments!")
};
pre_casted_frags
};
// determine bounds for the first fragment which depends on whether set is pre-linked
let mut lb = 0;
let mut ub = if previous_link_id.is_some() {
usize::min(message.len(), LINKED_FRAGMENTED_PAYLOAD_MAX_LEN)
} else {
// the set might be linked, but fragment itself is not (i.e. the set is linked at the tail)
UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN
};
let mut fragments = Vec::with_capacity(num_frags_usize);
for i in 1..(num_frags_usize + 1) {
// we can't use u8 directly here as upper (NON-INCLUSIVE, so i would always fit) bound could be u8::max_value() + 1
let fragment = Fragment::try_new_fragmented(
&message[lb..ub],
id,
num_frags_usize as u8,
i as u8,
if i == 1 { previous_link_id } else { None },
if i == num_frags_usize {
next_link_id
} else {
None
},
)
.unwrap();
fragments.push(fragment);
// update bounds for the next fragment
lb = ub;
ub = usize::min(message.len(), ub + UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN);
}
fragments
}
/// Based on total message length, determines the number of sets into which it is going to be split.
fn total_number_of_sets(message_len: usize) -> usize {
if message_len <= MAX_UNLINKED_SET_PAYLOAD_LENGTH {
1
} else if message_len > MAX_UNLINKED_SET_PAYLOAD_LENGTH
&& message_len <= 2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
{
2
} else {
let len_without_edges = message_len - 2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH;
// every set in between edges must be two way linked
(len_without_edges as f64 / TWO_WAY_LINKED_SET_PAYLOAD_LENGTH as f64).ceil() as usize + 2
}
}
/// Given part of the underlying message as well id of the set as well as its potential linked sets,
/// correctly delegates to appropriate set constructor.
fn prepare_fragment_set(
message: &[u8],
id: i32,
previous_link_id: Option<i32>,
next_link_id: Option<i32>,
) -> FragmentSet {
if previous_link_id.is_some() || next_link_id.is_some() {
prepare_linked_fragment_set(message, id, previous_link_id, next_link_id)
} else if message.len() > UNFRAGMENTED_PAYLOAD_MAX_LEN {
// the bounds on whether the message fits in an unlinked set should have been done by the callee
// when determining ids of other sets
prepare_unlinked_fragmented_set(message, id)
} else {
prepare_unfragmented_set(message)
}
}
/// Entry point for splitting whole message into possibly multiple `Set`s.
pub(crate) fn split_into_sets(message: &[u8]) -> Vec<FragmentSet> {
use rand::thread_rng;
let mut rng = thread_rng();
let num_of_sets = total_number_of_sets(message.len());
if num_of_sets == 1 {
let set_id = generate_set_id(&mut rng);
vec![prepare_fragment_set(message, set_id, None, None)]
} else {
let mut sets = Vec::with_capacity(num_of_sets);
// pre-generate all ids for the sets
let set_ids: Vec<_> = std::iter::repeat(())
.map(|_| generate_set_id(&mut rng))
.take(num_of_sets)
.collect();
// initial bounds for the set payloads
let mut lb = 0;
let mut ub = MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH;
for i in 0..num_of_sets {
let fragment_set = prepare_fragment_set(
&message[lb..ub],
set_ids[i],
if i == 0 { None } else { Some(set_ids[i - 1]) },
if i == (num_of_sets - 1) {
None
} else {
Some(set_ids[i + 1])
},
);
sets.push(fragment_set);
// update bounds for the next set
lb = ub;
ub = if i == num_of_sets - 2 {
// we're going to go into the last iteration now, hence the last set will be one-way linked
usize::min(message.len(), ub + MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH)
} else {
usize::min(message.len(), ub + TWO_WAY_LINKED_SET_PAYLOAD_LENGTH)
}
}
sets
}
}
// reason for top level tests module is to be able to use the helper functions to verify sets payloads
#[cfg(test)]
mod tests {
use super::*;
fn verify_unlinked_set_payload(mut set: FragmentSet, payload: &[u8]) {
for i in (0..set.len()).rev() {
assert_eq!(
set.pop().unwrap().extract_payload(),
payload[i * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN
..usize::min(payload.len(), (i + 1) * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN)]
.to_vec()
)
}
}
fn verify_pre_linked_set_payload(mut set: FragmentSet, payload: &[u8]) {
for i in (0..set.len()).rev() {
let lb = if i == 0 {
0
} else {
(i - 1) * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN + LINKED_FRAGMENTED_PAYLOAD_MAX_LEN
};
let ub = usize::min(
payload.len(),
i * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN + LINKED_FRAGMENTED_PAYLOAD_MAX_LEN,
);
assert_eq!(
set.pop().unwrap().extract_payload(),
payload[lb..ub].to_vec()
)
}
}
fn verify_post_linked_set_payload(mut set: FragmentSet, payload: &[u8]) {
for i in (0..set.len()).rev() {
let lb = i * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN;
let ub = if i == (u8::max_value() as usize - 1) {
i * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN + LINKED_FRAGMENTED_PAYLOAD_MAX_LEN
} else {
(i + 1) * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN
};
assert_eq!(
set.pop().unwrap().extract_payload(),
payload[lb..ub].to_vec(),
)
}
}
fn verify_two_way_linked_set_payload(mut set: FragmentSet, payload: &[u8]) {
for i in (0..set.len()).rev() {
let lb = if i == 0 {
0
} else {
(i - 1) * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN + LINKED_FRAGMENTED_PAYLOAD_MAX_LEN
};
let ub = if i == (u8::max_value() as usize - 1) {
(i - 1) * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN
+ 2 * LINKED_FRAGMENTED_PAYLOAD_MAX_LEN
} else {
i * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN + LINKED_FRAGMENTED_PAYLOAD_MAX_LEN
};
assert_eq!(
set.pop().unwrap().extract_payload(),
payload[lb..ub].to_vec(),
)
}
}
fn verfiy_correct_link(left: &FragmentSet, right: &FragmentSet) {
let first_id = left[0].id();
let post_id = left[254].next_fragments_set_id().unwrap();
let second_id = right[0].id();
let pre_id = right[0].previous_fragments_set_id().unwrap();
assert_eq!(first_id, pre_id);
assert_eq!(second_id, post_id);
}
#[cfg(test)]
mod preparing_unfragmented_set {
use super::*;
#[test]
fn makes_set_with_single_unfragmented_element_for_valid_message_lengths() {
let mut set = prepare_unfragmented_set(&[1]);
assert_eq!(1, set.len());
assert_eq!(set.pop().unwrap().extract_payload(), [1].to_vec());
let mut set = prepare_unfragmented_set(&[1u8; UNFRAGMENTED_PAYLOAD_MAX_LEN]);
assert_eq!(1, set.len());
assert_eq!(
set.pop().unwrap().extract_payload(),
[1u8; UNFRAGMENTED_PAYLOAD_MAX_LEN].to_vec()
);
}
#[test]
#[should_panic]
fn panics_for_too_long_payload() {
prepare_unfragmented_set(&[1u8; UNFRAGMENTED_PAYLOAD_MAX_LEN + 1]);
}
}
#[cfg(test)]
mod preparing_unlinked_set {
// remember this this is only called for a sole set with <= 255 fragments
use super::*;
use rand::{thread_rng, RngCore};
#[test]
fn makes_set_with_correctly_split_payload() {
let id = 12345;
let mut rng = thread_rng();
let mut two_element_set_payload = [0u8; UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN + 1];
rng.fill_bytes(&mut two_element_set_payload);
let two_element_set = prepare_unlinked_fragmented_set(&two_element_set_payload, id);
assert_eq!(2, two_element_set.len());
verify_unlinked_set_payload(two_element_set, &two_element_set_payload);
let mut forty_two_element_set_payload =
[0u8; 41 * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN + 42];
rng.fill_bytes(&mut forty_two_element_set_payload);
let forty_two_element_set =
prepare_unlinked_fragmented_set(&forty_two_element_set_payload, id);
assert_eq!(42, forty_two_element_set.len());
verify_unlinked_set_payload(forty_two_element_set, &forty_two_element_set_payload);
let mut max_fragments_set_payload =
[0u8; MAX_UNLINKED_SET_PAYLOAD_LENGTH - UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN + 1]; // last fragment should have a single byte of data
rng.fill_bytes(&mut max_fragments_set_payload);
let max_fragment_set = prepare_unlinked_fragmented_set(&max_fragments_set_payload, id);
assert_eq!(u8::max_value() as usize, max_fragment_set.len());
verify_unlinked_set_payload(max_fragment_set, &max_fragments_set_payload);
let mut full_set_payload = [0u8; MAX_UNLINKED_SET_PAYLOAD_LENGTH];
rng.fill_bytes(&mut full_set_payload);
let full_fragment_set = prepare_unlinked_fragmented_set(&full_set_payload, id);
assert_eq!(u8::max_value() as usize, full_fragment_set.len());
verify_unlinked_set_payload(full_fragment_set, &full_set_payload);
}
#[test]
#[should_panic]
fn panics_for_too_long_payload() {
prepare_unlinked_fragmented_set(&[0u8; MAX_UNLINKED_SET_PAYLOAD_LENGTH + 1], 12345);
}
}
#[cfg(test)]
mod preparing_linked_set {
use super::*;
use rand::{thread_rng, RngCore};
#[test]
fn makes_set_with_correctly_split_payload_for_pre_linked_set() {
let id = 12345;
let link_id = 1234;
let mut rng = thread_rng();
let mut two_element_set_payload = [0u8; LINKED_FRAGMENTED_PAYLOAD_MAX_LEN + 1];
rng.fill_bytes(&mut two_element_set_payload);
let two_element_set =
prepare_linked_fragment_set(&two_element_set_payload, id, Some(link_id), None);
assert_eq!(2, two_element_set.len());
verify_pre_linked_set_payload(two_element_set, &two_element_set_payload);
let mut forty_two_element_set_payload = [0u8; LINKED_FRAGMENTED_PAYLOAD_MAX_LEN
+ 40 * UNLINKED_FRAGMENTED_PAYLOAD_MAX_LEN
+ 42];
rng.fill_bytes(&mut forty_two_element_set_payload);
let forty_two_element_set = prepare_linked_fragment_set(
&forty_two_element_set_payload,
id,
Some(link_id),
None,
);
assert_eq!(42, forty_two_element_set.len());
verify_pre_linked_set_payload(forty_two_element_set, &forty_two_element_set_payload);
let mut max_fragments_set_payload =
[0u8; MAX_UNLINKED_SET_PAYLOAD_LENGTH - LINKED_FRAGMENTED_PAYLOAD_MAX_LEN + 1]; // last fragment should have a single byte of data
rng.fill_bytes(&mut max_fragments_set_payload);
let max_fragment_set =
prepare_linked_fragment_set(&max_fragments_set_payload, id, Some(link_id), None);
assert_eq!(u8::max_value() as usize, max_fragment_set.len());
verify_pre_linked_set_payload(max_fragment_set, &max_fragments_set_payload);
let mut full_set_payload = [0u8; MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH];
rng.fill_bytes(&mut full_set_payload);
let full_fragment_set =
prepare_linked_fragment_set(&full_set_payload, id, Some(link_id), None);
assert_eq!(u8::max_value() as usize, full_fragment_set.len());
verify_pre_linked_set_payload(full_fragment_set, &full_set_payload);
}
#[test]
#[should_panic]
fn panics_for_too_long_payload_for_pre_linked_set() {
prepare_linked_fragment_set(
&[0u8; MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH + 1],
12345,
Some(1234),
None,
);
}
#[test]
fn makes_set_with_correctly_split_payload_for_post_linked_set() {
let id = 12345;
let link_id = 1234;
let mut rng = thread_rng();
// if set is post-linked, there is only a single valid case - full length payload
let mut full_set_payload = [0u8; MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH];
rng.fill_bytes(&mut full_set_payload);
let full_fragment_set =
prepare_linked_fragment_set(&full_set_payload, id, None, Some(link_id));
assert_eq!(u8::max_value() as usize, full_fragment_set.len());
verify_post_linked_set_payload(full_fragment_set, &full_set_payload);
}
#[test]
#[should_panic]
fn panics_for_too_long_payload_for_post_linked_set() {
prepare_linked_fragment_set(
&[0u8; MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH + 1],
12345,
None,
Some(1234),
);
}
#[test]
#[should_panic]
fn panics_for_too_short_payload_for_post_linked_set() {
prepare_linked_fragment_set(
&[0u8; MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH - 1],
12345,
None,
Some(1234),
);
}
#[test]
fn makes_set_with_correctly_split_payload_for_two_way_linked_set() {
// again, relatively simple case -
// if set is two-way-linked, there is only a single valid case - full length payload
let id = 12345;
let pre_link_id = 1234;
let post_link_id = 123456;
let mut rng = thread_rng();
let mut full_set_payload = [0u8; TWO_WAY_LINKED_SET_PAYLOAD_LENGTH];
rng.fill_bytes(&mut full_set_payload);
let full_fragment_set = prepare_linked_fragment_set(
&full_set_payload,
id,
Some(pre_link_id),
Some(post_link_id),
);
assert_eq!(u8::max_value() as usize, full_fragment_set.len());
verify_two_way_linked_set_payload(full_fragment_set, &full_set_payload);
}
#[test]
#[should_panic]
fn panics_for_too_long_payload_for_two_way_linked_set() {
prepare_linked_fragment_set(
&[0u8; TWO_WAY_LINKED_SET_PAYLOAD_LENGTH + 1],
12345,
Some(123456),
Some(1234),
);
}
#[test]
#[should_panic]
fn panics_for_too_short_payload_for_two_way_linked_set() {
prepare_linked_fragment_set(
&[0u8; TWO_WAY_LINKED_SET_PAYLOAD_LENGTH - 1],
12345,
Some(123456),
Some(1234),
);
}
}
#[cfg(test)]
mod splitting_into_sets {
use super::*;
use rand::{thread_rng, RngCore};
#[test]
fn correctly_creates_set_with_single_unfragmented_element_when_expected() {
let mut rng = thread_rng();
let tiny_message = [1, 2, 3, 4, 5];
let mut max_unfragmented_message = [0u8; UNFRAGMENTED_PAYLOAD_MAX_LEN];
rng.fill_bytes(&mut max_unfragmented_message);
let mut sets = split_into_sets(&tiny_message);
assert_eq!(1, sets.len());
assert_eq!(1, sets[0].len());
assert_eq!(
tiny_message.to_vec(),
sets.pop().unwrap().pop().unwrap().extract_payload()
);
let mut sets = split_into_sets(&max_unfragmented_message);
assert_eq!(1, sets.len());
assert_eq!(1, sets[0].len());
assert_eq!(
max_unfragmented_message.to_vec(),
sets.pop().unwrap().pop().unwrap().extract_payload()
);
}
#[test]
fn correctly_creates_single_fragmented_set_when_expected() {
let mut rng = thread_rng();
let mut message = [0u8; MAX_UNLINKED_SET_PAYLOAD_LENGTH - 2345];
rng.fill_bytes(&mut message);
let mut sets = split_into_sets(&message);
assert_eq!(1, sets.len());
verify_unlinked_set_payload(sets.pop().unwrap(), &message);
}
// a very specific test case that would have saved a lot of headache if was introduced
// earlier...
#[test]
fn correctly_creates_two_singly_linked_sets_with_second_set_containing_data_fitting_in_unfragmented_payload(
) {
let mut rng = thread_rng();
let mut message = [0u8; MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH + 123];
rng.fill_bytes(&mut message);
let mut sets = split_into_sets(&message);
assert_eq!(2, sets.len());
verfiy_correct_link(&sets[0], &sets[1]);
verify_pre_linked_set_payload(
sets.pop().unwrap(),
&message[MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH..],
);
verify_post_linked_set_payload(
sets.pop().unwrap(),
&message[..MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
}
#[test]
fn correctly_creates_two_singly_linked_sets_when_expected() {
let mut rng = thread_rng();
let mut message = [0u8; MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH + 2345];
rng.fill_bytes(&mut message);
let mut sets = split_into_sets(&message);
assert_eq!(2, sets.len());
verfiy_correct_link(&sets[0], &sets[1]);
verify_pre_linked_set_payload(
sets.pop().unwrap(),
&message[MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH..],
);
verify_post_linked_set_payload(
sets.pop().unwrap(),
&message[..MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
let mut message = [0u8; 2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH];
rng.fill_bytes(&mut message);
let mut sets = split_into_sets(&message);
assert_eq!(2, sets.len());
assert_eq!(sets[0].len(), u8::max_value() as usize);
assert_eq!(sets[1].len(), u8::max_value() as usize);
verfiy_correct_link(&sets[0], &sets[1]);
verify_pre_linked_set_payload(
sets.pop().unwrap(),
&message[MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH..],
);
verify_post_linked_set_payload(
sets.pop().unwrap(),
&message[..MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
}
#[test]
fn correctly_creates_four_correctly_formed_sets_when_expected() {
let mut rng = thread_rng();
let mut message = [0u8; 2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
+ MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
+ 2345];
rng.fill_bytes(&mut message);
let mut sets = split_into_sets(&message);
assert_eq!(4, sets.len());
assert_eq!(sets[0].len(), u8::max_value() as usize);
assert_eq!(sets[1].len(), u8::max_value() as usize);
assert_eq!(sets[2].len(), u8::max_value() as usize);
verfiy_correct_link(&sets[0], &sets[1]);
verfiy_correct_link(&sets[1], &sets[2]);
verfiy_correct_link(&sets[2], &sets[3]);
verify_pre_linked_set_payload(
sets.pop().unwrap(),
&message[2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
+ MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH..],
);
verify_two_way_linked_set_payload(
sets.pop().unwrap(),
&message[TWO_WAY_LINKED_SET_PAYLOAD_LENGTH + MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
..2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
+ MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
verify_two_way_linked_set_payload(
sets.pop().unwrap(),
&message[MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
..TWO_WAY_LINKED_SET_PAYLOAD_LENGTH + MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
verify_post_linked_set_payload(
sets.pop().unwrap(),
&message[..MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
let mut message = [0u8; 2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
+ 2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH];
rng.fill_bytes(&mut message);
let mut sets = split_into_sets(&message);
assert_eq!(4, sets.len());
assert_eq!(sets[0].len(), u8::max_value() as usize);
assert_eq!(sets[1].len(), u8::max_value() as usize);
assert_eq!(sets[2].len(), u8::max_value() as usize);
assert_eq!(sets[3].len(), u8::max_value() as usize);
verfiy_correct_link(&sets[0], &sets[1]);
verfiy_correct_link(&sets[1], &sets[2]);
verfiy_correct_link(&sets[2], &sets[3]);
verify_pre_linked_set_payload(
sets.pop().unwrap(),
&message[2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
+ MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH..],
);
verify_two_way_linked_set_payload(
sets.pop().unwrap(),
&message[TWO_WAY_LINKED_SET_PAYLOAD_LENGTH + MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
..2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
+ MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
verify_two_way_linked_set_payload(
sets.pop().unwrap(),
&message[MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
..TWO_WAY_LINKED_SET_PAYLOAD_LENGTH + MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
verify_post_linked_set_payload(
sets.pop().unwrap(),
&message[..MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH],
);
}
}
#[cfg(test)]
mod helpers {
use super::*;
#[test]
fn total_number_of_sets() {
assert_eq!(
1,
super::total_number_of_sets(MAX_UNLINKED_SET_PAYLOAD_LENGTH - 1)
);
assert_eq!(
1,
super::total_number_of_sets(MAX_UNLINKED_SET_PAYLOAD_LENGTH)
);
assert_eq!(
2,
super::total_number_of_sets(MAX_UNLINKED_SET_PAYLOAD_LENGTH + 1)
);
assert_eq!(
2,
super::total_number_of_sets(2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH)
);
assert_eq!(
3,
super::total_number_of_sets(2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH + 1)
);
assert_eq!(
3,
super::total_number_of_sets(
2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH + TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
- 1
)
);
assert_eq!(
3,
super::total_number_of_sets(
2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH + TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
)
);
assert_eq!(
4,
super::total_number_of_sets(
2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
+ TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
+ 1
)
);
assert_eq!(
4,
super::total_number_of_sets(
2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
+ 2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
- 1
)
);
assert_eq!(
4,
super::total_number_of_sets(
2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
+ 2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
)
);
assert_eq!(
5,
super::total_number_of_sets(
2 * MAX_ONE_WAY_LINKED_SET_PAYLOAD_LENGTH
+ 2 * TWO_WAY_LINKED_SET_PAYLOAD_LENGTH
+ 1
)
);
}
}
}
+9
View File
@@ -0,0 +1,9 @@
pub mod addressing;
pub mod chunking;
// Future consideration: currently in a lot of places, the payloads have randomised content
// which is not a perfect testing strategy as it might not detect some edge cases I never would
// have assumed could be possible. A better approach would be to research some Fuzz testing
// library like: https://github.com/rust-fuzz/afl.rs and use that instead for the inputs.
// perhaps it might be useful down the line for interaction testing between client,mixes,etc?
+2 -2
View File
@@ -15,9 +15,9 @@ rand = "0.7.2"
serde = { version = "1.0.104", features = ["derive"] }
## internal
addressing = {path = "../addressing"}
nymsphinx = {path = "../nymsphinx"}
version-checker = {path = "../version-checker" }
## will be moved to proper dependencies once released
sphinx = { git = "https://github.com/nymtech/sphinx", rev="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
# sphinx = { path = "../../../sphinx"}
+3 -1
View File
@@ -114,7 +114,9 @@ pub trait NymTopology: Sized + std::fmt::Debug + Send + Sync + Clone {
}
fn can_construct_path_through(&self) -> bool {
self.make_layered_topology().is_ok()
!self.mix_nodes().is_empty()
&& !self.providers().is_empty()
&& self.make_layered_topology().is_ok()
}
}
+4 -3
View File
@@ -1,6 +1,7 @@
use crate::filter;
use nymsphinx::addressing::nodes::NymNodeRoutingAddress;
use sphinx::route::Node as SphinxNode;
use sphinx::route::NodeAddressBytes;
use std::convert::TryInto;
use std::net::SocketAddr;
#[derive(Debug, Clone)]
@@ -29,10 +30,10 @@ impl filter::Versioned for Node {
impl Into<SphinxNode> for Node {
fn into(self) -> SphinxNode {
let address_bytes = addressing::encoded_bytes_from_socket_address(self.host);
let node_address_bytes = NymNodeRoutingAddress::from(self.host).try_into().unwrap();
let key_bytes = self.get_pub_key_bytes();
let key = sphinx::key::new(key_bytes);
SphinxNode::new(NodeAddressBytes::from_bytes(address_bytes), key)
SphinxNode::new(node_address_bytes, key)
}
}
+6 -3
View File
@@ -1,6 +1,7 @@
use crate::filter;
use nymsphinx::addressing::nodes::NymNodeRoutingAddress;
use sphinx::route::Node as SphinxNode;
use sphinx::route::NodeAddressBytes;
use std::convert::TryInto;
use std::net::SocketAddr;
#[derive(Debug, Clone)]
@@ -35,10 +36,12 @@ impl filter::Versioned for Node {
impl Into<SphinxNode> for Node {
fn into(self) -> SphinxNode {
let address_bytes = addressing::encoded_bytes_from_socket_address(self.mixnet_listener);
let node_address_bytes = NymNodeRoutingAddress::from(self.mixnet_listener)
.try_into()
.unwrap();
let key_bytes = self.get_pub_key_bytes();
let key = sphinx::key::new(key_bytes);
SphinxNode::new(NodeAddressBytes::from_bytes(address_bytes), key)
SphinxNode::new(node_address_bytes, key)
}
}
+4 -3
View File
@@ -1,7 +1,7 @@
[package]
build = "build.rs"
name = "nym-mixnode"
version = "0.5.0"
version = "0.6.0"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
edition = "2018"
@@ -20,15 +20,16 @@ serde = { version = "1.0.104", features = ["derive"] }
tokio = { version = "0.2", features = ["full"] }
## internal
addressing = {path = "../common/addressing" }
config = {path = "../common/config"}
crypto = {path = "../common/crypto"}
directory-client = { path = "../common/clients/directory-client" }
multi-tcp-client = { path = "../common/clients/multi-tcp-client" }
nymsphinx = {path = "../common/nymsphinx" }
pemstore = {path = "../common/pemstore"}
topology = {path = "../common/topology"}
## will be moved to proper dependencies once released
sphinx = { git = "https://github.com/nymtech/sphinx", rev="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
[build-dependencies]
built = "0.3.2"
+21 -1
View File
@@ -17,8 +17,10 @@ const DEFAULT_LISTENING_PORT: u16 = 1789;
// where applicable, the below are defined in milliseconds
const DEFAULT_PRESENCE_SENDING_DELAY: u64 = 1500;
const DEFAULT_METRICS_SENDING_DELAY: u64 = 1000;
const DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY: u64 = 60_000; // 1min
const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: u64 = 10_000; // 10s
const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: u64 = 300_000; // 5min
const DEFAULT_INITIAL_CONNECTION_TIMEOUT: u64 = 1_500; // 1.5s
#[derive(Debug, Default, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
@@ -213,6 +215,10 @@ impl Config {
time::Duration::from_millis(self.debug.metrics_sending_delay)
}
pub fn get_metrics_running_stats_logging_delay(&self) -> time::Duration {
time::Duration::from_millis(self.debug.metrics_running_stats_logging_delay)
}
pub fn get_layer(&self) -> u64 {
self.mixnode.layer
}
@@ -232,6 +238,10 @@ impl Config {
pub fn get_packet_forwarding_maximum_backoff(&self) -> time::Duration {
time::Duration::from_millis(self.debug.packet_forwarding_maximum_backoff)
}
pub fn get_initial_connection_timeout(&self) -> time::Duration {
time::Duration::from_millis(self.debug.initial_connection_timeout)
}
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
@@ -313,7 +323,7 @@ impl Default for Logging {
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
#[serde(default, deny_unknown_fields)]
pub struct Debug {
// The idea of additional 'directory servers' is to let mixes report their presence
// and metrics to separate places
@@ -331,6 +341,10 @@ pub struct Debug {
/// The provided value is interpreted as milliseconds.
metrics_sending_delay: u64,
/// Delay between each subsequent running metrics statistics being logged.
/// The provided value is interpreted as milliseconds.
metrics_running_stats_logging_delay: u64,
/// Initial value of an exponential backoff to reconnect to dropped TCP connection when
/// forwarding sphinx packets.
/// The provided value is interpreted as milliseconds.
@@ -340,6 +354,10 @@ pub struct Debug {
/// forwarding sphinx packets.
/// The provided value is interpreted as milliseconds.
packet_forwarding_maximum_backoff: u64,
/// Timeout for establishing initial connection when trying to forward a sphinx packet.
/// The provider value is interpreted as milliseconds.
initial_connection_timeout: u64,
}
impl Debug {
@@ -360,8 +378,10 @@ impl Default for Debug {
presence_sending_delay: DEFAULT_PRESENCE_SENDING_DELAY,
metrics_directory_server: Self::default_directory_server(),
metrics_sending_delay: DEFAULT_METRICS_SENDING_DELAY,
metrics_running_stats_logging_delay: DEFAULT_METRICS_RUNNING_STATS_LOGGING_DELAY,
packet_forwarding_initial_backoff: DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF,
packet_forwarding_maximum_backoff: DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF,
initial_connection_timeout: DEFAULT_INITIAL_CONNECTION_TIMEOUT,
}
}
}
-30
View File
@@ -53,35 +53,5 @@ nym_root_directory = '{{ mixnode.nym_root_directory }}'
# TODO
##### debug configuration options #####
# The following options should not be modified unless you know EXACTLY what you are doing
# as if set incorrectly, they may impact your anonymity.
[debug]
# Directory server to which the server will be reporting their presence data.
presence_directory_server = '{{ debug.presence_directory_server}}'
# Delay between each subsequent presence data being sent.
# The provided value is interpreted as milliseconds.
presence_sending_delay = {{ debug.presence_sending_delay }}
# Directory server to which the server will be reporting their metrics data.
metrics_directory_server = '{{ debug.metrics_directory_server }}'
# Delay between each subsequent metrics data being sent.
# The provided value is interpreted as milliseconds.
metrics_sending_delay = {{ debug.metrics_sending_delay }}
# Initial value of an exponential backoff to reconnect to dropped TCP connection when
# forwarding sphinx packets.
# The provided value is interpreted as milliseconds.
packet_forwarding_initial_backoff = {{ debug.packet_forwarding_initial_backoff }}
# Maximum value of an exponential backoff to reconnect to dropped TCP connection when
# forwarding sphinx packets.
# The provided value is interpreted as milliseconds.
packet_forwarding_maximum_backoff = {{ debug.packet_forwarding_maximum_backoff }}
"#
}
+12 -4
View File
@@ -48,6 +48,8 @@ async fn process_socket_connection(
return;
}
Ok(n) => {
// If I understand it correctly, this if should never be executed as if `read_exact`
// does not fill buffer, it will throw UnexpectedEof?
if n != sphinx::PACKET_SIZE {
warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, sphinx::PACKET_SIZE);
continue;
@@ -63,10 +65,16 @@ async fn process_socket_connection(
))
}
Err(e) => {
warn!(
"failed to read from socket. Closing the connection; err = {:?}",
e
);
if e.kind() == io::ErrorKind::UnexpectedEof {
debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\
Also closing the connection on this end.", socket.peer_addr())
} else {
warn!(
"failed to read from socket (source: {:?}). Closing the connection; err = {:?}",
socket.peer_addr(),
e
);
}
return;
}
};
+80 -4
View File
@@ -4,13 +4,15 @@ use directory_client::DirectoryClient;
use futures::channel::mpsc;
use futures::lock::Mutex;
use futures::StreamExt;
use log::{debug, error};
use log::*;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use std::time::{Duration, SystemTime};
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
type SentMetricsMap = HashMap<String, u64>;
pub(crate) enum MetricEvent {
Sent(String),
Received,
@@ -25,7 +27,7 @@ struct MixMetrics {
struct MixMetricsInner {
received: u64,
sent: HashMap<String, u64>,
sent: SentMetricsMap,
}
impl MixMetrics {
@@ -49,7 +51,7 @@ impl MixMetrics {
*receiver_count += 1;
}
async fn acquire_and_reset_metrics(&mut self) -> (u64, HashMap<String, u64>) {
async fn acquire_and_reset_metrics(&mut self) -> (u64, SentMetricsMap) {
let mut unlocked = self.inner.lock().await;
let received = unlocked.received;
@@ -92,6 +94,7 @@ struct MetricsSender {
directory_client: directory_client::Client,
pub_key_str: String,
sending_delay: Duration,
metrics_informer: MetricsInformer,
}
impl MetricsSender {
@@ -100,6 +103,7 @@ impl MetricsSender {
directory_server: String,
pub_key_str: String,
sending_delay: Duration,
running_logging_delay: Duration,
) -> Self {
MetricsSender {
metrics,
@@ -108,6 +112,7 @@ impl MetricsSender {
)),
pub_key_str,
sending_delay,
metrics_informer: MetricsInformer::new(running_logging_delay),
}
}
@@ -118,6 +123,10 @@ impl MetricsSender {
let sending_delay = tokio::time::delay_for(self.sending_delay);
let (received, sent) = self.metrics.acquire_and_reset_metrics().await;
self.metrics_informer.update_running_stats(received, &sent);
self.metrics_informer.log_report_stats(received, &sent);
self.metrics_informer.try_log_running_stats();
match self.directory_client.metrics_post.post(&MixMetric {
pub_key: self.pub_key_str.clone(),
received,
@@ -134,6 +143,71 @@ impl MetricsSender {
}
}
struct MetricsInformer {
total_received: u64,
sent_map: SentMetricsMap,
running_stats_logging_delay: Duration,
last_reported_stats: SystemTime,
}
impl MetricsInformer {
fn new(running_stats_logging_delay: Duration) -> Self {
MetricsInformer {
total_received: 0,
sent_map: HashMap::new(),
running_stats_logging_delay,
last_reported_stats: SystemTime::now(),
}
}
fn should_log_running_stats(&self) -> bool {
self.last_reported_stats + self.running_stats_logging_delay < SystemTime::now()
}
fn try_log_running_stats(&mut self) {
if self.should_log_running_stats() {
self.log_running_stats()
}
}
fn update_running_stats(&mut self, pre_reset_received: u64, pre_reset_sent: &SentMetricsMap) {
self.total_received += pre_reset_received;
for (mix, count) in pre_reset_sent.iter() {
*self.sent_map.entry(mix.clone()).or_insert(0) += *count;
}
}
fn log_report_stats(&self, pre_reset_received: u64, pre_reset_sent: &SentMetricsMap) {
debug!(
"Since last metrics report mixed {} packets!",
pre_reset_received
);
debug!(
"Since last metrics report received {} packets",
pre_reset_sent.values().sum::<u64>()
);
trace!(
"Since last metrics report sent packets to the following: \n{:#?}",
pre_reset_sent
);
}
fn log_running_stats(&mut self) {
info!(
"Since startup mixed {} packets!",
self.sent_map.values().sum::<u64>()
);
debug!("Since startup received {} packets", self.total_received);
trace!(
"Since startup sent packets to the following: \n{:#?}",
self.sent_map
);
self.last_reported_stats = SystemTime::now();
}
}
#[derive(Clone)]
pub struct MetricsReporter {
metrics_tx: mpsc::UnboundedSender<MetricEvent>,
@@ -173,6 +247,7 @@ impl MetricsController {
directory_server: String,
pub_key_str: String,
sending_delay: Duration,
running_stats_logging_delay: Duration,
) -> Self {
let (metrics_tx, metrics_rx) = mpsc::unbounded();
let shared_metrics = MixMetrics::new();
@@ -183,6 +258,7 @@ impl MetricsController {
directory_server,
pub_key_str,
sending_delay,
running_stats_logging_delay,
),
receiver: MetricsReceiver::new(shared_metrics, metrics_rx),
reporter: MetricsReporter::new(metrics_tx),
+25
View File
@@ -2,11 +2,13 @@ use crate::config::persistence::pathfinder::MixNodePathfinder;
use crate::config::Config;
use crate::node::packet_processing::PacketProcessor;
use crypto::encryption;
use directory_client::presence::Topology;
use futures::channel::mpsc;
use log::*;
use pemstore::pemstore::PemStore;
use std::net::SocketAddr;
use tokio::runtime::Runtime;
use topology::NymTopology;
mod listener;
mod metrics;
@@ -62,6 +64,7 @@ impl MixNode {
self.config.get_metrics_directory_server(),
self.sphinx_keypair.public_key().to_base58_string(),
self.config.get_metrics_sending_delay(),
self.config.get_metrics_running_stats_logging_delay(),
)
.start(self.runtime.handle())
}
@@ -92,17 +95,39 @@ impl MixNode {
packet_forwarding::PacketForwarder::new(
self.config.get_packet_forwarding_initial_backoff(),
self.config.get_packet_forwarding_maximum_backoff(),
self.config.get_initial_connection_timeout(),
)
})
.start(self.runtime.handle())
}
fn check_if_same_ip_node_exists(&self) -> Option<String> {
// TODO: once we change to graph topology this here will need to be updated!
let topology = Topology::new(self.config.get_presence_directory_server());
let existing_mixes_presence = topology.mix_nodes;
existing_mixes_presence
.iter()
.find(|node| node.host == self.config.get_announce_address())
.map(|node| node.pub_key.clone())
}
pub fn run(&mut self) {
info!("Starting nym mixnode");
if let Some(duplicate_node_key) = self.check_if_same_ip_node_exists() {
error!(
"Our announce-host is identical to one of existing nodes! (its key is {:?}",
duplicate_node_key
);
return;
}
let forwarding_channel = self.start_packet_forwarder();
let metrics_reporter = self.start_metrics_reporter();
self.start_socket_listener(metrics_reporter, forwarding_channel);
self.start_presence_notifier();
info!("Finished nym mixnode startup procedure - it should now be able to receive mix traffic!");
if let Err(e) = self.runtime.block_on(tokio::signal::ctrl_c()) {
error!(
"There was an error while capturing SIGINT - {:?}. We will terminate regardless",
+2
View File
@@ -14,10 +14,12 @@ impl PacketForwarder {
pub(crate) fn new(
initial_reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
initial_connection_timeout: Duration,
) -> PacketForwarder {
let tcp_client_config = multi_tcp_client::Config::new(
initial_reconnection_backoff,
maximum_reconnection_backoff,
initial_connection_timeout,
);
let (conn_tx, conn_rx) = mpsc::unbounded();
+5 -5
View File
@@ -1,10 +1,11 @@
use crate::node::metrics;
use addressing::AddressTypeError;
use crypto::encryption;
use log::*;
use nymsphinx::addressing::nodes::{NymNodeRoutingAddress, NymNodeRoutingAddressError};
use sphinx::header::delays::Delay as SphinxDelay;
use sphinx::route::NodeAddressBytes;
use sphinx::{ProcessedPacket, SphinxPacket};
use std::convert::TryFrom;
use std::net::SocketAddr;
use std::ops::Deref;
use std::sync::Arc;
@@ -32,8 +33,8 @@ impl From<sphinx::ProcessingError> for MixProcessingError {
}
}
impl From<addressing::AddressTypeError> for MixProcessingError {
fn from(_: AddressTypeError) -> Self {
impl From<NymNodeRoutingAddressError> for MixProcessingError {
fn from(_: NymNodeRoutingAddressError) -> Self {
use MixProcessingError::*;
InvalidHopAddress
@@ -68,8 +69,7 @@ impl PacketProcessor {
forward_address: NodeAddressBytes,
delay: SphinxDelay,
) -> Result<MixProcessingResult, MixProcessingError> {
let next_hop_address =
addressing::socket_address_from_encoded_bytes(forward_address.to_bytes())?;
let next_hop_address: SocketAddr = NymNodeRoutingAddress::try_from(forward_address)?.into();
// Delay packet for as long as required
tokio::time::delay_for(delay.to_duration()).await;
+3 -3
View File
@@ -1,7 +1,7 @@
[package]
build = "build.rs"
name = "nym-client"
version = "0.5.0"
version = "0.6.0"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
edition = "2018"
@@ -28,20 +28,20 @@ tokio = { version = "0.2", features = ["full"] }
tokio-tungstenite = "0.10.1"
## internal
addressing = {path = "../common/addressing" }
config = {path = "../common/config"}
crypto = {path = "../common/crypto"}
directory-client = { path = "../common/clients/directory-client" }
healthcheck = { path = "../common/healthcheck" }
mix-client = { path = "../common/clients/mix-client" }
multi-tcp-client = { path = "../common/clients/multi-tcp-client" }
nymsphinx = { path = "../common/nymsphinx" }
pemstore = {path = "../common/pemstore"}
provider-client = { path = "../common/clients/provider-client" }
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="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
# sphinx = { path = "../../sphinx"}
[build-dependencies]
@@ -99,6 +99,9 @@ impl<T: 'static + NymTopology> LoopCoverTrafficStream<T> {
self.mix_tx
.unbounded_send(MixMessage::new(cover_message.0, cover_message.1))
.unwrap();
// JS: due to identical logical structure to OutQueueControl::on_message(), this is also
// presumably required to prevent bugs in the future. Exact reason is still unknown to me.
tokio::task::yield_now().await;
}
async fn run(&mut self) {
+2
View File
@@ -26,11 +26,13 @@ impl MixTrafficController {
pub(crate) fn new(
initial_reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
initial_connection_timeout: Duration,
mix_rx: MixMessageReceiver,
) -> Self {
let tcp_client_config = multi_tcp_client::Config::new(
initial_reconnection_backoff,
maximum_reconnection_backoff,
initial_connection_timeout,
);
MixTrafficController {
+54 -24
View File
@@ -12,12 +12,12 @@ use crate::config::{Config, SocketType};
use crate::sockets::{tcp, websocket};
use crypto::identity::MixIdentityKeyPair;
use directory_client::presence;
use futures::channel::mpsc;
use futures::channel::{mpsc, oneshot};
use log::*;
use nymsphinx::chunking::split_and_prepare_payloads;
use pemstore::pemstore::PemStore;
use sfw_provider_requests::AuthToken;
use sphinx::route::Destination;
use std::net::SocketAddr;
use tokio::runtime::Runtime;
use topology::NymTopology;
@@ -38,6 +38,9 @@ pub struct NymClient {
// to be used by "send" function or socket, etc
input_tx: Option<InputMessageSender>,
// to be used by "receive" function or socket, etc
receive_tx: Option<ReceivedBufferRequestSender>,
}
#[derive(Debug)]
@@ -64,6 +67,7 @@ impl NymClient {
config,
identity_keypair,
input_tx: None,
receive_tx: None,
}
}
@@ -75,18 +79,6 @@ impl NymClient {
)
}
async fn get_provider_socket_address<T: NymTopology>(
provider_id: String,
mut topology_accessor: TopologyAccessor<T>,
) -> SocketAddr {
topology_accessor.get_current_topology_clone().await.as_ref().expect("The current network topology is empty - are you using correct directory server?")
.providers()
.iter()
.find(|provider| provider.pub_key == provider_id)
.unwrap_or_else( || panic!("Could not find provider with id {:?} - are you sure it is still online? Perhaps try to run `nym-client init` again to obtain a new provider", provider_id))
.client_listener
}
// future constantly pumping loop cover traffic at some specified average rate
// the pumped traffic goes to the MixTrafficController
fn start_cover_traffic_stream<T: 'static + NymTopology>(
@@ -149,16 +141,26 @@ impl NymClient {
// the received messages are sent to ReceivedMessagesBuffer to be available to rest of the system
fn start_provider_poller<T: NymTopology>(
&mut self,
topology_accessor: TopologyAccessor<T>,
mut topology_accessor: TopologyAccessor<T>,
poller_input_tx: PolledMessagesSender,
) {
info!("Starting provider poller...");
// we already have our provider written in the config
let provider_id = self.config.get_provider_id();
// TODO: a slightly more graceful termination here
if !self.runtime.block_on(topology_accessor.is_routable()) {
panic!(
"The current network topology seem to be insufficient to route any packets through\
- check if enough nodes and a sfw-provider are online"
);
}
// TODO: a slightly more graceful termination here
let provider_client_listener_address = self.runtime.block_on(
Self::get_provider_socket_address(provider_id, topology_accessor),
);
topology_accessor.get_provider_socket_addr(&provider_id)
).unwrap_or_else(|| panic!("Could not find provider with id {:?}. It does not seem to be present in the current network topology.\
Are you sure it is still online? Perhaps try to run `nym-client init` again to obtain a new provider", provider_id));
let mut provider_poller = provider_poller::ProviderPoller::new(
poller_input_tx,
@@ -193,6 +195,7 @@ impl NymClient {
self.config.get_topology_refresh_rate(),
healthcheck_keys,
self.config.get_topology_resolution_timeout(),
self.config.get_healthcheck_connection_timeout(),
self.config.get_number_of_healthcheck_test_packets() as usize,
self.config.get_node_score_threshold(),
);
@@ -217,6 +220,7 @@ impl NymClient {
MixTrafficController::new(
self.config.get_packet_forwarding_initial_backoff(),
self.config.get_packet_forwarding_maximum_backoff(),
self.config.get_initial_connection_timeout(),
mix_rx,
)
})
@@ -255,13 +259,36 @@ impl NymClient {
}
/// EXPERIMENTAL DIRECT RUST API
/// It's entirely untested and there are absolutely no guarantees about it
pub fn send_message(&self, destination: Destination, message: Vec<u8>) {
self.input_tx
/// It's untested and there are absolutely no guarantees about it (but seems to have worked
/// well enough in local tests)
pub fn send_message(&mut self, destination: Destination, message: Vec<u8>) {
let split_message = split_and_prepare_payloads(&message);
debug!(
"Splitting message into {:?} fragments!",
split_message.len()
);
for message_fragment in split_message {
let input_msg = InputMessage(destination.clone(), message_fragment);
self.input_tx
.as_ref()
.expect("start method was not called before!")
.unbounded_send(input_msg)
.unwrap()
}
}
/// EXPERIMENTAL DIRECT RUST API
/// It's untested and there are absolutely no guarantees about it (but seems to have worked
/// well enough in local tests)
pub async fn check_for_messages_async(&self) -> Vec<Vec<u8>> {
let (res_tx, res_rx) = oneshot::channel();
self.receive_tx
.as_ref()
.expect("start method was not called before!")
.unbounded_send(InputMessage(destination, message))
.unwrap()
.unbounded_send(res_tx)
.unwrap();
res_rx.await.unwrap()
}
/// blocking version of `start` method. Will run forever (or until SIGINT is sent)
@@ -275,7 +302,7 @@ impl NymClient {
}
println!(
"Received SIGINT - the mixnode will terminate now (threads are not YET nicely stopped)"
"Received SIGINT - the client will terminate now (threads are not YET nicely stopped)"
);
}
@@ -316,9 +343,12 @@ impl NymClient {
self.start_real_traffic_stream(shared_topology_accessor.clone(), mix_tx, input_rx);
self.start_socket_listener(
shared_topology_accessor,
received_messages_buffer_output_tx,
received_messages_buffer_output_tx.clone(),
input_tx.clone(),
);
self.input_tx = Some(input_tx);
self.receive_tx = Some(received_messages_buffer_output_tx);
info!("Client startup finished!");
}
}
+8 -2
View File
@@ -78,7 +78,11 @@ impl ProviderPoller {
Ok(messages) => messages,
};
let good_messages = messages
// todo: this will probably need to be updated at some point due to changed structure
// of nym-sphinx messages sent. However, for time being this is still compatible.
// Basically it is more of a personal note of if client keeps getting errors and is
// not getting messages, look at this filter here.
let good_messages: Vec<_> = messages
.into_iter()
.filter(|message| message != loop_message && message != dummy_message)
.collect();
@@ -88,7 +92,9 @@ impl ProviderPoller {
// - we run out of memory
// - the receiver channel is closed
// in either case there's no recovery and we can only panic
self.poller_tx.unbounded_send(good_messages).unwrap();
if !good_messages.is_empty() {
self.poller_tx.unbounded_send(good_messages).unwrap();
}
tokio::time::delay_for(self.polling_rate).await;
}
@@ -127,6 +127,9 @@ impl<T: 'static + NymTopology> OutQueueControl<T> {
self.mix_tx
.unbounded_send(MixMessage::new(next_packet.0, next_packet.1))
.unwrap();
// JS: Not entirely sure why or how it fixes stuff, but without the yield call,
// the UnboundedReceiver [of mix_rx] will not get a chance to read anything
tokio::task::yield_now().await;
}
pub(crate) async fn run_out_queue_control(mut self) {
+32 -3
View File
@@ -3,6 +3,7 @@ use futures::channel::{mpsc, oneshot};
use futures::lock::Mutex;
use futures::StreamExt;
use log::*;
use nymsphinx::chunking::reconstruction::MessageReconstructor;
use std::sync::Arc;
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
@@ -13,6 +14,7 @@ pub(crate) type ReceivedBufferRequestReceiver = mpsc::UnboundedReceiver<Received
struct ReceivedMessagesBufferInner {
messages: Vec<Vec<u8>>,
message_reconstructor: MessageReconstructor,
}
#[derive(Debug, Clone)]
@@ -27,15 +29,40 @@ impl ReceivedMessagesBuffer {
ReceivedMessagesBuffer {
inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner {
messages: Vec::new(),
message_reconstructor: MessageReconstructor::new(),
})),
}
}
async fn add_new_messages(&mut self, msgs: Vec<Vec<u8>>) {
async fn add_reconstructed_messages(&mut self, msgs: Vec<Vec<u8>>) {
debug!("Adding {:?} new messages to the buffer!", msgs.len());
trace!("Adding new messages to the buffer! {:?}", msgs);
self.inner.lock().await.messages.extend(msgs)
}
async fn add_new_message_fragments(&mut self, msgs: Vec<Vec<u8>>) {
debug!(
"Adding {:?} new message fragments to the buffer!",
msgs.len()
);
trace!("Adding new message fragments to the buffer! {:?}", msgs);
let mut completed_messages = Vec::new();
let mut inner_guard = self.inner.lock().await;
for msg_fragment in msgs {
if let Some(reconstructed_message) =
inner_guard.message_reconstructor.new_fragment(msg_fragment)
{
completed_messages.push(reconstructed_message);
}
}
// make sure to drop the lock to not deadlock
drop(inner_guard);
if !completed_messages.is_empty() {
self.add_reconstructed_messages(completed_messages).await;
}
}
async fn acquire_and_empty(&mut self) -> Vec<Vec<u8>> {
trace!("Emptying the buffer and returning all messages");
let mut mutex_guard = self.inner.lock().await;
@@ -67,7 +94,7 @@ impl RequestReceiver {
error!(
"Failed to send the messages to the requester. Adding them back to the buffer"
);
self.received_buffer.add_new_messages(failed_messages).await;
self.received_buffer.add_reconstructed_messages(failed_messages).await;
}
}
})
@@ -92,7 +119,9 @@ impl MessageReceiver {
fn start(mut self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
while let Some(new_messages) = self.poller_receiver.next().await {
self.received_buffer.add_new_messages(new_messages).await;
self.received_buffer
.add_new_message_fragments(new_messages)
.await;
}
})
}
+26
View File
@@ -8,6 +8,7 @@ use std::time;
use std::time::Duration;
use tokio::runtime::Handle;
// use tokio::sync::RwLock;
use std::net::SocketAddr;
use tokio::task::JoinHandle;
use topology::{provider, NymTopology};
@@ -41,6 +42,25 @@ impl<T: NymTopology> TopologyAccessor<T> {
self.inner.lock().await.update(new_topology);
}
pub(crate) async fn get_provider_socket_addr(&mut self, id: &str) -> Option<SocketAddr> {
match &self.inner.lock().await.0 {
None => None,
Some(ref topology) => topology
.providers()
.iter()
.find(|provider| provider.pub_key == id)
.map(|provider| provider.client_listener),
}
}
// only used by the client at startup to get a slightly more reasonable error message
pub(crate) async fn is_routable(&self) -> bool {
match &self.inner.lock().await.0 {
None => false,
Some(ref topology) => topology.can_construct_path_through(),
}
}
// Unless you absolutely need the entire topology, use `random_route` instead
pub(crate) async fn get_current_topology_clone(&mut self) -> Option<T> {
self.inner.lock().await.0.clone()
@@ -90,6 +110,7 @@ pub(crate) struct TopologyRefresherConfig {
refresh_rate: time::Duration,
identity_keypair: MixIdentityKeyPair,
resolution_timeout: time::Duration,
connection_timeout: time::Duration,
number_test_packets: usize,
node_score_threshold: f64,
}
@@ -100,6 +121,7 @@ impl TopologyRefresherConfig {
refresh_rate: time::Duration,
identity_keypair: MixIdentityKeyPair,
resolution_timeout: time::Duration,
connection_timeout: time::Duration,
number_test_packets: usize,
node_score_threshold: f64,
) -> Self {
@@ -108,6 +130,7 @@ impl TopologyRefresherConfig {
refresh_rate,
identity_keypair,
resolution_timeout,
connection_timeout,
number_test_packets,
node_score_threshold,
}
@@ -130,6 +153,7 @@ impl<T: 'static + NymTopology> TopologyRefresher<T> {
// this is a temporary solution as the healthcheck will eventually be moved to validators
let health_checker = healthcheck::HealthChecker::new(
cfg.resolution_timeout,
cfg.connection_timeout,
cfg.number_test_packets,
cfg.identity_keypair,
);
@@ -163,6 +187,8 @@ impl<T: 'static + NymTopology> TopologyRefresher<T> {
Ok(scores) => scores,
};
debug!("{}", healthcheck_scores);
let healthy_topology = healthcheck_scores
.filter_topology_by_score(&version_filtered_topology, self.node_score_threshold);
+22 -1
View File
@@ -20,6 +20,8 @@ const DEFAULT_TOPOLOGY_REFRESH_RATE: u64 = 10_000;
const DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT: u64 = 5_000;
const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: u64 = 10_000; // 10s
const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: u64 = 300_000; // 5min
const DEFAULT_INITIAL_CONNECTION_TIMEOUT: u64 = 1_500; // 1.5s
const DEFAULT_HEALTHCHECK_CONNECTION_TIMEOUT: u64 = DEFAULT_INITIAL_CONNECTION_TIMEOUT;
const DEFAULT_NUMBER_OF_HEALTHCHECK_TEST_PACKETS: u64 = 2;
const DEFAULT_NODE_SCORE_THRESHOLD: f64 = 0.0;
@@ -213,6 +215,14 @@ impl Config {
pub fn get_packet_forwarding_maximum_backoff(&self) -> time::Duration {
time::Duration::from_millis(self.debug.packet_forwarding_maximum_backoff)
}
pub fn get_initial_connection_timeout(&self) -> time::Duration {
time::Duration::from_millis(self.debug.initial_connection_timeout)
}
pub fn get_healthcheck_connection_timeout(&self) -> time::Duration {
time::Duration::from_millis(self.debug.healthcheck_connection_timeout)
}
}
fn de_option_string<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
@@ -317,7 +327,7 @@ impl Default for Logging {
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
#[serde(default, deny_unknown_fields)]
pub struct Debug {
/// The parameter of Poisson distribution determining how long, on average,
/// sent packet is going to be delayed at any given mix node.
@@ -376,6 +386,15 @@ pub struct Debug {
/// forwarding sphinx packets.
/// The provided value is interpreted as milliseconds.
packet_forwarding_maximum_backoff: u64,
/// Timeout for establishing initial connection when trying to forward a sphinx packet.
/// The provider value is interpreted as milliseconds.
initial_connection_timeout: u64,
/// Timeout for establishing initial connection when trying to forward a sphinx packet
/// during healthcheck.
/// The provider value is interpreted as milliseconds.
healthcheck_connection_timeout: u64,
}
impl Default for Debug {
@@ -392,6 +411,8 @@ impl Default for Debug {
node_score_threshold: DEFAULT_NODE_SCORE_THRESHOLD,
packet_forwarding_initial_backoff: DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF,
packet_forwarding_maximum_backoff: DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF,
initial_connection_timeout: DEFAULT_INITIAL_CONNECTION_TIMEOUT,
healthcheck_connection_timeout: DEFAULT_HEALTHCHECK_CONNECTION_TIMEOUT,
}
}
}
-52
View File
@@ -62,62 +62,10 @@ listening_port = {{ socket.listening_port }}
[debug]
# The parameter of Poisson distribution determining how long, on average,
# sent packet is going to be delayed at any given mix node.
# So for a packet going through three mix nodes, on average, it will take three times this value
# until the packet reaches its destination.
# The provided value is interpreted as milliseconds.
average_packet_delay = {{ debug.average_packet_delay }}
# The parameter of Poisson distribution determining how long, on average,
# it is going to take for another loop cover traffic message to be sent.
# The provided value is interpreted as milliseconds.
loop_cover_traffic_average_delay = {{ debug.loop_cover_traffic_average_delay }}
# The uniform delay every which clients are querying the providers for received packets.
# The provided value is interpreted as milliseconds.
fetch_message_delay = {{ debug.fetch_message_delay }}
# The parameter of Poisson distribution determining how long, on average,
# it is going to take another 'real traffic stream' message to be sent.
# If no real packets are available and cover traffic is enabled,
# a loop cover message is sent instead in order to preserve the rate.
# The provided value is interpreted as milliseconds.
message_sending_average_delay = {{ debug.message_sending_average_delay }}
# Whether loop cover messages should be sent to respect message_sending_rate.
# In the case of it being disabled and not having enough real traffic
# waiting to be sent the actual sending rate is going be lower than the desired value
# thus decreasing the anonymity.
rate_compliant_cover_messages_disabled = {{ debug.rate_compliant_cover_messages_disabled }}
# The uniform delay every which clients are querying the directory server
# to try to obtain a compatible network topology to send sphinx packets through.
# The provided value is interpreted as milliseconds.
topology_refresh_rate = {{ debug.topology_refresh_rate }}
# During topology refresh, test packets are sent through every single possible network
# path. This timeout determines waiting period until it is decided that the packet
# did not reach its destination.
# The provided value is interpreted as milliseconds.
topology_resolution_timeout = {{ debug.topology_resolution_timeout }}
# How many packets should be sent through each path during the healthcheck
number_of_healthcheck_test_packets = {{ debug.number_of_healthcheck_test_packets }}
# In the current healthcheck implementation, threshold indicating percentage of packets
# node received during healthcheck. Node's score must be above that value to be
# considered healthy.
node_score_threshold = {{ debug.node_score_threshold }}
# Initial value of an exponential backoff to reconnect to dropped TCP connection when
# forwarding sphinx packets.
# The provided value is interpreted as milliseconds.
packet_forwarding_initial_backoff = {{ debug.packet_forwarding_initial_backoff }}
# Maximum value of an exponential backoff to reconnect to dropped TCP connection when
# forwarding sphinx packets.
# The provided value is interpreted as milliseconds.
packet_forwarding_maximum_backoff = {{ debug.packet_forwarding_maximum_backoff }}
"#
}
+13 -12
View File
@@ -6,6 +6,7 @@ use futures::future::FutureExt;
use futures::io::Error;
use futures::SinkExt;
use log::*;
use nymsphinx::chunking::split_and_prepare_payloads;
use sphinx::route::{Destination, DestinationAddressBytes};
use std::convert::TryFrom;
use std::io;
@@ -94,21 +95,21 @@ impl ClientRequest {
async fn handle_send(
msg: Vec<u8>,
recipient_address: DestinationAddressBytes,
mut input_tx: mpsc::UnboundedSender<InputMessage>,
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
),
};
}
let dummy_surb = [0; 16];
let input_msg = InputMessage(Destination::new(recipient_address, dummy_surb), msg);
input_tx.send(input_msg).await.unwrap();
// in case the message is too long and needs to be split into multiple packets:
let split_message = split_and_prepare_payloads(&msg);
for message_fragment in split_message {
let input_msg = InputMessage(
Destination::new(recipient_address.clone(), dummy_surb),
message_fragment,
);
input_tx.unbounded_send(input_msg).unwrap();
}
ServerResponse::Send
}
+10 -11
View File
@@ -5,6 +5,7 @@ use crate::sockets::websocket::types::{ClientRequest, ServerResponse};
use futures::channel::oneshot;
use futures::{SinkExt, StreamExt};
use log::*;
use nymsphinx::chunking::split_and_prepare_payloads;
use sphinx::route::{Destination, DestinationAddressBytes};
use std::convert::TryFrom;
use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
@@ -71,23 +72,21 @@ impl<T: NymTopology> Connection<T> {
fn handle_text_send(&self, msg: String, recipient_address: String) -> ServerResponse {
let message_bytes = msg.into_bytes();
if message_bytes.len() > sphinx::constants::MAXIMUM_PLAINTEXT_LENGTH {
return ServerResponse::Error {
message: format!(
"message too long. Sent {} bytes, but the maximum is {}",
message_bytes.len(),
sphinx::constants::MAXIMUM_PLAINTEXT_LENGTH
),
};
}
// TODO: the below can panic if recipient_address is malformed, but it should be
// resolved when refactoring sphinx code to make `from_base58_string` return a Result
let address = DestinationAddressBytes::from_base58_string(recipient_address);
let dummy_surb = [0; 16];
let input_msg = InputMessage(Destination::new(address, dummy_surb), message_bytes);
self.msg_input.unbounded_send(input_msg).unwrap();
// in case the message is too long and needs to be split into multiple packets:
let split_message = split_and_prepare_payloads(&message_bytes);
for message_fragment in split_message {
let input_msg = InputMessage(
Destination::new(address.clone(), dummy_surb),
message_fragment,
);
self.msg_input.unbounded_send(input_msg).unwrap();
}
ServerResponse::Send
}
+8
View File
@@ -1,3 +1,11 @@
#!/bin/bash
# set CHANGELOG_GITHUB_TOKEN in your .bashrc file
# For each version, you can add a release summary with text, images, gif animations, etc, and show new features and notes clearly to the user. This is done using GitHub metadata.
# Example: adding the release summary for v1.0.0:
# 1. Create a new GitHub Issue
# 2. In the Issue's Description field, add your release summary content
# 3. Set the Issue Label `release-summary` and add it to the GitHub Milestone `v1.0.0`
# 4. Close the Issue and execute `github-changelog-generator`
github_changelog_generator -u nymtech -p nym --exclude-tags 0.1.0 --token "$CHANGELOG_GITHUB_TOKEN"
Executable → Regular
View File
+3 -2
View File
@@ -1,7 +1,7 @@
[package]
build = "build.rs"
name = "nym-sfw-provider"
version = "0.5.0"
version = "0.6.0"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
edition = "2018"
@@ -29,6 +29,7 @@ config = {path = "../common/config"}
directory-client = { path = "../common/clients/directory-client" }
pemstore = {path = "../common/pemstore"}
sfw-provider-requests = { path = "./sfw-provider-requests" }
topology = {path = "../common/topology"}
# this dependency is due to requiring to know content of loop message. however, mix-client module itself
# is going to be removed or renamed at some point as it no longer servers its purpose of sending traffic to mix network
@@ -37,7 +38,7 @@ mix-client = { path = "../common/clients/mix-client" }
## will be moved to proper dependencies once released
sphinx = { git = "https://github.com/nymtech/sphinx", rev="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
[build-dependencies]
built = "0.3.2"
@@ -8,4 +8,4 @@ edition = "2018"
[dependencies]
bs58 = "0.3.0"
sphinx = { git = "https://github.com/nymtech/sphinx", rev="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sphinx = { git = "https://github.com/nymtech/sphinx", rev="44d8f2aece5049eaa4fe84b7948758ce82b4b80d" }
+1 -1
View File
@@ -468,7 +468,7 @@ impl Default for Logging {
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
#[serde(default, deny_unknown_fields)]
pub struct Debug {
/// Directory server to which the server will be reporting their presence data.
presence_directory_server: String,
-21
View File
@@ -75,26 +75,5 @@ ledger_path = '{{ clients_endpoint.ledger_path }}'
# TODO
##### debug configuration options #####
# The following options should not be modified unless you know EXACTLY what you are doing
# as if set incorrectly, they may impact your anonymity.
[debug]
# Directory server to which the server will be reporting their presence data.
presence_directory_server = '{{ debug.presence_directory_server}}'
# Delay between each subsequent presence data being sent.
presence_sending_delay = {{ debug.presence_sending_delay }}
# Length of filenames for new client messages.
stored_messages_filename_length = {{ debug.stored_messages_filename_length }}
# number of messages client gets on each request
# if there are no real messages, dummy ones are create to always return
# `message_retrieval_limit` total messages
message_retrieval_limit = {{ debug.message_retrieval_limit }}
"#
}
@@ -65,10 +65,16 @@ async fn process_socket_connection(
// plus realistically it wouldn't really introduce any speed up
Ok(n) => process_request(&mut socket, &buf[0..n], &mut request_processor).await,
Err(e) => {
warn!(
"failed to read from socket. Closing the connection; err = {:?}",
e
);
if e.kind() == io::ErrorKind::UnexpectedEof {
debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\
Also closing the connection on this end.", socket.peer_addr())
} else {
warn!(
"failed to read from socket (source: {:?}). Closing the connection; err = {:?}",
socket.peer_addr(),
e
);
}
return;
}
};
@@ -36,6 +36,8 @@ async fn process_socket_connection(
return;
}
Ok(n) => {
// If I understand it correctly, this if should never be executed as if `read_exact`
// does not fill buffer, it will throw UnexpectedEof?
if n != sphinx::PACKET_SIZE {
warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, sphinx::PACKET_SIZE);
continue;
@@ -45,10 +47,16 @@ async fn process_socket_connection(
tokio::spawn(process_received_packet(buf, packet_processor.clone()))
}
Err(e) => {
warn!(
"failed to read from socket. Closing the connection; err = {:?}",
e
);
if e.kind() == io::ErrorKind::UnexpectedEof {
debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\
Also closing the connection on this end.", socket.peer_addr())
} else {
warn!(
"failed to read from socket (source: {:?}). Closing the connection; err = {:?}",
socket.peer_addr(),
e
);
}
return;
}
};
+26
View File
@@ -3,9 +3,11 @@ use crate::config::Config;
use crate::provider::client_handling::ledger::ClientLedger;
use crate::provider::storage::ClientStorage;
use crypto::encryption;
use directory_client::presence::Topology;
use log::*;
use pemstore::pemstore::PemStore;
use tokio::runtime::Runtime;
use topology::NymTopology;
mod client_handling;
mod mix_handling;
@@ -85,11 +87,33 @@ impl ServiceProvider {
);
}
fn check_if_same_ip_provider_exists(&self) -> Option<String> {
// TODO: once we change to graph topology this here will need to be updated!
let topology = Topology::new(self.config.get_presence_directory_server());
let existing_providers_presence = topology.mix_provider_nodes;
existing_providers_presence
.iter()
.find(|node| {
node.mixnet_listener == self.config.get_mix_announce_address()
|| node.client_listener == self.config.get_clients_announce_address()
})
.map(|node| node.pub_key.clone())
}
pub fn run(&mut self) {
// A possible future optimisation, depending on bottlenecks and resource usage:
// considering, presumably, there will be more mix packets received than client requests:
// create 2 separate runtimes - one with bigger threadpool dedicated solely for
// the mix handling and the other one for the rest of tasks
info!("Starting nym sfw-provider");
if let Some(duplicate_provider_key) = self.check_if_same_ip_provider_exists() {
error!(
"Our announce-host is identical to one of existing nodes! (its key is {:?}",
duplicate_provider_key
);
return;
}
let client_storage = ClientStorage::new(
self.config.get_message_retrieval_limit() as usize,
@@ -101,6 +125,8 @@ impl ServiceProvider {
self.start_mix_socket_listener(client_storage.clone());
self.start_client_socket_listener(client_storage);
info!("Finished nym sfw-provider startup procedure - it should now be able to receive mix and client traffic!");
if let Err(e) = self.runtime.block_on(tokio::signal::ctrl_c()) {
error!(
"There was an error while capturing SIGINT - {:?}. We will terminate regardless",
+2 -2
View File
@@ -1,8 +1,8 @@
[package]
build = "build.rs"
name = "nym-validator"
version = "0.5.0"
authors = ["Jedrzej Stuczynski <andrew@nymtech.net>"]
version = "0.6.0"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jedrzej Stuczynski <andrew@nymtech.net>"]
edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+11 -1
View File
@@ -11,6 +11,7 @@ mod template;
// 'MIXMINING'
const DEFAULT_MIX_MINING_DELAY: u64 = 10_000;
const DEFAULT_MIX_MINING_RESOLUTION_TIMEOUT: u64 = 5_000;
const DEFAULT_MIX_MINING_CONNECTION_TIMEOUT: u64 = 1_500;
const DEFAULT_NUMBER_OF_MIX_MINING_TEST_PACKETS: u64 = 2;
@@ -139,6 +140,10 @@ impl Config {
pub fn get_mix_mining_number_of_test_packets(&self) -> u64 {
self.mix_mining.number_of_test_packets
}
pub fn get_healthcheck_connection_timeout(&self) -> time::Duration {
time::Duration::from_millis(self.mix_mining.connection_timeout)
}
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
@@ -190,6 +195,10 @@ pub struct MixMining {
/// The provided value is interpreted as milliseconds.
resolution_timeout: u64,
/// Timeout for trying to establish connection to node endpoints.
/// The provided value is interpreted as milliseconds.
connection_timeout: u64,
/// How many packets should be sent through each path during the mix-mining procedure.
number_of_test_packets: u64,
}
@@ -201,6 +210,7 @@ impl Default for MixMining {
run_delay: DEFAULT_MIX_MINING_DELAY,
resolution_timeout: DEFAULT_MIX_MINING_RESOLUTION_TIMEOUT,
number_of_test_packets: DEFAULT_NUMBER_OF_MIX_MINING_TEST_PACKETS,
connection_timeout: DEFAULT_MIX_MINING_CONNECTION_TIMEOUT,
}
}
}
@@ -226,7 +236,7 @@ impl Default for Logging {
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
#[serde(default, deny_unknown_fields)]
pub struct Debug {
/// Directory server to which the server will be reporting their presence data.
presence_directory_server: String,
+4 -13
View File
@@ -43,6 +43,10 @@ run_delay = {{ mix_mining.run_delay }}
# The provided value is interpreted as milliseconds.
resolution_timeout = {{ mix_mining.resolution_timeout }}
# Timeout for trying to establish connection to node endpoints.
# The provided value is interpreted as milliseconds.
connection_timeout = {{ mix_mining.connection_timeout }}
# How many packets should be sent through each path during the mix-mining procedure.
number_of_test_packets = {{ mix_mining.number_of_test_packets }}
@@ -59,18 +63,5 @@ number_of_test_packets = {{ mix_mining.number_of_test_packets }}
# TODO
##### debug configuration options #####
# The following options should not be modified unless you know EXACTLY what you are doing
# as if set incorrectly, they may impact your anonymity.
[debug]
# Directory server to which the server will be reporting their presence data.
presence_directory_server = '{{ debug.presence_directory_server }}'
# Delay between each subsequent presence data being sent.
presence_sending_delay = {{ debug.presence_sending_delay }}
"#
}
+1
View File
@@ -20,6 +20,7 @@ impl Validator {
let dummy_healthcheck_keypair = MixIdentityKeyPair::new();
let hc = HealthChecker::new(
config.get_mix_mining_resolution_timeout(),
config.get_healthcheck_connection_timeout(),
config.get_mix_mining_number_of_test_packets() as usize,
dummy_healthcheck_keypair,
);