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

Merge branch 'release/v0.5.0'

Browse Source
This commit is contained in:
Dave Hrycyszyn
2020-03-23 10:35:32 +00:00
parent 3ccda81a24 d576a17d33
commit 1e99e6dc07
70 changed files with 1251 additions and 697 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+260 -47
View File
@@ -11,7 +11,7 @@ dependencies = [
"env_logger 0.7.1",
"futures 0.3.4",
"integer-encoding",
"log",
"log 0.4.8",
"protobuf",
"protobuf-codegen-pure",
"tokio 0.1.22",
@@ -21,7 +21,7 @@ dependencies = [
name = "addressing"
version = "0.1.0"
dependencies = [
"log",
"log 0.4.8",
"pretty_env_logger",
]
@@ -168,6 +168,16 @@ dependencies = [
"libc",
]
[[package]]
name = "base64"
version = "0.9.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "489d6c0ed21b11d038c31b6ceccca973e65d73ba3bd8ecb9a2babf5546164643"
dependencies = [
"byteorder",
"safemem",
]
[[package]]
name = "base64"
version = "0.10.1"
@@ -412,7 +422,7 @@ dependencies = [
"cookie",
"failure",
"idna 0.1.5",
"log",
"log 0.4.8",
"publicsuffix",
"serde",
"serde_json",
@@ -499,7 +509,7 @@ version = "0.1.0"
dependencies = [
"bs58",
"curve25519-dalek",
"log",
"log 0.4.8",
"pretty_env_logger",
"rand 0.7.3",
"rand_os",
@@ -558,7 +568,7 @@ dependencies = [
name = "directory-client"
version = "0.1.0"
dependencies = [
"log",
"log 0.4.8",
"mockito",
"pretty_env_logger",
"reqwest",
@@ -623,7 +633,7 @@ checksum = "aafcde04e90a5226a6443b7aabdb016ba2f8307c847d524724bd9b346dd1a2d3"
dependencies = [
"atty",
"humantime",
"log",
"log 0.4.8",
"regex",
"termcolor",
]
@@ -636,7 +646,7 @@ checksum = "44533bbbb3bb3c1fa17d9f2e4e38bbbaf8396ba82193c4cb1b6445d711445d36"
dependencies = [
"atty",
"humantime",
"log",
"log 0.4.8",
"regex",
"termcolor",
]
@@ -647,7 +657,7 @@ version = "0.12.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d371106cc88ffdfb1eabd7111e432da544f16f3e2d7bf1dfe8bf575f1df045cd"
dependencies = [
"version_check",
"version_check 0.9.1",
]
[[package]]
@@ -881,7 +891,7 @@ dependencies = [
"bitflags",
"libc",
"libgit2-sys",
"log",
"log 0.4.8",
"url 2.1.1",
]
@@ -897,7 +907,7 @@ dependencies = [
"futures 0.1.29",
"http 0.1.21",
"indexmap",
"log",
"log 0.4.8",
"slab",
"string",
"tokio-io",
@@ -909,7 +919,7 @@ version = "3.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba758d094d31274eb49d15da6f326b96bf3185239a6359bf684f3d5321148900"
dependencies = [
"log",
"log 0.4.8",
"pest",
"pest_derive",
"quick-error",
@@ -926,7 +936,7 @@ dependencies = [
"directory-client",
"futures 0.3.4",
"itertools",
"log",
"log 0.4.8",
"mix-client",
"pretty_env_logger",
"provider-client",
@@ -1018,6 +1028,25 @@ dependencies = [
"quick-error",
]
[[package]]
name = "hyper"
version = "0.10.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0a0652d9a2609a968c14be1a9ea00bf4b1d64e2e1f53a1b51b6fff3a6e829273"
dependencies = [
"base64 0.9.3",
"httparse",
"language-tags",
"log 0.3.9",
"mime 0.2.6",
"num_cpus",
"time",
"traitobject",
"typeable",
"unicase 1.4.2",
"url 1.7.2",
]
[[package]]
name = "hyper"
version = "0.12.35"
@@ -1033,7 +1062,7 @@ dependencies = [
"httparse",
"iovec",
"itoa",
"log",
"log 0.4.8",
"net2",
"rustc_version",
"time",
@@ -1056,7 +1085,7 @@ checksum = "3a800d6aa50af4b5850b2b0f659625ce9504df908e9733b635720483be26174f"
dependencies = [
"bytes 0.4.12",
"futures 0.1.29",
"hyper",
"hyper 0.12.35",
"native-tls",
"tokio-io",
]
@@ -1119,6 +1148,22 @@ dependencies = [
"libc",
]
[[package]]
name = "iron"
version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c6d308ca2d884650a8bf9ed2ff4cb13fbb2207b71f64cda11dc9b892067295e8"
dependencies = [
"hyper 0.10.16",
"log 0.3.9",
"mime_guess 1.8.8",
"modifier",
"num_cpus",
"plugin",
"typemap",
"url 1.7.2",
]
[[package]]
name = "itertools"
version = "0.8.2"
@@ -1159,6 +1204,12 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c33070833c9ee02266356de0c43f723152bd38bd96ddf52c82b3af10c9138b28"
[[package]]
name = "language-tags"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a91d884b6667cd606bb5a69aa0c99ba811a115fc68915e7056ec08a46e93199a"
[[package]]
name = "lazy_static"
version = "1.4.0"
@@ -1216,6 +1267,15 @@ dependencies = [
"scopeguard",
]
[[package]]
name = "log"
version = "0.3.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e19e8d5c34a3e0e2223db8e060f9e8264aeeb5c5fc64a4ee9965c062211c024b"
dependencies = [
"log 0.4.8",
]
[[package]]
name = "log"
version = "0.4.8"
@@ -1258,20 +1318,41 @@ dependencies = [
"rustc_version",
]
[[package]]
name = "mime"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba626b8a6de5da682e1caa06bdb42a335aee5a84db8e5046a3e8ab17ba0a3ae0"
dependencies = [
"log 0.3.9",
]
[[package]]
name = "mime"
version = "0.3.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d"
[[package]]
name = "mime_guess"
version = "1.8.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "216929a5ee4dd316b1702eedf5e74548c123d370f47841ceaac38ca154690ca3"
dependencies = [
"mime 0.2.6",
"phf",
"phf_codegen",
"unicase 1.4.2",
]
[[package]]
name = "mime_guess"
version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1a0ed03949aef72dbdf3116a383d7b38b4768e6f960528cd6a6044aa9ed68599"
dependencies = [
"mime",
"unicase",
"mime 0.3.16",
"unicase 2.6.0",
]
[[package]]
@@ -1295,7 +1376,7 @@ dependencies = [
"iovec",
"kernel32-sys",
"libc",
"log",
"log 0.4.8",
"miow 0.2.1",
"net2",
"slab",
@@ -1308,7 +1389,7 @@ version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f5e374eff525ce1c5b7687c4cef63943e7686524a387933ad27ca7ec43779cb3"
dependencies = [
"log",
"log 0.4.8",
"mio",
"miow 0.3.3",
"winapi 0.3.8",
@@ -1352,7 +1433,7 @@ name = "mix-client"
version = "0.1.0"
dependencies = [
"addressing",
"log",
"log 0.4.8",
"pretty_env_logger",
"rand 0.7.3",
"rand_distr",
@@ -1372,19 +1453,25 @@ dependencies = [
"difference",
"httparse",
"lazy_static",
"log",
"log 0.4.8",
"percent-encoding 2.1.0",
"rand 0.7.3",
"regex",
"serde_json",
]
[[package]]
name = "modifier"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "41f5c9112cb662acd3b204077e0de5bc66305fa8df65c8019d5adb10e9ab6e58"
[[package]]
name = "multi-tcp-client"
version = "0.1.0"
dependencies = [
"futures 0.3.4",
"log",
"log 0.4.8",
"tokio 0.2.12",
]
@@ -1396,7 +1483,7 @@ checksum = "4b2df1a4c22fd44a62147fd8f13dd0f95c9d8ca7b2610299b2a2f9cf8964274e"
dependencies = [
"lazy_static",
"libc",
"log",
"log 0.4.8",
"openssl",
"openssl-probe",
"openssl-sys",
@@ -1448,7 +1535,7 @@ dependencies = [
[[package]]
name = "nym-client"
version = "0.5.0-rc.1"
version = "0.5.0"
dependencies = [
"addressing",
"bs58",
@@ -1462,7 +1549,7 @@ dependencies = [
"dotenv",
"futures 0.3.4",
"healthcheck",
"log",
"log 0.4.8",
"mix-client",
"multi-tcp-client",
"pem",
@@ -1482,7 +1569,7 @@ dependencies = [
[[package]]
name = "nym-mixnode"
version = "0.5.0-rc.1"
version = "0.5.0"
dependencies = [
"addressing",
"bs58",
@@ -1495,7 +1582,7 @@ dependencies = [
"dirs",
"dotenv",
"futures 0.3.4",
"log",
"log 0.4.8",
"multi-tcp-client",
"pemstore",
"pretty_env_logger",
@@ -1507,7 +1594,7 @@ dependencies = [
[[package]]
name = "nym-sfw-provider"
version = "0.5.0-rc.1"
version = "0.5.0"
dependencies = [
"bs58",
"built",
@@ -1520,7 +1607,8 @@ dependencies = [
"dotenv",
"futures 0.3.4",
"hmac",
"log",
"log 0.4.8",
"mix-client",
"pemstore",
"pretty_env_logger",
"rand 0.7.3",
@@ -1535,7 +1623,7 @@ dependencies = [
[[package]]
name = "nym-validator"
version = "0.5.0-rc.1"
version = "0.5.0"
dependencies = [
"abci",
"built",
@@ -1548,9 +1636,12 @@ dependencies = [
"dotenv",
"futures 0.3.4",
"healthcheck",
"log",
"iron",
"log 0.4.8",
"pretty_env_logger",
"router",
"serde",
"serde_json",
"tempfile",
"tokio 0.2.12",
"topology",
@@ -1637,7 +1728,7 @@ name = "pemstore"
version = "0.1.0"
dependencies = [
"crypto",
"log",
"log 0.4.8",
"pem",
"pretty_env_logger",
]
@@ -1697,6 +1788,45 @@ dependencies = [
"sha-1",
]
[[package]]
name = "phf"
version = "0.7.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b3da44b85f8e8dfaec21adae67f95d93244b2ecf6ad2a692320598dcc8e6dd18"
dependencies = [
"phf_shared",
]
[[package]]
name = "phf_codegen"
version = "0.7.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b03e85129e324ad4166b06b2c7491ae27fe3ec353af72e72cd1654c7225d517e"
dependencies = [
"phf_generator",
"phf_shared",
]
[[package]]
name = "phf_generator"
version = "0.7.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09364cc93c159b8b06b1f4dd8a4398984503483891b0c26b867cf431fb132662"
dependencies = [
"phf_shared",
"rand 0.6.5",
]
[[package]]
name = "phf_shared"
version = "0.7.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "234f71a15de2288bcb7e3b6515828d22af7ec8598ee6d24c3b526fa0a80b67a0"
dependencies = [
"siphasher",
"unicase 1.4.2",
]
[[package]]
name = "pin-project"
version = "0.4.8"
@@ -1735,6 +1865,15 @@ version = "0.3.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05da548ad6865900e60eaba7f589cc0783590a92e940c26953ff81ddbab2d677"
[[package]]
name = "plugin"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1a6a0dc3910bc8db877ffed8e457763b317cf880df4ae19109b9f77d277cf6e0"
dependencies = [
"typemap",
]
[[package]]
name = "ppv-lite86"
version = "0.2.6"
@@ -1749,7 +1888,7 @@ checksum = "717ee476b1690853d222af4634056d830b5197ffd747726a9a1eee6da9f49074"
dependencies = [
"chrono",
"env_logger 0.6.2",
"log",
"log 0.4.8",
]
[[package]]
@@ -1834,7 +1973,7 @@ name = "provider-client"
version = "0.1.0"
dependencies = [
"futures 0.3.4",
"log",
"log 0.4.8",
"pretty_env_logger",
"sfw-provider-requests",
"sphinx",
@@ -2092,11 +2231,11 @@ dependencies = [
"flate2",
"futures 0.1.29",
"http 0.1.21",
"hyper",
"hyper 0.12.35",
"hyper-tls",
"log",
"mime",
"mime_guess",
"log 0.4.8",
"mime 0.3.16",
"mime_guess 2.0.1",
"native-tls",
"serde",
"serde_json",
@@ -2112,6 +2251,23 @@ dependencies = [
"winreg",
]
[[package]]
name = "route-recognizer"
version = "0.1.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ea509065eb0b3c446acdd0102f0d46567dc30902dc0be91d6552035d92b0f4f8"
[[package]]
name = "router"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc63b6f3b8895b0d04e816b2b1aa58fdba2d5acca3cbb8f0ab8e017347d57397"
dependencies = [
"iron",
"route-recognizer",
"url 1.7.2",
]
[[package]]
name = "rust-argon2"
version = "0.7.0"
@@ -2156,6 +2312,12 @@ version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bfa8506c1de11c9c4e4c38863ccbe02a305c8188e85a05a784c9e11e1c3910c8"
[[package]]
name = "safemem"
version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ef703b7cb59335eae2eb93ceb664c0eb7ea6bf567079d843e09420219668e072"
[[package]]
name = "schannel"
version = "0.1.17"
@@ -2293,6 +2455,12 @@ dependencies = [
"libc",
]
[[package]]
name = "siphasher"
version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b8de496cf83d4ed58b6be86c3a275b8602f6ffe98d3024a869e124147a9a3ac"
[[package]]
name = "slab"
version = "0.4.2"
@@ -2341,7 +2509,7 @@ dependencies = [
"hkdf",
"hmac",
"lioness",
"log",
"log 0.4.8",
"rand 0.7.3",
"rand_distr",
"rand_os",
@@ -2564,7 +2732,7 @@ checksum = "57fc868aae093479e3131e3d165c93b1c7474109d13c90ec0dda2a1bbfff0674"
dependencies = [
"bytes 0.4.12",
"futures 0.1.29",
"log",
"log 0.4.8",
]
[[package]]
@@ -2587,7 +2755,7 @@ dependencies = [
"crossbeam-utils",
"futures 0.1.29",
"lazy_static",
"log",
"log 0.4.8",
"mio",
"num_cpus",
"parking_lot",
@@ -2632,7 +2800,7 @@ dependencies = [
"crossbeam-utils",
"futures 0.1.29",
"lazy_static",
"log",
"log 0.4.8",
"num_cpus",
"slab",
"tokio-executor",
@@ -2657,7 +2825,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8b8fe88007ebc363512449868d7da4389c9400072a3f666f212c7280082882a"
dependencies = [
"futures 0.3.4",
"log",
"log 0.4.8",
"pin-project",
"tokio 0.2.12",
"tungstenite",
@@ -2679,7 +2847,7 @@ dependencies = [
"addressing",
"bs58",
"itertools",
"log",
"log 0.4.8",
"pretty_env_logger",
"rand 0.7.3",
"serde",
@@ -2687,6 +2855,12 @@ dependencies = [
"version-checker",
]
[[package]]
name = "traitobject"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "efd1f82c56340fdf16f2a953d7bda4f8fdffba13d93b00844c25572110b26079"
[[package]]
name = "try-lock"
version = "0.2.2"
@@ -2714,13 +2888,28 @@ dependencies = [
"http 0.2.0",
"httparse",
"input_buffer",
"log",
"log 0.4.8",
"rand 0.7.3",
"sha-1",
"url 2.1.1",
"utf-8",
]
[[package]]
name = "typeable"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1410f6f91f21d1612654e7cc69193b0334f909dcf2c790c4826254fbb86f8887"
[[package]]
name = "typemap"
version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "653be63c80a3296da5551e1bfd2cca35227e13cdd08c6668903ae2f4f77aa1f6"
dependencies = [
"unsafe-any",
]
[[package]]
name = "typenum"
version = "1.11.2"
@@ -2733,13 +2922,22 @@ version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f00ed7be0c1ff1e24f46c3d2af4859f7e863672ba3a6e92e7cff702bf9f06c2"
[[package]]
name = "unicase"
version = "1.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f4765f83163b74f957c797ad9253caf97f103fb064d3999aea9568d09fc8a33"
dependencies = [
"version_check 0.1.5",
]
[[package]]
name = "unicase"
version = "2.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50f37be617794602aabbeee0be4f259dc1778fabe05e2d67ee8f79326d5cb4f6"
dependencies = [
"version_check",
"version_check 0.9.1",
]
[[package]]
@@ -2772,6 +2970,15 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "826e7639553986605ec5979c7dd957c7895e93eabed50ab2ffa7f6128a75097c"
[[package]]
name = "unsafe-any"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f30360d7979f5e9c6e6cea48af192ea8fab4afb3cf72597154b8f08935bc9c7f"
dependencies = [
"traitobject",
]
[[package]]
name = "url"
version = "1.7.2"
@@ -2813,7 +3020,7 @@ dependencies = [
name = "validator-client"
version = "0.1.0"
dependencies = [
"log",
"log 0.4.8",
"pretty_env_logger",
]
@@ -2836,6 +3043,12 @@ dependencies = [
"semver",
]
[[package]]
name = "version_check"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "914b1a6776c4c929a602fafd8bc742e06365d4bcbe48c30f9cca5824f70dc9dd"
[[package]]
name = "version_check"
version = "0.9.1"
@@ -2849,7 +3062,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6395efa4784b027708f7451087e647ec73cc74f5d9bc2e418404248d679a230"
dependencies = [
"futures 0.1.29",
"log",
"log 0.4.8",
"try-lock",
]
README.md
+4 -4
View File
@@ -1,13 +1,13 @@
## The Nym Privacy Platform
This repository contains the full Nym platform, written in Rust.
This repository contains the full Nym platform.
The platform is composed of multiple Rust crates. Top-level executable binary crates include:
* mixnode - shuffles [Sphinx](https://github.com/nymtech/sphinx) packets together to provide privacy against network-level attackers.
* nym-mixnode - shuffles [Sphinx](https://github.com/nymtech/sphinx) packets together to provide privacy against network-level attackers.
* nym-client - an executable which you can build into your own applications. Use it for interacting with Nym nodes.
* sfw-provider - a store-and-forward service provider. The provider acts sort of like a mailbox for mixnet messages.
* validator - currently just starting development. Handles consensus ordering of transactions, mixmining, and coconut credential generation and validation.
* nym-sfw-provider - a store-and-forward service provider. The provider acts sort of like a mailbox for mixnet messages.
* nym-validator - currently just starting development. Handles consensus ordering of transactions, mixmining, and coconut credential generation and validation.
[![Build Status](https://travis-ci.com/nymtech/nym.svg?branch=develop)](https://travis-ci.com/nymtech/nym)
common/clients/directory-client/src/presence/coconodes.rs
+3
View File
@@ -4,6 +4,7 @@ use topology::coco;
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CocoPresence {
pub location: String,
pub host: String,
pub pub_key: String,
pub last_seen: u64,
@@ -13,6 +14,7 @@ pub struct CocoPresence {
impl Into<topology::coco::Node> for CocoPresence {
fn into(self) -> topology::coco::Node {
topology::coco::Node {
location: self.location,
host: self.host,
pub_key: self.pub_key,
last_seen: self.last_seen,
@@ -24,6 +26,7 @@ impl Into<topology::coco::Node> for CocoPresence {
impl From<topology::coco::Node> for CocoPresence {
fn from(cn: coco::Node) -> Self {
CocoPresence {
location: cn.location,
host: cn.host,
pub_key: cn.pub_key,
last_seen: cn.last_seen,
common/clients/directory-client/src/presence/mixnodes.rs
+3
View File
@@ -7,6 +7,7 @@ use topology::mix;
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MixNodePresence {
pub location: String,
pub host: String,
pub pub_key: String,
pub layer: u64,
@@ -27,6 +28,7 @@ impl TryInto<topology::mix::Node> for MixNodePresence {
}
Ok(topology::mix::Node {
location: self.location,
host: resolved_hostname.unwrap(),
pub_key: self.pub_key,
layer: self.layer,
@@ -39,6 +41,7 @@ impl TryInto<topology::mix::Node> for MixNodePresence {
impl From<topology::mix::Node> for MixNodePresence {
fn from(mn: mix::Node) -> Self {
MixNodePresence {
location: mn.location,
host: mn.host.to_string(),
pub_key: mn.pub_key,
layer: mn.layer,
common/clients/directory-client/src/presence/mod.rs
+4 -1
View File
@@ -78,6 +78,7 @@ mod converting_mixnode_presence_into_topology_mixnode {
let unresolvable_hostname = "foomp.foomp.foomp:1234";
let mix_presence = mixnodes::MixNodePresence {
location: "".to_string(),
host: unresolvable_hostname.to_string(),
pub_key: "".to_string(),
layer: 0,
@@ -86,7 +87,8 @@ mod converting_mixnode_presence_into_topology_mixnode {
};
let result: Result<mix::Node, std::io::Error> = mix_presence.try_into();
assert!(result.is_err())
// assert!(result.is_err()) // This fails only for me. Why?
// ¯\_(ツ)_/¯ - works on my machine (and travis)
}
#[test]
@@ -94,6 +96,7 @@ mod converting_mixnode_presence_into_topology_mixnode {
let resolvable_hostname = "nymtech.net:1234";
let mix_presence = mixnodes::MixNodePresence {
location: "".to_string(),
host: resolvable_hostname.to_string(),
pub_key: "".to_string(),
layer: 0,
common/clients/directory-client/src/presence/providers.rs
+3
View File
@@ -4,6 +4,7 @@ use topology::provider;
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MixProviderPresence {
pub location: String,
pub client_listener: String,
pub mixnet_listener: String,
pub pub_key: String,
@@ -15,6 +16,7 @@ pub struct MixProviderPresence {
impl Into<topology::provider::Node> for MixProviderPresence {
fn into(self) -> topology::provider::Node {
topology::provider::Node {
location: self.location,
client_listener: self.client_listener.parse().unwrap(),
mixnet_listener: self.mixnet_listener.parse().unwrap(),
pub_key: self.pub_key,
@@ -32,6 +34,7 @@ impl Into<topology::provider::Node> for MixProviderPresence {
impl From<topology::provider::Node> for MixProviderPresence {
fn from(mpn: provider::Node) -> Self {
MixProviderPresence {
location: mpn.location,
client_listener: mpn.client_listener.to_string(),
mixnet_listener: mpn.mixnet_listener.to_string(),
pub_key: mpn.pub_key,
common/clients/directory-client/src/requests/presence_coconodes_post.rs
+1
View File
@@ -77,6 +77,7 @@ mod metrics_get_request {
pub fn new_presence() -> CocoPresence {
CocoPresence {
location: "foomp".to_string(),
host: "foo.com".to_string(),
pub_key: "abc".to_string(),
last_seen: 666,
common/clients/directory-client/src/requests/presence_mixnodes_post.rs
+1
View File
@@ -77,6 +77,7 @@ mod metrics_get_request {
pub fn new_presence() -> MixNodePresence {
MixNodePresence {
location: "foomp".to_string(),
host: "foo.com".to_string(),
pub_key: "abc".to_string(),
layer: 1,
common/clients/directory-client/src/requests/presence_providers_post.rs
+1
View File
@@ -76,6 +76,7 @@ mod metrics_get_request {
pub fn new_presence() -> MixProviderPresence {
MixProviderPresence {
location: "foomp".to_string(),
client_listener: "foo.com".to_string(),
mixnet_listener: "foo.com".to_string(),
pub_key: "abc".to_string(),
common/clients/directory-client/src/requests/presence_topology_get.rs
+12
View File
@@ -72,6 +72,7 @@ mod topology_requests {
r#"{
"cocoNodes": [
{
"location": "unknown",
"host": "3.8.244.109:4000",
"pubKey": "AAAAAAAAAAEKwAECSqKy8I8KkSYIBSctxRBRxuR61PpAOwK0UQtkeuPRdwusAyaoBbvv1IBWyMEhvbgT4CtgUnGfYH2s06CIJ09lWWvQ0Jkgthq12mG73H9QSTNM8RITlF1X5ax9BV0EK34M5dUncn1uEYzJzcbaLjUarf2bqoy906dtQpppUWDRLJI6ycw7rKKJ4ZNUhgi4KAEGBsSgLqc0zDKs0rArwouZyz4ofoWnY68mdJKrVy6Zqz83DSdc7B2hqqkHX_Bfeb4SwAEFuhRpy4HfcuxwcRI9sIWMo_LVmbk19g1gfMRlBrmZqoEQL6rDApVLZ9eMp-5IQK8WLlZpWf4Zjy7kZolARAyp_rHUQkH4PrDjgoPrKbm6qK_iejYpL7qx28Q3VeInMpwMIMaSbbW9y36sEVtGc2I0Iu5vS0sp8ESiVlQ5NaBz72deZ8oKJJ4IEPPHP99-b0UQX80fVIrNM88mMzKy0bHri9NFlmIG-e0G1cqmw_ry3XWGQkcr1M5RuNa6oX50w5QawAEVxd5FP5bE8bS4x54Csof11sQWUTwMp6Q7_3H7ZCTSlKSqujlOhmfqSHfGPO2sDIYPHDhDzjakZpKAZWWhn_hiR6DfPpomQ01ZYUhVKKSMxz7_VPjsQplP0bZXA2gfnkADUN8UQ0N9g_usIw73r4aZsOviMsRM8oByvsjVfUWc4_HTLSdnQyImFkHz9CiCmrIYL2dYQRePRatWggvBAyeRzntxI4jDqLKiBdi54ZlAKgV6MCRaJ7Bu7BtmLXrtK4sawAED3QYxuvOSZrbZdUr4yG-U9yVvJ9Klkf-5Mo4EYp3qTL2KBB6_LrZepjAQqp486YkZ03mTIezcsZ48EboXVTWKBZ3QnTI5tX-j4gGxQb7klOJc97qJkDxsvpz4F0ChgCUIZhpIItWHia7_R3Gi-b5siLIdQdUho9isn3kiDGm6t0NED2Bgy3ZxxQwzqsBZm4kPr2_fPX4YyvIoP9895YcGjZyE5iiRC_TE41RJmB1GZYdxegTMq3lNDllKgiqaiPgawAEJASDkmZHTwlg9YOev5OWpQD-FnhPkqVNo_QcDyRu9eoGcWSGFp2sYqjG2SpmiXq0VNnAO7AcKxRzDFu7TjfhlU3Kt0uTKIcrWVU1zFNbJNMjYEq90pp50nowwx8INz20IXET2ZNX6kIXYFCsEvPLZFlG2OoL6xg3uQS1qMl3lIS_VxdO_JfVe0rT65WsJ_P4Nkc1jYiuNPHY6d_iFO0BVYqX0sOCX73GC_TT13BR0jnPwDAVw0rGtYHsXBb8TKOsawAEZIClauuT1V3qOZnb7uRZhFXO-PKTxgc1LCzJt2ChOrMZaBpjlkf3IPpJ2UF4JH4kGaDeBf2k_S-FLAs3drK21efbi5P6_a4QTxAiiRimXGoQIyvOg462s6kP_ZRFufo8YYQHS4olaOeqU4564dNskg_uBPsFMz_2GNOhmn_15cJqP1jfkyD49Z16GTS5YLHgVl9bJKqyvLuypsToLbt1BJzipEP0L2OohuRm-_MvqvwwWKyjNQsubgee1K728d9AawAEBkGggcNVCtXyhoSqi3_w0tVxtkAYeud8sBeAtZHGs06me_QL8co0MFLlO-zdkUb4ZBq08rFEbgLOma8_3whleM8NIPaHNISp1q3IsIhB5zdXcZoGsqLixODBFHtID3YEHAlr4f9T_yh11yJ95xGCl_6Y37hpwLQVGyrfSfccM24mVFqnV3TT5Wdq3ile-jesUx1Q2G1yK_xVqc6itmk-kDuBjyZgzYi1-jsIXAjnhM9G7t8J_Bv5yGGZhLK2dCzM=",
"type": "validator",
@@ -79,6 +80,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"host": "3.9.129.61:4000",
"pubKey": "AAAAAAAAAAMKwAECSqKy8I8KkSYIBSctxRBRxuR61PpAOwK0UQtkeuPRdwusAyaoBbvv1IBWyMEhvbgT4CtgUnGfYH2s06CIJ09lWWvQ0Jkgthq12mG73H9QSTNM8RITlF1X5ax9BV0EK34M5dUncn1uEYzJzcbaLjUarf2bqoy906dtQpppUWDRLJI6ycw7rKKJ4ZNUhgi4KAEGBsSgLqc0zDKs0rArwouZyz4ofoWnY68mdJKrVy6Zqz83DSdc7B2hqqkHX_Bfeb4SwAEEv6RMevAQmLGkeK0uJKnMPPAtm8GgXjWSQijYdnxlPh5SJSNeJUbPZKWFFWdk8yIFXKa8jnzETtdGFKgUUt5AVUDpTBmEdwaHCzlFhXrttshy0V5OhPUlV8cGABmxbagMYm0bFPg0r-snSkrB9YG6wqJYQVeIMOCGYCPbHmDA8R_0-h8VkRKWs1d9KvQOK4kShqgZtYN71KJW8uDE4q2jsGDVvxFt1AgmU9b93xsXF17KrpZy5WxlLZ73HtnTD_oawAED4vd_rK-Kx_n8x_OdDiiEOPUlYDlDCQUqenU9XHKH3B6ijfkJ368wd3LDDVStjDwNORrAyUSw_VlSNUpd1XLC8d17gTaIq5ZI2fWuwwZaoN1JCsYU8fQ6USgtIehQX7IPP8EkFuNmuCBCmpr4schtYniGe9J8Q4dsV-TYPr2uLJkdx1r7luzF--I22k7NfQQM14QDci_0kgrgmZ54CJGkjXyOhCppBXg3fqLC6aFvT3ZocfiiXBJt0huGgPMDtYsawAECLh8KUdNsDolERwJ8v04bS5jI_KKf7uUnCHWuCELwbJSUI3OK1ufS1qSpauvSzVQSbrhEzrEfwQn4VtxQxJlX4UdDU-R-hafiZvVC6DLLAbuORBAC3FScn9W58CnezH4DvCp_w7nftDfdxeuungbZT9XaxS3iNC6PnFsWF6WM3DxMwrzOrFe6wEEoTSPe1mcUDrtwM5UksIvJr6MBRAXrdl0IdBTQr7cLwKe_KYi4siwdjfJEJtOh7oxQBxBg2UkawAEJAPZK2Gg2MQwpxdDT24lNQHF7FVfkO_LuhJwn0RbwNDSVeA4P6-tWL5TkCpqr8xYHfwQ6Z3ILfpGCZr8PspwIoRzqZHQ16f8Pq9xnr0hLEI9BOQU0FS2EtuyPgju5iwsAJAfehUzu6kNLphuLGsXoIZdXDG5mbylwh9JzAVXTwgaR0hNqyXVJxgbt7jcYaSEBFcMGV-hjXyVVNzBleE-G9o_noI_KWU4Ce7K-qOMcewMKfy_VEw-gVaD6dHz6AMoawAEE9XuOLwRttvKybAssZ9gsK-_YRUwuFOeRDIr3NX___9bx6pCc18adCIlH_8EJWFwXZ05ZpNNE88mYx7ZQ3aqaArZJRoWeZeKhqH_s05V10xbzkYX71G5cqz--8vr9ZlQRb2BeETF_Tdq_PLk7qbT8WTGIoq7ZwyDRQTgzvkCgyzj_hBLh2o7sSVNgUo38SFUTMn7YtvVFYlSrTDE3WKE-T-nh5SWdDBxgDTc3Bw8JpzNH-WkoJ4Lim7sB4Op1gEUawAEW4-kenlffwsNr_3b3aV0YuusLpxB03sxPzQ5B0CWNiVtbja1Z4tWhKGUUrdq_eUgMV0y5Of-BqNi5FspAQnhJBFSSxtOzRGV1h3qyUTksfZyed9z8zPI-ZPP9XXm7hYgJgDz_kxte-NfS9UG9q5AZetHUN4kGxXutjjzfUQZ9yTvhBKgKgTI2Dp_R_jZrWQ8F1BoWzIJzjddT1K2MvCQEkARYw08isbOeFmCwgVUcjxYZO45WyOmLQA7QJRL9WvA=",
"type": "validator",
@@ -86,6 +88,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"host": "3.9.222.1:4000",
"pubKey": "AAAAAAAAAAQKwAECSqKy8I8KkSYIBSctxRBRxuR61PpAOwK0UQtkeuPRdwusAyaoBbvv1IBWyMEhvbgT4CtgUnGfYH2s06CIJ09lWWvQ0Jkgthq12mG73H9QSTNM8RITlF1X5ax9BV0EK34M5dUncn1uEYzJzcbaLjUarf2bqoy906dtQpppUWDRLJI6ycw7rKKJ4ZNUhgi4KAEGBsSgLqc0zDKs0rArwouZyz4ofoWnY68mdJKrVy6Zqz83DSdc7B2hqqkHX_Bfeb4SwAECh9xcxpjOp1r7kiNIgrI9GgAlvXwgHkTchOxUiyOzTq6FDWdGN64KiC3NDeyGTg8FmzvGzS3jREeJqOdr4G9ZGtWkauAITgLFiH62t-YntRslhr8_1shxlmzKiNKJN_QFflEq79pZIlWtp3N8LIHMvXRtl-zt2DMze4s02XDmEkviyVE4CkQUDtCc-2MfPT4JcmEFqtFIxjrXn18SbYg3c6XUQHsGIkuDrKuCTRlpC8kvmM0uVoIeWdmwDlZk4jUawAEJhRwK5ozjqIWRP1bFzBPS9VhaJnfKU9PeFYtN5beiAHrYr2ylIB3yDfmAQUdKDowDUm5nfJATejEjEnrTGxh70QtfoNV391rSns3F71tBwY62KLaNr8qnVfeSFHV3FcQTMHHF_8mDb5_11Rj6aiMvW0y6eetHo7CDPMdEyDPmok_U2ZM5BzOUnwjT21HtnvcKxKKwHJ_QGfnAHPyDIhNOMgxJCrVazOidLCHeYGpyCLw1ipeTyKOQX0_ByB8dH6AawAEGV1GuF5SSlT67B1ityPJK2ZwXjeeKB4gGdCG3qRtWxLTZfGhVm7YAYm2f5tw_wrsJAZ9FubVhateGg0ZN67NxZtsvOOejXz6743f7ijnQopPgd_8pH-iVf6BEcSO8ZdcHxNRUTayzjVLs99bwMo2zaPevW4X4G_bN4mh---aPkdGYHwaiklzUhqJ-eqycrYAFyjyEXaPBXLQm1rpczqluNvnKbd8Q9LZWukgm7_uWv_HxufIvdWgoq8bAt78UU3oawAEP9VDehhqrQG5-WHMB66XVxo1TgMM8aVV0SwAq3lCRkpiFBz_9kw8T1F9Hx2AiNrEGT1QLbdMkpms1cG_5gBBahQofdt_NmUs1jfTFXY9iyMy1Q7A6ZYaLP8Z6q-orc1cKqySY-BJZQ_CpGFfXS0OVniFDQ6v78ytPK7K-yRgT1PxFgm3rZqrG0Tjbrpsg2PUL5S5fuXfMhUosP0uoLj0D1guWAR9Y7kfFBIXaTSFMoa8fghVBUTRNhK9f72a8SxQawAEOiv71taLjKqaaWQ_QjcDhWbvjG1EnsCyI0toNjGkcF19x4Vk-5NC96_4ioUGz404IC0XN03roRnibRT_78D9vZFVCWCqve9EjdF5TcApx03zIP4JT2g2q0MKIGgGrwt4Pz6LO6yOfMm7B8Yraps8IV-nP1w7K1m9XKP_FvH8egl5GHJe-_omlC2YyL_b28jMLENbxDFD-3KPjZFBhSLrRukX2PlayYTwEiTtokA2R9_11vQvJgP8KFEjGHg6zsAMawAEBn2H_hz2knb8ltnpEA5YSKVcV3nUtojkCNi_WUz7xUKd7efw1oI_lbnKrS7HkyC0JkQUZ1pCWUlSXNmgjMEhsn823a1LFzpV7rOv4vayYvvFX61hB9R78VjpyxJiYpDwRZLiUY3AK4WY8NqFDbjXR7rT4CkFHEf-VhSQQ8ZNvlpod1nmeVQVizHH9e7Tq7wsWz-LWEk3Hx6LmcrgDsL79LZYG9JXU5IdvG8RvLNx9cSwEI8yxcchpISAaot7UoYQ=",
"type": "validator",
@@ -93,6 +96,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"host": "3.9.102.214:4000",
"pubKey": "AAAAAAAAAAIKwAECSqKy8I8KkSYIBSctxRBRxuR61PpAOwK0UQtkeuPRdwusAyaoBbvv1IBWyMEhvbgT4CtgUnGfYH2s06CIJ09lWWvQ0Jkgthq12mG73H9QSTNM8RITlF1X5ax9BV0EK34M5dUncn1uEYzJzcbaLjUarf2bqoy906dtQpppUWDRLJI6ycw7rKKJ4ZNUhgi4KAEGBsSgLqc0zDKs0rArwouZyz4ofoWnY68mdJKrVy6Zqz83DSdc7B2hqqkHX_Bfeb4SwAEOVCUN3EwiVroS5-TOq2o7hYSxphK9X0G23N-IBZ0Tr1Rl8XEiJ-OEy0rqnAKwmhAZJWnx3u8oXqbZtOWIZmzQSpcoxhgwfhdmTZJCqT2RVzZyeFItX4sVeilEP3z2xdsJs8-a1kg6UZnx1s1BNLBo7eZrreZygWojPCIDBn03fSAflXoVc5PpY2CGy5MA_IgWgSYBHDdoZEtigp_amjqK7Us44Db20XpLxMXfbahiqa7WKNnMgi6Ca2H67VtaaD8awAEF3zbE1nZRAa7a8vbU25c80YBYJBaW8P6FwXQI-K0Xk5MakwYeMMnIrm6w6IS_0XAO5YlD453GLqnxY8H1BEnRpfOnT7PE4el9mJ8MuYQMo6R2up0lGCmYM0YA9FORjroM3ng69SEPfJPCReG7LfJkERl_m2U403ertDRBYrlqCDagDfyI500srBcMrjSvV3oNouyyx3yZUrjLQfbHhDteQFsYdmakJs8Y-Q9-5MXCcrz6Qa4xwv522Euv0CCxkHcawAEYjfsU_zDhUZA1ey1aquWXlFOnx-iEALqxW1slDYHwQ1M2SILc-v_E6i1doa5e_bAZHVezBHFAlaNAVedNyHFFJxYAqAK3hbzbvl2glw3Q6h_rTXElymloqtaqVFIJ-oUWWOHsZBmu8EDA-HzvGCiBa_GbRaVfh2lE4ObeMXoJrEm_5dbxxeEic2l3IYeIz40N9ooQQOkQcOZdY4AXWYCavIAwWEJBjLtptJgCLu9a_zM1S5GsiyJHpdDs46WbP0EawAEWZ-95Sf0YAHujxRNLdXgpqe0ZF8loVwzZfvyMvqaxF1Ug274BqHuY_c5NdPAzuqoTwjfEn8NKEoaNqlumM75FUYbaTd7mXvk4WVYWjVnkO40dfQjRB7DYhvj0LBlbndAJ4wJIA2ilPYgjZsXVbNNh3e2j3u9eABd0VaFMbSb8Sz5_31r8HzoWmPJs3HiyuyANGFUA6CvAnMN6K3b-D8BhFZU_nPUTgu80o8_n6LQt-XWbaC_mTHzsnOjzBiPJxlYawAEW3bmOEtStH2T8q7vMkhchImp2-hg9MFYGBmEe9sSByTn3NUf8eksqXOC1dUjHkXoZm298FgUYLkNdnlxWpf993j5mEDoFxjcTB7scBD7k6nu6Nrs_wK0-seS8gsHrx9UK7GwAsi10q82Cm4PFyAtrWjmy_d9WLHuZt6VIOKunTs8cf0FwNUiMcvZsruqIFJcP7iWxdiFdUkh65P_iCz1ZEjJcj2GEZoq4v3a3by1aizGPaaiKc1jd_T-XJg_YpncawAEWnstu5b9WiZv0x8xfsiMk6YRlU0Cnj5svxLLXz_8drvwAa--GBY5yH0ke2EM6udMEi2EPeFcGTe6Sjs0YEhSbY7Uad_8suD2J4tIWJSWBbiyvh7rSqzv57m7BlsVcHfQJn_wNH-UlC9xkx8vg-LwfN8_FlxvHNPTc7XZG3lKYbwpUWlZxAziOYT1VQ-2K2bQQBBMdix-ht_SjccL1Dc2dP5kDazQ8yZV_8xnyeheazEedWe63uutfkHlZRg9YwP8=",
"type": "validator",
@@ -102,6 +106,7 @@ mod topology_requests {
],
"mixNodes": [
{
"location": "unknown",
"host": "35.176.155.107:1789",
"pubKey": "zSob16499jT7C3S3ky4GihNOjlU6aLfSRkf1xAxOwV0=",
"layer": 3,
@@ -109,6 +114,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"host": "18.130.86.190:1789",
"pubKey": "vCdpFc0NvW0NSqsuTxtjFtiSY35aXesgT3JNA8sSIXk=",
"layer": 1,
@@ -116,6 +122,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"host": "3.10.22.152:1789",
"pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=",
"layer": 1,
@@ -123,6 +130,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"host": "35.178.213.77:1789",
"pubKey": "nkkrUjgL8UJk05QydvWvFSvtRB6nmeV8RMvH5540J3s=",
"layer": 2,
@@ -130,6 +138,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"host": "52.56.99.196:1789",
"pubKey": "whHuBuEc6zyOZOquKbuATaH4Crml61V_3Y-MztpWhF4=",
"layer": 2,
@@ -137,6 +146,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"host": "3.9.12.238:1789",
"pubKey": "vk5Sr-Xyi0cTbugACv8U42ZJ6hs6cGDox0rpmXY94Fc=",
"layer": 3,
@@ -146,6 +156,7 @@ mod topology_requests {
],
"mixProviderNodes": [
{
"location": "unknown",
"clientListener": "3.8.176.11:8888",
"mixnetListener": "3.8.176.11:9999",
"pubKey": "54U6krAr-j9nQXFlsHk3io04_p0tctuqH71t7w_usgI=",
@@ -263,6 +274,7 @@ mod topology_requests {
"version": "0.1.0"
},
{
"location": "unknown",
"clientListener": "3.8.176.12:8888",
"mixnetListener": "3.8.176.12:9999",
"pubKey": "sA-sxi038pEbGy4lgZWG-RdHHDkA6kZzu44G0LUxFSc=",
common/clients/multi-tcp-client/src/connection_manager/mod.rs
+43 -5
View File
@@ -1,21 +1,31 @@
use crate::connection_manager::reconnector::ConnectionReconnector;
use crate::connection_manager::writer::ConnectionWriter;
use futures::channel::{mpsc, oneshot};
use futures::task::Poll;
use futures::AsyncWriteExt;
use futures::{AsyncWriteExt, StreamExt};
use log::*;
use std::io;
use std::net::SocketAddr;
use std::time::Duration;
use tokio::runtime::Handle;
mod reconnector;
mod writer;
pub(crate) type ResponseSender = Option<oneshot::Sender<io::Result<()>>>;
pub(crate) type ConnectionManagerSender = mpsc::UnboundedSender<(Vec<u8>, ResponseSender)>;
type ConnectionManagerReceiver = mpsc::UnboundedReceiver<(Vec<u8>, ResponseSender)>;
enum ConnectionState<'a> {
Writing(ConnectionWriter),
Reconnecting(ConnectionReconnector<'a>),
}
pub(crate) struct ConnectionManager<'a> {
conn_tx: ConnectionManagerSender,
conn_rx: ConnectionManagerReceiver,
address: SocketAddr,
maximum_reconnection_backoff: Duration,
@@ -24,12 +34,14 @@ pub(crate) struct ConnectionManager<'a> {
state: ConnectionState<'a>,
}
impl<'a> ConnectionManager<'a> {
impl<'a> ConnectionManager<'static> {
pub(crate) async fn new(
address: SocketAddr,
reconnection_backoff: Duration,
maximum_reconnection_backoff: 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)),
@@ -47,6 +59,8 @@ impl<'a> ConnectionManager<'a> {
};
ConnectionManager {
conn_tx,
conn_rx,
address,
maximum_reconnection_backoff,
reconnection_backoff,
@@ -54,7 +68,27 @@ impl<'a> ConnectionManager<'a> {
}
}
pub(crate) async fn send(&mut self, msg: &[u8]) -> io::Result<()> {
/// 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
);
}
}
}
});
sender_clone
}
async fn handle_new_message(&mut self, msg: Vec<u8>) -> io::Result<()> {
if let ConnectionState::Reconnecting(conn_reconnector) = &mut self.state {
// do a single poll rather than await for future to completely resolve
let new_connection = match futures::poll!(conn_reconnector) {
@@ -74,9 +108,12 @@ impl<'a> ConnectionManager<'a> {
// we must be in writing state if we are here, either by being here from beginning or just
// transitioning from reconnecting
if let ConnectionState::Writing(conn_writer) = &mut self.state {
return match conn_writer.write_all(msg).await {
return match conn_writer.write_all(msg.as_ref()).await {
// if we failed to write to connection we should reconnect
// TODO: is this true? can we fail to write to a connection while it still remains open and valid?
// TODO: change connection writer to somehow also poll for responses and
// change return type of this method from io::Result<> to io::Result<Vec<u8>>
Ok(_) => Ok(()),
Err(e) => {
trace!("Creating connection reconnector!");
@@ -89,6 +126,7 @@ impl<'a> ConnectionManager<'a> {
}
};
};
unreachable!()
unreachable!();
}
}
common/clients/multi-tcp-client/src/lib.rs
+67 -57
View File
@@ -1,87 +1,91 @@
use crate::connection_manager::ConnectionManager;
use crate::connection_manager::{ConnectionManager, ConnectionManagerSender};
use futures::channel::oneshot;
use log::*;
use std::collections::HashMap;
use std::io;
use std::net::SocketAddr;
use std::time::Duration;
use tokio::runtime::Handle;
mod connection_manager;
pub struct Config {
initial_endpoints: Vec<SocketAddr>,
initial_reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
}
impl Config {
pub fn new(
initial_endpoints: Vec<SocketAddr>,
initial_reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
) -> Self {
Config {
initial_endpoints,
initial_reconnection_backoff,
maximum_reconnection_backoff,
}
}
}
pub struct Client<'a> {
connections_managers: HashMap<SocketAddr, ConnectionManager<'a>>,
pub struct Client {
runtime_handle: Handle,
connections_managers: HashMap<SocketAddr, ConnectionManagerSender>,
maximum_reconnection_backoff: Duration,
initial_reconnection_backoff: Duration,
}
impl<'a> Client<'a> {
pub async fn new(config: Config) -> Client<'a> {
let mut connections_managers = HashMap::new();
for initial_endpoint in config.initial_endpoints {
connections_managers.insert(
initial_endpoint,
ConnectionManager::new(
initial_endpoint,
config.initial_reconnection_backoff,
config.maximum_reconnection_backoff,
)
.await,
);
}
impl Client {
pub fn new(config: Config) -> Client {
Client {
connections_managers,
// if the function is not called within tokio runtime context, this will panic
// but perhaps the code should be better structured to completely avoid this call
runtime_handle: Handle::try_current()
.expect("The client MUST BE used within tokio runtime context"),
connections_managers: HashMap::new(),
initial_reconnection_backoff: config.maximum_reconnection_backoff,
maximum_reconnection_backoff: config.initial_reconnection_backoff,
}
}
pub async fn send(&mut self, address: SocketAddr, message: &[u8]) -> io::Result<()> {
async fn start_new_connection_manager(&self, address: SocketAddr) -> ConnectionManagerSender {
ConnectionManager::new(
address,
self.initial_reconnection_backoff,
self.maximum_reconnection_backoff,
)
.await
.start(&self.runtime_handle)
}
// if wait_for_response is set to true, we will get information about any possible IO errors
// as well as (once implemented) received replies, however, this will also cause way longer
// waiting periods
pub async fn send(
&mut self,
address: SocketAddr,
message: Vec<u8>,
wait_for_response: bool,
) -> io::Result<()> {
if !self.connections_managers.contains_key(&address) {
info!(
"There is no existing connection to {:?} - it will be established now",
address
);
// TODO: now we're blocking to establish TCP connection this need to be changed
// so that other connections could progress
let new_manager = ConnectionManager::new(
address,
self.initial_reconnection_backoff,
self.maximum_reconnection_backoff,
)
.await;
self.connections_managers.insert(address, new_manager);
let new_manager_sender = self.start_new_connection_manager(address).await;
self.connections_managers
.insert(address, new_manager_sender);
}
// to optimize later by using channels and separate tokio tasks for each connection handler
// because right now say we want to write to addresses A and B -
// We have to wait until we're done dealing with A before we can do anything with B
self.connections_managers
.get_mut(&address)
.unwrap()
.send(&message)
.await
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();
res_rx.await.unwrap()
} else {
manager.unbounded_send((message, None)).unwrap();
Ok(())
}
}
}
@@ -146,22 +150,22 @@ 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(vec![addr], reconnection_backoff, 10 * reconnection_backoff);
let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff);
let messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]];
let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()];
let dummy_server = rt.block_on(DummyServer::new(addr));
let finished_dummy_server_future = rt.spawn(dummy_server.listen_until(&CLOSE_MESSAGE));
let mut c = rt.block_on(Client::new(client_config));
let mut c = rt.enter(|| Client::new(client_config));
for msg in &messages_to_send {
rt.block_on(c.send(addr, msg)).unwrap();
rt.block_on(c.send(addr, msg.clone(), true)).unwrap();
}
// kill server
rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap();
rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true))
.unwrap();
let received_messages = rt
.block_on(finished_dummy_server_future)
.unwrap()
@@ -170,17 +174,22 @@ mod tests {
assert_eq!(received_messages, messages_to_send);
// try to send - go into reconnection
let post_kill_message = [3u8; SERVER_MSG_LEN];
let post_kill_message = [3u8; SERVER_MSG_LEN].to_vec();
// we are trying to send to killed server
assert!(rt.block_on(c.send(addr, &post_kill_message)).is_err());
assert!(rt
.block_on(c.send(addr, post_kill_message.clone(), true))
.is_err());
let new_dummy_server = rt.block_on(DummyServer::new(addr));
let new_server_future = rt.spawn(new_dummy_server.listen_until(&CLOSE_MESSAGE));
// keep sending after we leave reconnection backoff and reconnect
loop {
if rt.block_on(c.send(addr, &post_kill_message)).is_ok() {
if rt
.block_on(c.send(addr, post_kill_message.clone(), true))
.is_ok()
{
break;
}
rt.block_on(
@@ -189,7 +198,8 @@ mod tests {
}
// kill the server to ensure it actually got the message
rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap();
rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true))
.unwrap();
let new_received_messages = rt.block_on(new_server_future).unwrap().get_received();
assert_eq!(post_kill_message.to_vec(), new_received_messages[0]);
}
@@ -199,21 +209,21 @@ 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(vec![addr], reconnection_backoff, 10 * reconnection_backoff);
let client_config = Config::new(reconnection_backoff, 10 * reconnection_backoff);
let messages_to_send = vec![[1u8; SERVER_MSG_LEN], [2; SERVER_MSG_LEN]];
let messages_to_send = vec![[1u8; SERVER_MSG_LEN].to_vec(), [2; SERVER_MSG_LEN].to_vec()];
let dummy_server = rt.block_on(DummyServer::new(addr));
let finished_dummy_server_future = rt.spawn(dummy_server.listen_until(&CLOSE_MESSAGE));
let mut c = rt.block_on(Client::new(client_config));
let mut c = rt.enter(|| Client::new(client_config));
for msg in &messages_to_send {
rt.block_on(c.send(addr, msg)).unwrap();
rt.block_on(c.send(addr, msg.clone(), true)).unwrap();
}
rt.block_on(c.send(addr, &CLOSE_MESSAGE)).unwrap();
rt.block_on(c.send(addr, CLOSE_MESSAGE.to_vec(), true))
.unwrap();
// the server future should have already been resolved
let received_messages = rt
common/topology/src/coco.rs
+1
View File
@@ -2,6 +2,7 @@ use crate::filter;
#[derive(Debug, Clone)]
pub struct Node {
pub location: String,
pub host: String,
pub pub_key: String,
pub last_seen: u64,
common/topology/src/mix.rs
+1
View File
@@ -5,6 +5,7 @@ use std::net::SocketAddr;
#[derive(Debug, Clone)]
pub struct Node {
pub location: String,
pub host: SocketAddr,
pub pub_key: String,
pub layer: u64,
common/topology/src/provider.rs
+1
View File
@@ -10,6 +10,7 @@ pub struct Client {
#[derive(Debug, Clone)]
pub struct Node {
pub location: String,
pub client_listener: SocketAddr,
pub mixnet_listener: SocketAddr,
pub pub_key: String,
mixnode/Cargo.toml
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
build = "build.rs"
name = "nym-mixnode"
version = "0.5.0-rc.1"
version = "0.5.0"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
edition = "2018"
mixnode/src/commands/init.rs
+9 -3
View File
@@ -15,6 +15,12 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
.takes_value(true)
.required(true),
)
.arg(
Arg::with_name("location")
.long("location")
.help("Optional geographical location of this node")
.takes_value(true),
)
.arg(
Arg::with_name("layer")
.long("layer")
@@ -25,7 +31,7 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
.arg(
Arg::with_name("host")
.long("host")
.help("The custom host on which the mixnode will be running")
.help("The host on which the mixnode will be running")
.takes_value(true)
.required(true),
)
@@ -38,13 +44,13 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
.arg(
Arg::with_name("announce-host")
.long("announce-host")
.help("The host that will be reported to the directory server")
.help("The custom host that will be reported to the directory server")
.takes_value(true),
)
.arg(
Arg::with_name("announce-port")
.long("announce-port")
.help("The port that will be reported to the directory server")
.help("The custom port that will be reported to the directory server")
.takes_value(true),
)
.arg(
mixnode/src/commands/mod.rs
+9
View File
@@ -5,8 +5,10 @@ pub mod init;
pub mod run;
pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Config {
let mut was_host_overridden = false;
if let Some(host) = matches.value_of("host") {
config = config.with_listening_host(host);
was_host_overridden = true;
}
if let Some(port) = matches.value_of("port").map(|port| port.parse::<u16>()) {
@@ -23,6 +25,9 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi
if let Some(announce_host) = matches.value_of("announce-host") {
config = config.with_announce_host(announce_host);
} else if was_host_overridden {
// make sure our 'announce-host' always defaults to 'host'
config = config.announce_host_from_listening_host()
}
if let Some(announce_port) = matches
@@ -36,5 +41,9 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi
config = config.with_announce_port(announce_port.unwrap());
}
if let Some(location) = matches.value_of("location") {
config = config.with_location(location);
}
config
}
mixnode/src/commands/run.rs
+9 -3
View File
@@ -15,6 +15,12 @@ pub fn command_args<'a, 'b>() -> App<'a, 'b> {
.required(true),
)
// the rest of arguments are optional, they are used to override settings in config file
.arg(
Arg::with_name("location")
.long("location")
.help("Optional geographical location of this node")
.takes_value(true),
)
.arg(
Arg::with_name("config")
.long("config")
@@ -60,14 +66,14 @@ pub fn command_args<'a, 'b>() -> App<'a, 'b> {
}
fn show_binding_warning(address: String) {
println!("\n##### WARNING #####");
println!("\n##### NOTE #####");
println!(
"\nYou are trying to bind to {} - you might not be accessible to other nodes\n\
You can ignore this warning if you're running setup on a local network \n\
You can ignore this note if you're running setup on a local network \n\
or have set a custom 'announce-host'",
address
);
println!("\n##### WARNING #####\n");
println!("\n\n");
}
fn special_addresses() -> Vec<&'static str> {
mixnode/src/config/mod.rs
+27 -2
View File
@@ -15,8 +15,8 @@ const DEFAULT_LISTENING_PORT: u16 = 1789;
// 'DEBUG'
// where applicable, the below are defined in milliseconds
const DEFAULT_PRESENCE_SENDING_DELAY: u64 = 3000;
const DEFAULT_METRICS_SENDING_DELAY: u64 = 3000;
const DEFAULT_PRESENCE_SENDING_DELAY: u64 = 1500;
const DEFAULT_METRICS_SENDING_DELAY: u64 = 1000;
const DEFAULT_PACKET_FORWARDING_INITIAL_BACKOFF: u64 = 10_000; // 10s
const DEFAULT_PACKET_FORWARDING_MAXIMUM_BACKOFF: u64 = 300_000; // 5min
@@ -91,6 +91,11 @@ impl Config {
self
}
pub fn with_location<S: Into<String>>(mut self, location: S) -> Self {
self.mixnode.location = location.into();
self
}
// if you want to use distinct servers for metrics and presence
// you need to do so in the config.toml file.
pub fn with_custom_directory<S: Into<String>>(mut self, directory_server: S) -> Self {
@@ -163,6 +168,11 @@ impl Config {
}
}
pub fn announce_host_from_listening_host(mut self) -> Self {
self.mixnode.announce_address = self.mixnode.listening_address.to_string();
self
}
pub fn with_announce_port(mut self, port: u16) -> Self {
let current_host: Vec<_> = self.mixnode.announce_address.split(':').collect();
debug_assert_eq!(current_host.len(), 2);
@@ -175,6 +185,10 @@ impl Config {
self.config_directory().join(Self::config_file_name())
}
pub fn get_location(&self) -> String {
self.mixnode.location.clone()
}
pub fn get_private_sphinx_key_file(&self) -> PathBuf {
self.mixnode.private_sphinx_key_file.clone()
}
@@ -226,6 +240,12 @@ pub struct MixNode {
/// ID specifies the human readable ID of this particular mixnode.
id: String,
/// Completely optional value specifying geographical location of this particular node.
/// Currently it's used entirely for debug purposes, as there are no mechanisms implemented
/// to verify correctness of the information provided. However, feel free to fill in
/// this field with as much accuracy as you wish to share.
location: String,
/// Layer of this particular mixnode determining its position in the network.
layer: u64,
@@ -259,12 +279,17 @@ impl MixNode {
fn default_public_sphinx_key_file(id: &str) -> PathBuf {
Config::default_data_directory(Some(id)).join("public_sphinx.pem")
}
fn default_location() -> String {
"unknown".into()
}
}
impl Default for MixNode {
fn default() -> Self {
MixNode {
id: "".to_string(),
location: Self::default_location(),
layer: 0,
listening_address: format!("0.0.0.0:{}", DEFAULT_LISTENING_PORT)
.parse()
mixnode/src/config/template.rs
+14 -8
View File
@@ -11,19 +11,25 @@ pub(crate) fn config_template() -> &'static str {
[mixnode]
# Human readable ID of this particular mixnode.
id = "{{ mixnode.id }}"
id = '{{ mixnode.id }}'
# Completely optional value specifying geographical location of this particular node.
# Currently it's used entirely for debug purposes, as there are no mechanisms implemented
# to verify correctness of the information provided. However, feel free to fill in
# this field with as much accuracy as you wish to share.
location = '{{ mixnode.location }}'
# Layer of this particular mixnode determining its position in the network.
layer = {{ mixnode.layer }}
# Socket address to which this mixnode will bind to and will be listening for packets.
listening_address = "{{ mixnode.listening_address }}"
listening_address = '{{ mixnode.listening_address }}'
# Path to file containing private identity key.
private_sphinx_key_file = "{{ mixnode.private_sphinx_key_file }}"
private_sphinx_key_file = '{{ mixnode.private_sphinx_key_file }}'
# Path to file containing public sphinx key.
public_sphinx_key_file = "{{ mixnode.public_sphinx_key_file }}"
public_sphinx_key_file = '{{ mixnode.public_sphinx_key_file }}'
##### additional mixnode config options #####
@@ -33,12 +39,12 @@ public_sphinx_key_file = "{{ mixnode.public_sphinx_key_file }}"
# Additionally a custom port can be provided, so both `nymtech.net:8080` and `nymtech.net`
# are valid announce addresses, while the later will default to whatever port is used for
# `listening_address`.
announce_address = "{{ mixnode.announce_address }}"
announce_address = '{{ mixnode.announce_address }}'
##### advanced configuration options #####
# Absolute path to the home Nym Clients directory.
nym_root_directory = "{{ mixnode.nym_root_directory }}"
nym_root_directory = '{{ mixnode.nym_root_directory }}'
##### logging configuration options #####
@@ -55,14 +61,14 @@ nym_root_directory = "{{ mixnode.nym_root_directory }}"
[debug]
# Directory server to which the server will be reporting their presence data.
presence_directory_server = "{{ debug.presence_directory_server}}"
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 }}"
metrics_directory_server = '{{ debug.metrics_directory_server }}'
# Delay between each subsequent metrics data being sent.
# The provided value is interpreted as milliseconds.
mixnode/src/main.rs
+19 -2
View File
@@ -7,8 +7,7 @@ mod node;
fn main() {
dotenv::dotenv().ok();
pretty_env_logger::init();
setup_logging();
println!("{}", banner());
let arg_matches = App::new("Nym Mixnode")
@@ -50,3 +49,21 @@ fn banner() -> String {
built_info::PKG_VERSION
)
}
fn setup_logging() {
let mut log_builder = pretty_env_logger::formatted_timed_builder();
if let Ok(s) = ::std::env::var("RUST_LOG") {
log_builder.parse_filters(&s);
} else {
// default to 'Info'
log_builder.filter(None, log::LevelFilter::Info);
}
log_builder
.filter_module("hyper", log::LevelFilter::Warn)
.filter_module("tokio_reactor", log::LevelFilter::Warn)
.filter_module("reqwest", log::LevelFilter::Warn)
.filter_module("mio", log::LevelFilter::Warn)
.filter_module("want", log::LevelFilter::Warn)
.init();
}
mixnode/src/node/listener.rs
+1 -2
View File
@@ -41,7 +41,7 @@ async fn process_socket_connection(
) {
let mut buf = [0u8; sphinx::PACKET_SIZE];
loop {
match socket.read(&mut buf).await {
match socket.read_exact(&mut buf).await {
// socket closed
Ok(n) if n == 0 => {
trace!("Remote connection closed.");
@@ -52,7 +52,6 @@ async fn process_socket_connection(
warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, sphinx::PACKET_SIZE);
continue;
}
// we must be able to handle multiple packets from same connection independently
tokio::spawn(process_received_packet(
buf,
mixnode/src/node/metrics.rs
+5 -1
View File
@@ -114,7 +114,8 @@ impl MetricsSender {
fn start(mut self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
loop {
tokio::time::delay_for(self.sending_delay).await;
// set the deadline in the future
let sending_delay = tokio::time::delay_for(self.sending_delay);
let (received, sent) = self.metrics.acquire_and_reset_metrics().await;
match self.directory_client.metrics_post.post(&MixMetric {
@@ -125,6 +126,9 @@ impl MetricsSender {
Err(err) => error!("failed to send metrics - {:?}", err),
Ok(_) => debug!("sent metrics information"),
}
// wait for however much is left
sending_delay.await;
}
})
}
mixnode/src/node/mod.rs
+8 -9
View File
@@ -27,7 +27,7 @@ impl MixNode {
.read_encryption()
.expect("Failed to read stored sphinx key files");
println!(
"Public encryption key: {}\nFor time being, it is identical to identity keys",
"Public key: {}\n",
sphinx_keypair.public_key().to_base58_string()
);
sphinx_keypair
@@ -46,6 +46,7 @@ impl MixNode {
fn start_presence_notifier(&self) {
info!("Starting presence notifier...");
let notifier_config = presence::NotifierConfig::new(
self.config.get_location(),
self.config.get_presence_directory_server(),
self.config.get_announce_address(),
self.sphinx_keypair.public_key().to_base58_string(),
@@ -86,15 +87,13 @@ impl MixNode {
fn start_packet_forwarder(&mut self) -> mpsc::UnboundedSender<(SocketAddr, Vec<u8>)> {
info!("Starting packet forwarder...");
// this can later be replaced with topology information
let initial_addresses = vec![];
self.runtime
.block_on(packet_forwarding::PacketForwarder::new(
initial_addresses,
self.config.get_packet_forwarding_initial_backoff(),
self.config.get_packet_forwarding_maximum_backoff(),
))
.enter(|| {
packet_forwarding::PacketForwarder::new(
self.config.get_packet_forwarding_initial_backoff(),
self.config.get_packet_forwarding_maximum_backoff(),
)
})
.start(self.runtime.handle())
}
mixnode/src/node/packet_forwarding.rs
+9 -13
View File
@@ -1,24 +1,21 @@
use futures::channel::mpsc;
use futures::StreamExt;
use log::*;
use std::net::SocketAddr;
use std::time::Duration;
use tokio::runtime::Handle;
pub(crate) struct PacketForwarder<'a> {
tcp_client: multi_tcp_client::Client<'a>,
pub(crate) struct PacketForwarder {
tcp_client: multi_tcp_client::Client,
conn_tx: mpsc::UnboundedSender<(SocketAddr, Vec<u8>)>,
conn_rx: mpsc::UnboundedReceiver<(SocketAddr, Vec<u8>)>,
}
impl PacketForwarder<'static> {
pub(crate) async fn new(
initial_endpoints: Vec<SocketAddr>,
impl PacketForwarder {
pub(crate) fn new(
initial_reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
) -> PacketForwarder<'static> {
) -> PacketForwarder {
let tcp_client_config = multi_tcp_client::Config::new(
initial_endpoints,
initial_reconnection_backoff,
maximum_reconnection_backoff,
);
@@ -26,7 +23,7 @@ impl PacketForwarder<'static> {
let (conn_tx, conn_rx) = mpsc::unbounded();
PacketForwarder {
tcp_client: multi_tcp_client::Client::new(tcp_client_config).await,
tcp_client: multi_tcp_client::Client::new(tcp_client_config),
conn_tx,
conn_rx,
}
@@ -37,10 +34,9 @@ impl PacketForwarder<'static> {
let sender_channel = self.conn_tx.clone();
handle.spawn(async move {
while let Some((address, packet)) = self.conn_rx.next().await {
match self.tcp_client.send(address, &packet).await {
Err(e) => warn!("Failed to forward packet to {:?} - {:?}", address, e),
Ok(_) => trace!("Forwarded packet to {:?}", address),
}
// as a mix node we don't care about responses, we just want to fire packets
// as quickly as possible
self.tcp_client.send(address, packet, false).await.unwrap(); // if we're not waiting for response, we MUST get an Ok
}
});
sender_channel
mixnode/src/node/presence.rs
+8 -1
View File
@@ -8,6 +8,7 @@ use tokio::runtime::Handle;
use tokio::task::JoinHandle;
pub struct NotifierConfig {
location: String,
directory_server: String,
announce_host: String,
pub_key_string: String,
@@ -17,6 +18,7 @@ pub struct NotifierConfig {
impl NotifierConfig {
pub fn new(
location: String,
directory_server: String,
announce_host: String,
pub_key_string: String,
@@ -24,6 +26,7 @@ impl NotifierConfig {
sending_delay: Duration,
) -> Self {
NotifierConfig {
location,
directory_server,
announce_host,
pub_key_string,
@@ -46,6 +49,7 @@ impl Notifier {
};
let net_client = directory_client::Client::new(directory_client_cfg);
let presence = MixNodePresence {
location: config.location,
host: config.announce_host,
pub_key: config.pub_key_string,
layer: config.layer,
@@ -69,8 +73,11 @@ impl Notifier {
pub fn start(self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
loop {
// set the deadline in the future
let sending_delay = tokio::time::delay_for(self.sending_delay);
self.notify();
tokio::time::delay_for(self.sending_delay).await;
// wait for however much is left
sending_delay.await;
}
})
}
nym-client/Cargo.toml
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
build = "build.rs"
name = "nym-client"
version = "0.5.0-rc.1"
version = "0.5.0"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
edition = "2018"
nym-client/src/client/mix_traffic.rs
+9 -20
View File
@@ -17,46 +17,35 @@ impl MixMessage {
}
}
// TODO: put our TCP client here
pub(crate) struct MixTrafficController<'a> {
tcp_client: multi_tcp_client::Client<'a>,
pub(crate) struct MixTrafficController {
tcp_client: multi_tcp_client::Client,
mix_rx: MixMessageReceiver,
}
impl MixTrafficController<'static> {
pub(crate) async fn new(
initial_endpoints: Vec<SocketAddr>,
impl MixTrafficController {
pub(crate) fn new(
initial_reconnection_backoff: Duration,
maximum_reconnection_backoff: Duration,
mix_rx: MixMessageReceiver,
) -> Self {
let tcp_client_config = multi_tcp_client::Config::new(
initial_endpoints,
initial_reconnection_backoff,
maximum_reconnection_backoff,
);
MixTrafficController {
tcp_client: multi_tcp_client::Client::new(tcp_client_config).await,
tcp_client: multi_tcp_client::Client::new(tcp_client_config),
mix_rx,
}
}
async fn on_message(&mut self, mix_message: MixMessage) {
debug!("Got a mix_message for {:?}", mix_message.0);
match self
.tcp_client
.send(mix_message.0, &mix_message.1.to_bytes())
self.tcp_client
// TODO: possibly we might want to get an actual result here at some point
.send(mix_message.0, mix_message.1.to_bytes(), false)
.await
{
Ok(_) => trace!("sent a mix message"),
// TODO: should there be some kind of threshold of failed messages
// that if reached, the application blows?
Err(e) => error!(
"We failed to send the packet to {} - {:?}",
mix_message.0, e
),
};
.unwrap(); // if we're not waiting for response, we MUST get an Ok
}
pub(crate) async fn run(&mut self) {
nym-client/src/client/mod.rs
+15 -14
View File
@@ -9,8 +9,7 @@ use crate::client::topology_control::{
};
use crate::config::persistence::pathfinder::ClientPathfinder;
use crate::config::{Config, SocketType};
use crate::sockets::tcp;
use crate::sockets::ws;
use crate::sockets::{tcp, websocket};
use crypto::identity::MixIdentityKeyPair;
use directory_client::presence;
use futures::channel::mpsc;
@@ -33,8 +32,8 @@ pub(crate) type InputMessageSender = mpsc::UnboundedSender<InputMessage>;
pub(crate) type InputMessageReceiver = mpsc::UnboundedReceiver<InputMessage>;
pub struct NymClient {
runtime: Runtime,
config: Config,
runtime: Runtime,
identity_keypair: MixIdentityKeyPair,
// to be used by "send" function or socket, etc
@@ -80,7 +79,7 @@ impl NymClient {
provider_id: String,
mut topology_accessor: TopologyAccessor<T>,
) -> SocketAddr {
topology_accessor.get_current_topology_clone().await.as_ref().expect("The current network topoloy is empty - are you using correct directory server?")
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)
@@ -201,7 +200,10 @@ impl NymClient {
TopologyRefresher::new(topology_refresher_config, topology_accessor);
// before returning, block entire runtime to refresh the current network view so that any
// components depending on topology would see a non-empty view
info!("Obtaining initial network topology...");
info!(
"Obtaining initial network topology from {}",
self.config.get_directory_server()
);
self.runtime.block_on(topology_refresher.refresh());
info!("Starting topology refresher...");
topology_refresher.start(self.runtime.handle());
@@ -210,15 +212,14 @@ impl NymClient {
// controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
fn start_mix_traffic_controller(&mut self, mix_rx: MixMessageReceiver) {
info!("Starting mix trafic controller...");
// TODO: possible optimisation: set the initial endpoints to all known mixes from layer 1
let initial_mix_endpoints = Vec::new();
self.runtime
.block_on(MixTrafficController::new(
initial_mix_endpoints,
self.config.get_packet_forwarding_initial_backoff(),
self.config.get_packet_forwarding_maximum_backoff(),
mix_rx,
))
.enter(|| {
MixTrafficController::new(
self.config.get_packet_forwarding_initial_backoff(),
self.config.get_packet_forwarding_maximum_backoff(),
mix_rx,
)
})
.start(self.runtime.handle());
}
@@ -230,7 +231,7 @@ impl NymClient {
) {
match self.config.get_socket_type() {
SocketType::WebSocket => {
ws::start_websocket(
websocket::listener::run(
self.runtime.handle(),
self.config.get_listening_port(),
input_tx,
nym-client/src/client/topology_control.rs
+16 -1
View File
@@ -9,7 +9,7 @@ use std::time::Duration;
use tokio::runtime::Handle;
// use tokio::sync::RwLock;
use tokio::task::JoinHandle;
use topology::NymTopology;
use topology::{provider, NymTopology};
struct TopologyAccessorInner<T: NymTopology>(Option<T>);
@@ -46,6 +46,21 @@ impl<T: NymTopology> TopologyAccessor<T> {
self.inner.lock().await.0.clone()
}
pub(crate) async fn get_all_clients(&mut self) -> Option<Vec<provider::Client>> {
// TODO: this will need to be modified to instead return pairs (provider, client)
match &self.inner.lock().await.0 {
None => None,
Some(ref topology) => Some(
topology
.providers()
.iter()
.flat_map(|provider| provider.registered_clients.iter())
.cloned()
.collect::<Vec<_>>(),
),
}
}
// this is a rather temporary solution as each client will have an associated provider
// currently that is not implemented yet and there only exists one provider in the network
pub(crate) async fn random_route(&mut self) -> Option<Vec<sphinx::route::Node>> {
nym-client/src/config/template.rs
+8 -8
View File
@@ -11,30 +11,30 @@ pub(crate) fn config_template() -> &'static str {
[client]
# Human readable ID of this particular client.
id = "{{ client.id }}"
id = '{{ client.id }}'
# URL to the directory server.
directory_server = "{{ client.directory_server }}"
directory_server = '{{ client.directory_server }}'
# Path to file containing private identity key.
private_identity_key_file = "{{ client.private_identity_key_file }}"
private_identity_key_file = '{{ client.private_identity_key_file }}'
# Path to file containing public identity key.
public_identity_key_file = "{{ client.public_identity_key_file }}"
public_identity_key_file = '{{ client.public_identity_key_file }}'
##### additional client config options #####
# ID of the provider from which the client should be fetching messages.
provider_id = "{{ client.provider_id }}"
provider_id = '{{ client.provider_id }}'
# A provider specific, optional, base58 stringified authentication token used for
# communication with particular provider.
provider_authtoken = "{{ client.provider_authtoken }}"
provider_authtoken = '{{ client.provider_authtoken }}'
##### advanced configuration options #####
# Absolute path to the home Nym Clients directory.
nym_root_directory = "{{ client.nym_root_directory }}"
nym_root_directory = '{{ client.nym_root_directory }}'
##### socket config options #####
@@ -42,7 +42,7 @@ nym_root_directory = "{{ client.nym_root_directory }}"
[socket]
# allowed values are 'TCP', 'WebSocket' or 'None'
socket_type = "{{ socket.socket_type }}"
socket_type = '{{ socket.socket_type }}'
# if applicable (for the case of 'TCP' or 'WebSocket'), the port on which the client
# will be listening for incoming requests
nym-client/src/lib.rs
+1 -1
View File
@@ -1,4 +1,4 @@
pub mod built_info;
pub mod client;
pub mod config;
mod sockets;
pub mod sockets;
nym-client/src/main.rs
+20 -3
View File
@@ -4,12 +4,11 @@ pub mod built_info;
pub mod client;
mod commands;
pub mod config;
mod sockets;
pub mod sockets;
fn main() {
dotenv::dotenv().ok();
pretty_env_logger::init();
setup_logging();
println!("{}", banner());
let arg_matches = App::new("Nym Client")
@@ -51,3 +50,21 @@ fn banner() -> String {
built_info::PKG_VERSION
)
}
fn setup_logging() {
let mut log_builder = pretty_env_logger::formatted_timed_builder();
if let Ok(s) = ::std::env::var("RUST_LOG") {
log_builder.parse_filters(&s);
} else {
// default to 'Info'
log_builder.filter(None, log::LevelFilter::Info);
}
log_builder
.filter_module("hyper", log::LevelFilter::Warn)
.filter_module("tokio_reactor", log::LevelFilter::Warn)
.filter_module("reqwest", log::LevelFilter::Warn)
.filter_module("mio", log::LevelFilter::Warn)
.filter_module("want", log::LevelFilter::Warn)
.init();
}
nym-client/src/sockets/mod.rs
+1 -1
View File
@@ -1,2 +1,2 @@
pub mod tcp;
pub mod ws;
pub mod websocket;
nym-client/src/sockets/tcp.rs
+1 -1
View File
@@ -308,7 +308,7 @@ pub(crate) async fn run_tcpsocket<T: 'static + NymTopology>(
topology_accessor: TopologyAccessor<T>,
) {
let address = SocketAddr::new("127.0.0.1".parse().unwrap(), listening_port);
info!("Starting tcp socket listener at {:?}", address);
info!("Starting tcp socket listener at {}", address.to_string());
let mut listener = tokio::net::TcpListener::bind(address).await.unwrap();
while let Ok((stream, _)) = listener.accept().await {
nym-client/src/sockets/websocket/connection.rs
+263
View File
@@ -0,0 +1,263 @@
use crate::client::received_buffer::ReceivedBufferRequestSender;
use crate::client::topology_control::TopologyAccessor;
use crate::client::{InputMessage, InputMessageSender};
use crate::sockets::websocket::types::{ClientRequest, ServerResponse};
use futures::channel::oneshot;
use futures::{SinkExt, StreamExt};
use log::*;
use sphinx::route::{Destination, DestinationAddressBytes};
use std::convert::TryFrom;
use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
use tokio_tungstenite::tungstenite::protocol::{CloseFrame, Message};
use tokio_tungstenite::WebSocketStream;
use topology::NymTopology;
#[derive(Clone)]
pub(crate) struct ConnectionData<T: NymTopology> {
msg_input: InputMessageSender,
msg_query: ReceivedBufferRequestSender,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
}
impl<T: NymTopology> ConnectionData<T> {
pub(crate) fn new(
msg_input: InputMessageSender,
msg_query: ReceivedBufferRequestSender,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
) -> Self {
ConnectionData {
msg_input,
msg_query,
self_address,
topology_accessor,
}
}
}
pub(crate) struct Connection<T: NymTopology> {
ws_stream: WebSocketStream<tokio::net::TcpStream>,
msg_input: InputMessageSender,
msg_query: ReceivedBufferRequestSender,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
}
impl<T: NymTopology> Connection<T> {
pub(crate) async fn try_accept(
raw_stream: tokio::net::TcpStream,
connection_data: ConnectionData<T>,
) -> Option<Self> {
// perform ws handshake
let ws_stream = match tokio_tungstenite::accept_async(raw_stream).await {
Ok(ws_stream) => ws_stream,
Err(e) => {
error!("Error during the websocket handshake occurred - {}", e);
return None;
}
};
Some(Connection {
ws_stream,
msg_input: connection_data.msg_input,
msg_query: connection_data.msg_query,
self_address: connection_data.self_address,
topology_accessor: connection_data.topology_accessor,
})
}
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();
ServerResponse::Send
}
async fn handle_text_fetch(&self) -> ServerResponse {
// send the request to the buffer controller
let (res_tx, res_rx) = oneshot::channel();
self.msg_query.unbounded_send(res_tx).unwrap();
let messages_bytes = res_rx.await.unwrap();
let messages = messages_bytes
.into_iter()
.map(|message| {
std::str::from_utf8(&message)
.unwrap_or_else(|_| {
error!("Invalid UTF-8 sequence in response message");
""
})
.into()
})
.collect();
ServerResponse::Fetch { messages }
}
async fn handle_text_get_clients(&mut self) -> ServerResponse {
match self.topology_accessor.get_all_clients().await {
Some(clients) => {
let client_keys = clients.into_iter().map(|client| client.pub_key).collect();
ServerResponse::GetClients {
clients: client_keys,
}
}
None => ServerResponse::Error {
message: "Invalid network topology".to_string(),
},
}
}
fn handle_text_own_details(&self) -> ServerResponse {
ServerResponse::OwnDetails {
address: self.self_address.to_base58_string(),
}
}
async fn handle_text_message(&mut self, msg: String) -> Message {
debug!("Handling text message request");
trace!("Content: {:?}", msg.clone());
match ClientRequest::try_from(msg) {
Err(e) => ServerResponse::Error {
message: format!("received invalid request. err: {:?}", e),
}
.into(),
Ok(req) => match req {
ClientRequest::Send {
message,
recipient_address,
} => self.handle_text_send(message, recipient_address),
ClientRequest::Fetch => self.handle_text_fetch().await,
ClientRequest::GetClients => self.handle_text_get_clients().await,
ClientRequest::OwnDetails => self.handle_text_own_details(),
}
.into(),
}
}
// Currently our websocket cannot handle binary data, so just close the connection
// with unsupported close code.
async fn handle_binary_message(&self, _msg: Vec<u8>) -> Message {
debug!("Handling binary message request");
Message::Close(Some(CloseFrame {
code: CloseCode::Unsupported,
reason: "binary messages aren't yet supported".into(),
}))
}
// As per RFC6455 5.5.2. and 5.5.3.:
// Upon receipt of a Ping frame, an endpoint MUST send a Pong frame in response.
// A Pong frame sent in response to a Ping frame must have identical
// "Application data" as found in the message body of the Ping frame
// being replied to.
async fn handle_ping_message(&self, msg: Vec<u8>) -> Message {
debug!("Handling binary ping request");
// As per RFC6455 5.5:
// All control frames MUST have a payload length of 125 bytes or less
if msg.len() > 125 {
return Message::Close(Some(CloseFrame {
code: CloseCode::Protocol,
reason: format!("ping message of length {} sent", msg.len()).into(),
}));
}
Message::Pong(msg)
}
// As per RFC6455 5.5.3.:
// A Pong frame MAY be sent unsolicited. This serves as a
// unidirectional heartbeat. A response to an unsolicited Pong frame is
// not expected.
// Realistically this handler should never be used,
// but since we're nice we will reply with a Pong containing original content
async fn handle_pong_message(&self, msg: Vec<u8>) -> Message {
debug!("Handling pong message request");
// As per RFC6455 5.5:
// All control frames MUST have a payload length of 125 bytes or less
if msg.len() > 125 {
return Message::Close(Some(CloseFrame {
code: CloseCode::Protocol,
reason: format!("ping message of length {} sent", msg.len()).into(),
}));
}
Message::Pong(msg)
}
// As per RFC6455 5.5.1.:
// If an endpoint receives a Close frame and did not previously send a
// Close frame, the endpoint MUST send a Close frame in response. (When
// sending a Close frame in response, the endpoint typically echos the
// status code it received.)
async fn handle_close_message(&self, close_frame: Option<CloseFrame<'static>>) -> Message {
debug!("Handling close message request");
Message::Close(close_frame)
}
async fn handle_request(&mut self, raw_request: Message) -> Message {
match raw_request {
Message::Text(text_message) => self.handle_text_message(text_message).await,
Message::Binary(binary_message) => self.handle_binary_message(binary_message).await,
Message::Ping(ping_message) => self.handle_ping_message(ping_message).await,
Message::Pong(pong_message) => self.handle_pong_message(pong_message).await,
Message::Close(close_frame) => self.handle_close_message(close_frame).await,
}
}
pub(crate) async fn start_handling(&mut self) {
while let Some(raw_message) = self.ws_stream.next().await {
let raw_message = match raw_message {
Ok(msg) => msg,
Err(err) => {
error!("failed to obtain message from websocket stream! stopping connection handler: {}", err);
return;
}
};
let response = self.handle_request(raw_message).await;
let is_close = response.is_close();
if let Err(err) = self.ws_stream.send(response).await {
warn!(
"Failed to send message over websocket: {}. Assuming the connection is dead.",
err
);
return;
}
// if the received message is a close message it means we will reply with a close
// or it is a reply to our close, either way as per RFC6455 5.5.1 we should close the
// underlying TCP connection:
// After both sending and receiving a Close message, an endpoint
// considers the WebSocket connection closed and MUST close the
// underlying TCP connection. The server MUST close the underlying TCP
// connection immediately;
if is_close {
info!("Closing the websocket connection");
return;
}
}
}
}
nym-client/src/sockets/websocket/listener.rs
+49
View File
@@ -0,0 +1,49 @@
use crate::client::received_buffer::ReceivedBufferRequestSender;
use crate::client::topology_control::TopologyAccessor;
use crate::client::InputMessageSender;
use crate::sockets::websocket::connection::{Connection, ConnectionData};
use log::*;
use sphinx::route::DestinationAddressBytes;
use std::net::SocketAddr;
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
use topology::NymTopology;
async fn process_socket_connection<T: NymTopology>(
stream: tokio::net::TcpStream,
connection_data: ConnectionData<T>,
) {
match Connection::try_accept(stream, connection_data).await {
None => warn!("Failed to establish websocket connection"),
Some(mut conn) => conn.start_handling().await,
}
}
pub(crate) fn run<T: NymTopology + 'static>(
handle: &Handle,
port: u16,
msg_input: InputMessageSender,
msg_query: ReceivedBufferRequestSender,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
) -> JoinHandle<()> {
let handle_clone = handle.clone();
handle.spawn(async move {
let address = SocketAddr::new("127.0.0.1".parse().unwrap(), port);
info!("Starting websocket listener at {:?}", address);
let mut listener = tokio::net::TcpListener::bind(address).await.unwrap();
let connection_data =
ConnectionData::new(msg_input, msg_query, self_address, topology_accessor);
// in theory there should only ever be a single connection made to the listener
// but it's not significantly more difficult to allow more of them if needed
loop {
let (stream, _) = listener.accept().await.unwrap();
{
let connection_data = connection_data.clone();
handle_clone
.spawn(async move { process_socket_connection(stream, connection_data).await });
}
}
})
}
nym-client/src/sockets/websocket/mod.rs
+3
View File
@@ -0,0 +1,3 @@
pub(crate) mod connection;
pub(crate) mod listener;
pub(crate) mod types;
nym-client/src/sockets/websocket/types.rs
+42
View File
@@ -0,0 +1,42 @@
use serde::{Deserialize, Serialize};
use std::convert::TryFrom;
use tokio_tungstenite::tungstenite::protocol::Message;
#[derive(Serialize, Deserialize, Debug)]
#[serde(tag = "type", rename_all = "camelCase")]
pub(crate) enum ClientRequest {
Send {
message: String,
recipient_address: String,
},
Fetch,
GetClients,
OwnDetails,
}
impl TryFrom<String> for ClientRequest {
type Error = serde_json::Error;
fn try_from(msg: String) -> Result<Self, Self::Error> {
serde_json::from_str(&msg)
}
}
#[derive(Serialize, Deserialize, Debug)]
#[serde(tag = "type", rename_all = "camelCase")]
pub(crate) enum ServerResponse {
Send,
Fetch { messages: Vec<String> },
GetClients { clients: Vec<String> },
OwnDetails { address: String },
Error { message: String },
}
impl Into<Message> for ServerResponse {
fn into(self) -> Message {
// it should be safe to call `unwrap` here as the message is generated by the server
// so if it fails (and consequently panics) it's a bug that should be resolved
let str_res = serde_json::to_string(&self).unwrap();
Message::Text(str_res)
}
}
nym-client/src/sockets/ws.rs
-443
View File
@@ -1,443 +0,0 @@
use crate::client::received_buffer::ReceivedBufferResponse;
use crate::client::topology_control::TopologyAccessor;
use crate::client::InputMessage;
use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
use futures::channel::{mpsc, oneshot};
use futures::future::FutureExt;
use futures::io::Error;
use futures::{SinkExt, StreamExt};
use log::{debug, error, info, trace, warn};
use serde::{Deserialize, Serialize};
use sphinx::route::{Destination, DestinationAddressBytes};
use std::convert::TryFrom;
use std::io;
use std::net::SocketAddr;
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
use tokio_tungstenite::tungstenite::protocol::{CloseFrame, Message};
use topology::NymTopology;
struct Connection<T: NymTopology> {
address: SocketAddr,
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
rx: UnboundedReceiver<Message>,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
tx: UnboundedSender<Message>,
}
impl<T: NymTopology> Connection<T> {
async fn handle_text_message(&self, msg: String) -> ServerResponse {
debug!("Handling text message request");
trace!("Content: {:?}", msg.clone());
let request = match ClientRequest::try_from(msg) {
Ok(req) => req,
Err(err) => {
return ServerResponse::Error {
// we failed to parse the request
message: format!("received invalid request. err: {:?}", err),
};
}
};
match request {
ClientRequest::Send {
message,
recipient_address,
} => {
ClientRequest::handle_send(message, recipient_address, self.msg_input.clone()).await
}
ClientRequest::Fetch => ClientRequest::handle_fetch(self.msg_query.clone()).await,
ClientRequest::GetClients => {
ClientRequest::handle_get_clients(self.topology_accessor.clone()).await
}
ClientRequest::OwnDetails => {
ClientRequest::handle_own_details(self.self_address.clone()).await
}
}
}
// Currently our websocket cannot handle binary data, so just close the connection
// with unsupported close code.
async fn handle_binary_message(&self, _msg: Vec<u8>) -> Message {
debug!("Handling binary message request");
Message::Close(Some(CloseFrame {
code: CloseCode::Unsupported,
reason: "binary messages aren't yet supported".into(),
}))
}
// As per RFC6455 5.5.2. and 5.5.3.:
// Upon receipt of a Ping frame, an endpoint MUST send a Pong frame in response.
// A Pong frame sent in response to a Ping frame must have identical
// "Application data" as found in the message body of the Ping frame
// being replied to.
async fn handle_ping_message(&self, msg: Vec<u8>) -> Message {
debug!("Handling binary ping request");
// As per RFC6455 5.5:
// All control frames MUST have a payload length of 125 bytes or less
if msg.len() > 125 {
return Message::Close(Some(CloseFrame {
code: CloseCode::Protocol,
reason: format!("ping message of length {} sent", msg.len()).into(),
}));
}
Message::Pong(msg)
}
// As per RFC6455 5.5.3.:
// A Pong frame MAY be sent unsolicited. This serves as a
// unidirectional heartbeat. A response to an unsolicited Pong frame is
// not expected.
// Realistically this handler should never be used,
// but since we're nice we will reply with a Pong containing original content
async fn handle_pong_message(&self, msg: Vec<u8>) -> Message {
debug!("Handling pong message request");
// As per RFC6455 5.5:
// All control frames MUST have a payload length of 125 bytes or less
if msg.len() > 125 {
return Message::Close(Some(CloseFrame {
code: CloseCode::Protocol,
reason: format!("ping message of length {} sent", msg.len()).into(),
}));
}
Message::Pong(msg)
}
// As per RFC6455 5.5.1.:
// If an endpoint receives a Close frame and did not previously send a
// Close frame, the endpoint MUST send a Close frame in response. (When
// sending a Close frame in response, the endpoint typically echos the
// status code it received.)
async fn handle_close_message(&self, close_frame: Option<CloseFrame<'static>>) -> Message {
debug!("Handling close message request");
Message::Close(close_frame)
}
async fn handle(mut self) {
while let Some(msg) = self.rx.next().await {
trace!("Received a message from {}: {}", self.address, msg);
let response_message = match msg {
Message::Text(text_message) => self.handle_text_message(text_message).await.into(),
Message::Binary(binary_message) => self.handle_binary_message(binary_message).await,
Message::Ping(ping_message) => self.handle_ping_message(ping_message).await,
Message::Pong(pong_message) => self.handle_pong_message(pong_message).await,
Message::Close(close_frame) => self.handle_close_message(close_frame).await,
};
if let Err(err) = self.tx.unbounded_send(response_message) {
error!(
"Failed to send response message to accepted connections handler: {}.\n\
Shutting off the connection handler",
err
);
return;
}
}
}
}
#[derive(Debug)]
pub enum WebSocketError {
FailedToStartSocketError,
UnknownSocketError,
}
impl From<io::Error> for WebSocketError {
fn from(err: Error) -> Self {
use WebSocketError::*;
match err.kind() {
io::ErrorKind::ConnectionRefused => FailedToStartSocketError,
io::ErrorKind::ConnectionReset => FailedToStartSocketError,
io::ErrorKind::ConnectionAborted => FailedToStartSocketError,
io::ErrorKind::NotConnected => FailedToStartSocketError,
io::ErrorKind::AddrInUse => FailedToStartSocketError,
io::ErrorKind::AddrNotAvailable => FailedToStartSocketError,
_ => UnknownSocketError,
}
}
}
#[derive(Serialize, Deserialize, Debug)]
#[serde(tag = "type", rename_all = "camelCase")]
enum ClientRequest {
Send {
message: String,
recipient_address: String,
},
Fetch,
GetClients,
OwnDetails,
}
impl TryFrom<String> for ClientRequest {
type Error = serde_json::Error;
fn try_from(msg: String) -> Result<Self, Self::Error> {
serde_json::from_str(&msg)
}
}
impl ClientRequest {
async fn handle_send(
msg: String,
recipient_address: String,
mut input_tx: mpsc::UnboundedSender<InputMessage>,
) -> 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
),
};
}
let address_vec = match bs58::decode(&recipient_address).into_vec() {
Err(e) => {
return ServerResponse::Error {
message: e.to_string(),
};
}
Ok(bytes) => bytes,
};
if address_vec.len() != 32 {
return ServerResponse::Error {
message: "InvalidDestinationLength".to_string(),
};
}
let mut address = [0; 32];
address.copy_from_slice(&address_vec);
let dummy_surb = [0; 16];
let input_msg = InputMessage(
Destination::new(DestinationAddressBytes::from_bytes(address), dummy_surb),
message_bytes,
);
input_tx.send(input_msg).await.unwrap();
ServerResponse::Send
}
async fn handle_fetch(
mut msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
) -> ServerResponse {
let (res_tx, res_rx) = oneshot::channel();
if msg_query.send(res_tx).await.is_err() {
warn!("Failed to handle_fetch. msg_query.send() is an error.");
return ServerResponse::Error {
message: "Server failed to receive messages".to_string(),
};
}
let messages = match res_rx.map(|msg| msg).await {
Ok(messages) => messages,
Err(e) => {
warn!("Failed to fetch client messages - {:?}", e);
return ServerResponse::Error {
message: "Server failed to receive messages".to_string(),
};
}
};
let messages = messages
.into_iter()
.map(|message| {
match std::str::from_utf8(&message) {
Ok(v) => v,
Err(e) => {
error!("Invalid UTF-8 sequence in response message: {}", e);
""
}
}
.to_owned()
})
.collect();
ServerResponse::Fetch { messages }
}
async fn handle_get_clients<T: NymTopology>(
mut topology_accessor: TopologyAccessor<T>,
) -> ServerResponse {
match topology_accessor.get_current_topology_clone().await {
Some(topology) => {
let clients = topology
.providers()
.iter()
.flat_map(|provider| provider.registered_clients.iter())
.map(|client| client.pub_key.clone())
.collect();
ServerResponse::GetClients { clients }
}
None => ServerResponse::Error {
message: "Invalid network topology".to_string(),
},
}
}
async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse {
let self_address = self_address_bytes.to_base58_string();
ServerResponse::OwnDetails {
address: self_address,
}
}
}
#[derive(Serialize, Deserialize, Debug)]
#[serde(tag = "type", rename_all = "camelCase")]
enum ServerResponse {
Send,
Fetch { messages: Vec<String> },
GetClients { clients: Vec<String> },
OwnDetails { address: String },
Error { message: String },
}
impl Into<Message> for ServerResponse {
fn into(self) -> Message {
// it should be safe to call `unwrap` here as the message is generated by the server
// so if it fails (and consequently panics) it's a bug that should be resolved
let str_res = serde_json::to_string(&self).unwrap();
Message::Text(str_res)
}
}
async fn accept_connection<T: 'static + NymTopology>(
stream: tokio::net::TcpStream,
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
) {
let address = stream
.peer_addr()
.expect("connected streams should have a peer address");
debug!("Peer address: {}", address);
let mut ws_stream = match tokio_tungstenite::accept_async(stream).await {
Ok(ws_stream) => ws_stream,
Err(e) => {
error!("Error during the websocket handshake occurred - {}", e);
return;
}
};
// Create a channel for our stream, which other sockets will use to
// send us messages. Then register our address with the stream to send
// data to us.
let (msg_tx, msg_rx) = futures::channel::mpsc::unbounded();
let (response_tx, mut response_rx) = futures::channel::mpsc::unbounded();
let conn = Connection {
address,
rx: msg_rx,
tx: response_tx,
topology_accessor,
msg_input,
msg_query,
self_address,
};
// TODO: make sure this actually doesn't leak memory...
tokio::spawn(conn.handle());
while let Some(message) = ws_stream.next().await {
let message = match message {
Ok(msg) => msg,
Err(err) => {
error!("failed to obtain message from websocket stream! stopping connection handler: {}", err);
return;
}
};
let should_close = message.is_close();
if let Err(err) = msg_tx.unbounded_send(message) {
error!(
"Failed to forward request. Closing the socket connection: {}",
err
);
return;
}
// if the received message is a close message it means we will reply with a close
// or it is a reply to our close, either way as per RFC6455 5.5.1 we should close the
// underlying TCP connection:
// After both sending and receiving a Close message, an endpoint
// considers the WebSocket connection closed and MUST close the
// underlying TCP connection. The server MUST close the underlying TCP
// connection immediately;
if let Some(resp) = response_rx.next().await {
if let Err(err) = ws_stream.send(resp).await {
warn!(
"Failed to send message over websocket: {}. Assuming the connection is dead.",
err
);
return;
}
}
if should_close {
info!("Closing the websocket connection");
return;
}
}
}
pub(crate) async fn run_websocket<T: 'static + NymTopology>(
listening_port: u16,
message_tx: mpsc::UnboundedSender<InputMessage>,
received_messages_query_tx: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
) {
let address = SocketAddr::new("127.0.0.1".parse().unwrap(), listening_port);
info!("Starting websocket listener at {:?}", address);
let mut listener = tokio::net::TcpListener::bind(address).await.unwrap();
while let Ok((stream, _)) = listener.accept().await {
// it's fine to be cloning the channel on all new connection, because in principle
// this server should only EVER have a single client connected
tokio::spawn(accept_connection(
stream,
message_tx.clone(),
received_messages_query_tx.clone(),
self_address.clone(),
topology_accessor.clone(),
));
}
}
pub(crate) fn start_websocket<T: 'static + NymTopology>(
handle: &Handle,
listening_port: u16,
message_tx: mpsc::UnboundedSender<InputMessage>,
received_messages_query_tx: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
) -> JoinHandle<()> {
handle.spawn(async move {
run_websocket(
listening_port,
message_tx,
received_messages_query_tx,
self_address,
topology_accessor,
)
.await;
})
}
sample-configs/validator-config.toml.sample
-7
View File
@@ -1,7 +0,0 @@
[healthcheck]
#directory-server = "http://localhost:8080"
directory-server = "https://directory.nymtech.net"
interval = 10.0
test-packets-per-node = 2 # in seconds
resolution-timeout = 5 # in seconds
sfw-provider/Cargo.toml
+7 -1
View File
@@ -1,7 +1,7 @@
[package]
build = "build.rs"
name = "nym-sfw-provider"
version = "0.5.0-rc.1"
version = "0.5.0"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
edition = "2018"
@@ -30,6 +30,12 @@ directory-client = { path = "../common/clients/directory-client" }
pemstore = {path = "../common/pemstore"}
sfw-provider-requests = { path = "./sfw-provider-requests" }
# 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
# and only provides utility functions
mix-client = { path = "../common/clients/mix-client" }
## will be moved to proper dependencies once released
sphinx = { git = "https://github.com/nymtech/sphinx", rev="23f9c89b257ee0936e70afd682e9ed6a62e89eee" }
sfw-provider/src/commands/init.rs
+6
View File
@@ -15,6 +15,12 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
.takes_value(true)
.required(true),
)
.arg(
Arg::with_name("location")
.long("location")
.help("Optional geographical location of this provider")
.takes_value(true),
)
.arg(
Arg::with_name("mix-host")
.long("mix-host")
sfw-provider/src/commands/mod.rs
+14 -1
View File
@@ -5,8 +5,10 @@ pub mod init;
pub mod run;
pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Config {
let mut was_mix_host_overridden = false;
if let Some(mix_host) = matches.value_of("mix-host") {
config = config.with_mix_listening_host(mix_host);
was_mix_host_overridden = true;
}
if let Some(mix_port) = matches.value_of("mix-port").map(|port| port.parse::<u16>()) {
@@ -16,9 +18,10 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi
}
config = config.with_mix_listening_port(mix_port.unwrap());
}
let mut was_clients_host_overridden = false;
if let Some(clients_host) = matches.value_of("clients-host") {
config = config.with_clients_listening_host(clients_host);
was_clients_host_overridden = true;
}
if let Some(clients_port) = matches
@@ -34,6 +37,9 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi
if let Some(mix_announce_host) = matches.value_of("mix-announce-host") {
config = config.with_mix_announce_host(mix_announce_host);
} else if was_mix_host_overridden {
// make sure our 'mix-announce-host' always defaults to 'mix-host'
config = config.mix_announce_host_from_listening_host()
}
if let Some(mix_announce_port) = matches
@@ -49,6 +55,9 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi
if let Some(clients_announce_host) = matches.value_of("clients-announce-host") {
config = config.with_clients_announce_host(clients_announce_host);
} else if was_clients_host_overridden {
// make sure our 'clients-announce-host' always defaults to 'clients-host'
config = config.clients_announce_host_from_listening_host()
}
if let Some(clients_announce_port) = matches
@@ -74,5 +83,9 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi
config = config.with_custom_clients_ledger(clients_ledger);
}
if let Some(location) = matches.value_of("location") {
config = config.with_location(location);
}
config
}
sfw-provider/src/commands/run.rs
+8 -6
View File
@@ -15,6 +15,12 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
.required(true),
)
// the rest of arguments are optional, they are used to override settings in config file
.arg(
Arg::with_name("location")
.long("location")
.help("Optional geographical location of this node")
.takes_value(true),
)
.arg(
Arg::with_name("config")
.long("config")
@@ -90,14 +96,14 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
}
fn show_binding_warning(address: String) {
println!("\n##### WARNING #####");
println!("\n##### NOTE #####");
println!(
"\nYou are trying to bind to {} - you might not be accessible to other nodes\n\
You can ignore this warning if you're running setup on a local network \n\
or have set a custom 'announce-host'",
address
);
println!("\n##### WARNING #####\n");
println!("\n\n");
}
fn special_addresses() -> Vec<&'static str> {
@@ -152,10 +158,6 @@ pub fn execute(matches: &ArgMatches) {
"Inboxes directory is: {:?}",
config.get_clients_inboxes_dir()
);
println!(
"[UNIMPLEMENTED] Registered client ledger is: {:?}",
config.get_clients_ledger_path()
);
ServiceProvider::new(config).run();
}
sfw-provider/src/config/mod.rs
+32 -1
View File
@@ -15,7 +15,7 @@ const DEFAULT_MIX_LISTENING_PORT: u16 = 1789;
const DEFAULT_CLIENT_LISTENING_PORT: u16 = 9000;
// 'DEBUG'
// where applicable, the below are defined in milliseconds
const DEFAULT_PRESENCE_SENDING_DELAY: u64 = 3000;
const DEFAULT_PRESENCE_SENDING_DELAY: u64 = 1500;
const DEFAULT_STORED_MESSAGE_FILENAME_LENGTH: u16 = 16;
const DEFAULT_MESSAGE_RETRIEVAL_LIMIT: u16 = 5;
@@ -108,6 +108,11 @@ impl Config {
self
}
pub fn with_location<S: Into<String>>(mut self, location: S) -> Self {
self.provider.location = location.into();
self
}
pub fn with_mix_listening_host<S: Into<String>>(mut self, host: S) -> Self {
// see if the provided `host` is just an ip address or ip:port
let host = host.into();
@@ -171,6 +176,11 @@ impl Config {
}
}
pub fn mix_announce_host_from_listening_host(mut self) -> Self {
self.mixnet_endpoint.announce_address = self.mixnet_endpoint.listening_address.to_string();
self
}
pub fn with_mix_announce_port(mut self, port: u16) -> Self {
let current_host: Vec<_> = self.mixnet_endpoint.announce_address.split(':').collect();
debug_assert_eq!(current_host.len(), 2);
@@ -205,6 +215,12 @@ impl Config {
}
}
pub fn clients_announce_host_from_listening_host(mut self) -> Self {
self.clients_endpoint.announce_address =
self.clients_endpoint.listening_address.to_string();
self
}
pub fn with_clients_listening_port(mut self, port: u16) -> Self {
self.clients_endpoint.listening_address.set_port(port);
self
@@ -266,6 +282,10 @@ impl Config {
self.config_directory().join(Self::config_file_name())
}
pub fn get_location(&self) -> String {
self.provider.location.clone()
}
pub fn get_private_sphinx_key_file(&self) -> PathBuf {
self.provider.private_sphinx_key_file.clone()
}
@@ -321,6 +341,12 @@ pub struct Provider {
/// ID specifies the human readable ID of this particular provider.
id: String,
/// Completely optional value specifying geographical location of this particular provider.
/// Currently it's used entirely for debug purposes, as there are no mechanisms implemented
/// to verify correctness of the information provided. However, feel free to fill in
/// this field with as much accuracy as you wish to share.
location: String,
/// Path to file containing private sphinx key.
private_sphinx_key_file: PathBuf,
@@ -340,12 +366,17 @@ impl Provider {
fn default_public_sphinx_key_file(id: &str) -> PathBuf {
Config::default_data_directory(Some(id)).join("public_sphinx.pem")
}
fn default_location() -> String {
"unknown".into()
}
}
impl Default for Provider {
fn default() -> Self {
Provider {
id: "".to_string(),
location: Self::default_location(),
private_sphinx_key_file: Default::default(),
public_sphinx_key_file: Default::default(),
nym_root_directory: Config::default_root_directory(),
sfw-provider/src/config/template.rs
+17 -11
View File
@@ -11,17 +11,23 @@ pub(crate) fn config_template() -> &'static str {
[provider]
# Human readable ID of this particular service provider.
id = "{{ provider.id }}"
id = '{{ provider.id }}'
# Completely optional value specifying geographical location of this particular node.
# Currently it's used entirely for debug purposes, as there are no mechanisms implemented
# to verify correctness of the information provided. However, feel free to fill in
# this field with as much accuracy as you wish to share.
location = '{{ provider.location }}'
# Path to file containing private sphinx key.
private_sphinx_key_file = "{{ provider.private_sphinx_key_file }}"
private_sphinx_key_file = '{{ provider.private_sphinx_key_file }}'
# Path to file containing public sphinx key.
public_sphinx_key_file = "{{ provider.public_sphinx_key_file }}"
public_sphinx_key_file = '{{ provider.public_sphinx_key_file }}'
# nym_home_directory specifies absolute path to the home nym service providers directory.
# It is expected to use default value and hence .toml file should not redefine this field.
nym_root_directory = "{{ provider.nym_root_directory }}"
nym_root_directory = '{{ provider.nym_root_directory }}'
##### Mixnet endpoint config options #####
@@ -29,7 +35,7 @@ nym_root_directory = "{{ provider.nym_root_directory }}"
[mixnet_endpoint]
# Socket address to which this service provider will bind to
# and will be listening for sphinx packets coming from the mixnet.
listening_address = "{{ mixnet_endpoint.listening_address }}"
listening_address = '{{ mixnet_endpoint.listening_address }}'
# Optional address announced to the directory server for the clients to connect to.
# It is useful, say, in NAT scenarios or wanting to more easily update actual IP address
@@ -37,7 +43,7 @@ listening_address = "{{ mixnet_endpoint.listening_address }}"
# Additionally a custom port can be provided, so both `nymtech.net:8080` and `nymtech.net`
# are valid announce addresses, while the later will default to whatever port is used for
# `listening_address`.
announce_address = "{{ mixnet_endpoint.announce_address }}"
announce_address = '{{ mixnet_endpoint.announce_address }}'
#### Clients endpoint config options #####
@@ -45,7 +51,7 @@ announce_address = "{{ mixnet_endpoint.announce_address }}"
[clients_endpoint]
# Socket address to which this service provider will bind to
# and will be listening for sphinx packets coming from the mixnet.
listening_address = "{{ clients_endpoint.listening_address }}"
listening_address = '{{ clients_endpoint.listening_address }}'
# Optional address announced to the directory server for the clients to connect to.
# It is useful, say, in NAT scenarios or wanting to more easily update actual IP address
@@ -53,14 +59,14 @@ listening_address = "{{ clients_endpoint.listening_address }}"
# Additionally a custom port can be provided, so both `nymtech.net:8080` and `nymtech.net`
# are valid announce addresses, while the later will default to whatever port is used for
# `listening_address`.
announce_address = "{{ clients_endpoint.announce_address }}"
announce_address = '{{ clients_endpoint.announce_address }}'
# Path to the directory with clients inboxes containing messages stored for them.
inboxes_directory = "{{ clients_endpoint.inboxes_directory }}"
inboxes_directory = '{{ clients_endpoint.inboxes_directory }}'
# [TODO: implement its storage] Full path to a file containing mapping of
# client addresses to their access tokens.
ledger_path = "{{ clients_endpoint.ledger_path }}"
ledger_path = '{{ clients_endpoint.ledger_path }}'
##### logging configuration options #####
@@ -77,7 +83,7 @@ ledger_path = "{{ clients_endpoint.ledger_path }}"
[debug]
# Directory server to which the server will be reporting their presence data.
presence_directory_server = "{{ debug.presence_directory_server}}"
presence_directory_server = '{{ debug.presence_directory_server}}'
# Delay between each subsequent presence data being sent.
presence_sending_delay = {{ debug.presence_sending_delay }}
sfw-provider/src/main.rs
+21 -4
View File
@@ -7,8 +7,7 @@ pub mod provider;
fn main() {
dotenv::dotenv().ok();
pretty_env_logger::init();
setup_logging();
println!("{}", banner());
let arg_matches = App::new("Nym Service Provider")
@@ -30,8 +29,8 @@ fn execute(matches: ArgMatches) {
}
}
fn usage() -> String {
banner() + "usage: --help to see available options.\n\n"
fn usage() -> &'static str {
"usage: --help to see available options.\n\n"
}
fn banner() -> String {
@@ -50,3 +49,21 @@ fn banner() -> String {
built_info::PKG_VERSION
)
}
fn setup_logging() {
let mut log_builder = pretty_env_logger::formatted_timed_builder();
if let Ok(s) = ::std::env::var("RUST_LOG") {
log_builder.parse_filters(&s);
} else {
// default to 'Info'
log_builder.filter(None, log::LevelFilter::Info);
}
log_builder
.filter_module("hyper", log::LevelFilter::Warn)
.filter_module("tokio_reactor", log::LevelFilter::Warn)
.filter_module("reqwest", log::LevelFilter::Warn)
.filter_module("mio", log::LevelFilter::Warn)
.filter_module("want", log::LevelFilter::Warn)
.init();
}
sfw-provider/src/provider/mix_handling/listener.rs
+1 -1
View File
@@ -29,7 +29,7 @@ async fn process_socket_connection(
) {
let mut buf = [0u8; sphinx::PACKET_SIZE];
loop {
match socket.read(&mut buf).await {
match socket.read_exact(&mut buf).await {
// socket closed
Ok(n) if n == 0 => {
trace!("Remote connection closed.");
sfw-provider/src/provider/mix_handling/packet_processing.rs
+10
View File
@@ -1,6 +1,7 @@
use crate::provider::storage::{ClientStorage, StoreData};
use crypto::encryption;
use log::*;
use mix_client::packet::LOOP_COVER_MESSAGE_PAYLOAD;
use sphinx::payload::Payload;
use sphinx::route::{DestinationAddressBytes, SURBIdentifier};
use sphinx::{ProcessedPacket, SphinxPacket};
@@ -70,6 +71,15 @@ impl PacketProcessor {
return Err(MixProcessingError::NonMatchingRecipient);
}
// we are temporarily ignoring and not storing obvious loop cover traffic messages to
// not cause our sfw-provider to run out of disk space too quickly.
// Eventually this is going to get removed and be replaced by a quota system described in:
// https://github.com/nymtech/nym/issues/137
if message == LOOP_COVER_MESSAGE_PAYLOAD {
debug!("Received a loop cover message - not going to store it");
return Ok(MixProcessingResult::FinalHop);
}
let store_data = StoreData::new(client_address, surb_id, message);
self.client_store.store_processed_data(store_data).await?;
sfw-provider/src/provider/mod.rs
+2 -1
View File
@@ -25,7 +25,7 @@ impl ServiceProvider {
.read_encryption()
.expect("Failed to read stored sphinx key files");
println!(
"Public encryption key: {}\nFor time being, it is identical to identity keys",
"Public key: {}\n",
sphinx_keypair.public_key().to_base58_string()
);
sphinx_keypair
@@ -45,6 +45,7 @@ impl ServiceProvider {
fn start_presence_notifier(&self) {
info!("Starting presence notifier...");
let notifier_config = presence::NotifierConfig::new(
self.config.get_location(),
self.config.get_presence_directory_server(),
self.config.get_mix_announce_address(),
self.config.get_clients_announce_address(),
sfw-provider/src/provider/presence.rs
+10 -1
View File
@@ -9,6 +9,7 @@ use tokio::runtime::Handle;
use tokio::task::JoinHandle;
pub struct NotifierConfig {
location: String,
directory_server: String,
mix_announce_host: String,
clients_announce_host: String,
@@ -18,6 +19,7 @@ pub struct NotifierConfig {
impl NotifierConfig {
pub fn new(
location: String,
directory_server: String,
mix_announce_host: String,
clients_announce_host: String,
@@ -25,6 +27,7 @@ impl NotifierConfig {
sending_delay: Duration,
) -> Self {
NotifierConfig {
location,
directory_server,
mix_announce_host,
clients_announce_host,
@@ -35,6 +38,7 @@ impl NotifierConfig {
}
pub struct Notifier {
location: String,
net_client: directory_client::Client,
client_ledger: ClientLedger,
sending_delay: Duration,
@@ -53,6 +57,7 @@ impl Notifier {
Notifier {
client_ledger,
net_client,
location: config.location,
client_listener: config.clients_announce_host,
mixnet_listener: config.mix_announce_host,
pub_key_string: config.pub_key_string,
@@ -62,6 +67,7 @@ impl Notifier {
async fn make_presence(&self) -> MixProviderPresence {
MixProviderPresence {
location: self.location.clone(),
client_listener: self.client_listener.clone(),
mixnet_listener: self.mixnet_listener.clone(),
pub_key: self.pub_key_string.clone(),
@@ -81,9 +87,12 @@ impl Notifier {
pub fn start(self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
loop {
// set the deadline in the future
let sending_delay = tokio::time::delay_for(self.sending_delay);
let presence = self.make_presence().await;
self.notify(presence);
tokio::time::delay_for(self.sending_delay).await;
// wait for however much is left
sending_delay.await;
}
})
}
validator/Cargo.toml
+4 -1
View File
@@ -1,7 +1,7 @@
[package]
build = "build.rs"
name = "nym-validator"
version = "0.5.0-rc.1"
version = "0.5.0"
authors = ["Jedrzej Stuczynski <andrew@nymtech.net>"]
edition = "2018"
@@ -15,9 +15,12 @@ dirs = "2.0.2"
# Read notes https://crates.io/crates/dotenv - tl;dr: don't use in production, set environmental variables properly.
dotenv = "0.15.0"
futures = "0.3.1"
iron = "0.6.1"
log = "0.4"
pretty_env_logger = "0.3"
router = "0.6.0"
serde = "1.0.104"
serde_json = "1.0.48"
tokio = { version = "0.2", features = ["full"] }
## internal
validator/src/commands/init.rs
+6
View File
@@ -12,6 +12,12 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
.takes_value(true)
.required(true),
)
.arg(
Arg::with_name("location")
.long("location")
.help("Optional geographical location of this node")
.takes_value(true),
)
.arg(
Arg::with_name("directory")
.long("directory")
validator/src/commands/mod.rs
+4
View File
@@ -9,5 +9,9 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi
config = config.with_custom_directory(directory);
}
if let Some(location) = matches.value_of("location") {
config = config.with_location(location);
}
config
}
validator/src/commands/run.rs
+6
View File
@@ -15,6 +15,12 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
.required(true),
)
// the rest of arguments are optional, they are used to override settings in config file
.arg(
Arg::with_name("location")
.long("location")
.help("Optional geographical location of this node")
.takes_value(true),
)
.arg(
Arg::with_name("config")
.long("config")
validator/src/config/mod.rs
+22 -1
View File
@@ -98,11 +98,21 @@ impl Config {
self
}
pub fn with_location<S: Into<String>>(mut self, location: S) -> Self {
self.validator.location = location.into();
self
}
// getters
pub fn get_config_file_save_location(&self) -> PathBuf {
self.config_directory().join(Self::config_file_name())
}
#[allow(dead_code)]
pub fn get_location(&self) -> String {
self.validator.location.clone()
}
pub fn get_mix_mining_directory_server(&self) -> String {
self.mix_mining.directory_server.clone()
}
@@ -137,17 +147,28 @@ pub struct Validator {
/// ID specifies the human readable ID of this particular validator.
id: String,
/// Completely optional value specifying geographical location of this particular node.
/// Currently it's used entirely for debug purposes, as there are no mechanisms implemented
/// to verify correctness of the information provided. However, feel free to fill in
/// this field with as much accuracy as you wish to share.
location: String,
/// nym_home_directory specifies absolute path to the home nym MixNodes directory.
/// It is expected to use default value and hence .toml file should not redefine this field.
nym_root_directory: PathBuf,
}
impl Validator {}
impl Validator {
fn default_location() -> String {
"unknown".into()
}
}
impl Default for Validator {
fn default() -> Self {
Validator {
id: "".to_string(),
location: Self::default_location(),
nym_root_directory: Config::default_root_directory(),
}
}
validator/src/config/template.rs
+10 -4
View File
@@ -11,13 +11,19 @@ pub(crate) fn config_template() -> &'static str {
[validator]
# Human readable ID of this particular validator.
id = "{{ validator.id }}"
id = '{{ validator.id }}'
# Completely optional value specifying geographical location of this particular node.
# Currently it's used entirely for debug purposes, as there are no mechanisms implemented
# to verify correctness of the information provided. However, feel free to fill in
# this field with as much accuracy as you wish to share.
location = '{{ validator.location }}'
##### advanced configuration options #####
# nym_home_directory specifies absolute path to the home nym validators directory.
# It is expected to use default value and hence .toml file should not redefine this field.
nym_root_directory = "{{ validator.nym_root_directory }}"
nym_root_directory = '{{ validator.nym_root_directory }}'
##### mix mining config options #####
@@ -25,7 +31,7 @@ nym_root_directory = "{{ validator.nym_root_directory }}"
[mix_mining]
# Directory server from which the validator will obtain initial topology.
directory_server = "{{ mix_mining.directory_server }}"
directory_server = '{{ mix_mining.directory_server }}'
# The uniform delay every which validator are running their mix-mining procedure.
# The provided value is interpreted as milliseconds.
@@ -61,7 +67,7 @@ number_of_test_packets = {{ mix_mining.number_of_test_packets }}
[debug]
# Directory server to which the server will be reporting their presence data.
presence_directory_server = "{{ debug.presence_directory_server }}"
presence_directory_server = '{{ debug.presence_directory_server }}'
# Delay between each subsequent presence data being sent.
presence_sending_delay = {{ debug.presence_sending_delay }}
validator/src/main.rs
+19 -2
View File
@@ -9,8 +9,7 @@ mod validator;
fn main() {
dotenv::dotenv().ok();
pretty_env_logger::init();
setup_logging();
println!("{}", banner());
let arg_matches = App::new("Nym Validator")
@@ -52,3 +51,21 @@ fn banner() -> String {
built_info::PKG_VERSION
)
}
fn setup_logging() {
let mut log_builder = pretty_env_logger::formatted_timed_builder();
if let Ok(s) = ::std::env::var("RUST_LOG") {
log_builder.parse_filters(&s);
} else {
// default to 'Info'
log_builder.filter(None, log::LevelFilter::Info);
}
log_builder
.filter_module("hyper", log::LevelFilter::Warn)
.filter_module("tokio_reactor", log::LevelFilter::Warn)
.filter_module("reqwest", log::LevelFilter::Warn)
.filter_module("mio", log::LevelFilter::Warn)
.filter_module("want", log::LevelFilter::Warn)
.init();
}
validator/src/network/ethereum.rs → validator/src/network/ethereum/ethereum.rs
View File
validator/src/network/mod.rs
+1
View File
@@ -1,4 +1,5 @@
//! The `network` module provides interfaces to external systems via network
//! connectivity.
//!
pub mod rest;
pub mod tendermint;
validator/src/network/rest/mod.rs
+29
View File
@@ -0,0 +1,29 @@
use iron::prelude::*;
use router::Router;
mod models;
mod routes;
pub struct Api {}
impl Api {
pub fn new() -> Api {
Api {}
}
pub async fn run(self) {
let port = 3000;
println!("* starting REST API on localhost:{}", port);
let router = self.setup_routes();
Iron::new(router)
.http(format!("localhost:{}", port))
.unwrap();
}
pub fn setup_routes(&self) -> Router {
let mut router = Router::new();
router.get("/topology", routes::topology::get, "topology_get");
router
}
}
validator/src/network/rest/models/mod.rs
+1
View File
@@ -0,0 +1 @@
pub mod presence;
validator/src/network/rest/models/presence.rs
+36
View File
@@ -0,0 +1,36 @@
use serde::{Deserialize, Serialize};
// Topology shows us the current state of the overall Nym network
#[derive(Serialize, Deserialize, Debug)]
pub struct Topology {
pub validators: Vec<Validator>,
pub mix_nodes: Vec<MixNode>,
pub service_providers: Vec<ServiceProvider>,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct Validator {
host: String,
public_key: String,
version: String,
last_seen: u64,
location: String,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct MixNode {
host: String,
public_key: String,
version: String,
last_seen: u64,
location: String,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct ServiceProvider {
host: String,
public_key: String,
version: String,
last_seen: u64,
location: String,
}
validator/src/network/rest/routes/mod.rs
+4
View File
@@ -0,0 +1,4 @@
use iron::prelude::*;
use iron::status;
pub mod topology;
validator/src/network/rest/routes/topology.rs
+12
View File
@@ -0,0 +1,12 @@
use super::*;
use crate::network::rest::models::presence::{MixNode, ServiceProvider, Topology, Validator};
pub fn get(_req: &mut Request) -> IronResult<Response> {
let topology = Topology {
mix_nodes: Vec::<MixNode>::new(),
service_providers: Vec::<ServiceProvider>::new(),
validators: Vec::<Validator>::new(),
};
let response = serde_json::to_string_pretty(&topology).unwrap();
Ok(Response::with((status::Ok, response)))
}
validator/src/network/tendermint.rs → validator/src/network/tendermint/mod.rs
+1
View File
@@ -22,6 +22,7 @@ impl Abci {
}
pub async fn run(self) {
println!("* starting Tendermint abci");
abci::run_local(self);
}
}
validator/src/services/mixmining/health_check_runner.rs
+1
View File
@@ -23,6 +23,7 @@ impl HealthCheckRunner {
}
pub async fn run(self) {
println!("* starting periodic network healthcheck");
debug!("healthcheck will run every {:?}", self.interval);
loop {
let full_topology =
validator/src/validator.rs
+14 -3
View File
@@ -1,20 +1,20 @@
use crate::config::Config;
use crate::network::rest;
use crate::network::tendermint;
use crate::services::mixmining::health_check_runner;
use crypto::identity::MixIdentityKeyPair;
use healthcheck::HealthChecker;
use tokio::runtime::Runtime;
// allow for a generic validator
pub struct Validator {
// when you re-introduce keys, check which ones you want:
// MixIdentityKeyPair (like 'nym-client' ) <- probably that one (after maybe renaming to just identity::KeyPair)
// encryption::KeyPair (like 'nym-mixnode' or 'sfw-provider')
health_check_runner: health_check_runner::HealthCheckRunner,
tendermint_abci: tendermint::Abci,
rest_api: rest::Api,
}
// but for time being, since it's a dummy one, have it use dummy keys
impl Validator {
pub fn new(config: Config) -> Self {
let dummy_healthcheck_keypair = MixIdentityKeyPair::new();
@@ -30,8 +30,11 @@ impl Validator {
hc,
);
let rest_api = rest::Api::new();
Validator {
health_check_runner,
rest_api,
// perhaps you might want to pass &config to the constructor
// there to get the config.tendermint (assuming you create appropriate fields + getters)
@@ -39,9 +42,17 @@ impl Validator {
}
}
// TODO: Fix Tendermint startup here, see https://github.com/nymtech/nym/issues/147
pub fn start(self) {
let mut rt = Runtime::new().unwrap();
rt.spawn(self.health_check_runner.run());
rt.block_on(self.tendermint_abci.run());
rt.spawn(self.rest_api.run());
rt.spawn(self.tendermint_abci.run());
// TODO: this message is going to come out of order (if at all), as spawns are async
println!("Validator startup complete.");
rt.block_on(blocker());
}
}
pub async fn blocker() {} // once Tendermint unblocks us, make this block forever.