From bdfe91676c72f538d5a52b3a2a4fdbbaf58da63f Mon Sep 17 00:00:00 2001 From: Dave Hrycyszyn Date: Thu, 13 Aug 2020 11:21:59 +0100 Subject: [PATCH] Feature/client core (#303) * Adding a client-core crate. * Moved config into client_core * Socks5 client now depends on client-core/config * Native client now mostly using client-core internals * Socks client now uses client-core internals. --- Cargo.lock | 27 + Cargo.toml | 1 + clients/client-core/Cargo.toml | 35 ++ clients/client-core/build.rs | 19 + .../mod.rs => client-core/src/built_info.rs} | 3 +- .../src/client/cover_traffic_stream.rs | 6 +- .../src/client/inbound_messages.rs | 10 +- .../src/client/key_manager.rs | 18 +- .../src/client/mix_traffic.rs | 16 +- clients/client-core/src/client/mod.rs | 8 + .../acknowledgement_listener.rs | 0 .../input_message_listener.rs | 0 .../acknowlegement_control/mod.rs | 0 .../retransmission_request_listener.rs | 0 .../sent_notification_listener.rs | 0 .../src/client/real_messages_control/mod.rs | 10 +- .../real_traffic_stream.rs | 0 .../src/client/received_buffer.rs | 16 +- .../src/client/reply_key_storage.rs | 8 +- .../src/client/topology_control.rs | 26 +- .../{native => client-core}/src/config/mod.rs | 0 .../src/config/persistence/key_pathfinder.rs | 0 .../src/config/persistence/mod.rs | 0 .../src/config/template.rs | 0 clients/client-core/src/lib.rs | 3 + clients/native/Cargo.toml | 11 +- clients/native/src/client/inbound_messages.rs | 33 -- clients/native/src/client/key_manager.rs | 168 ------ clients/native/src/client/mix_traffic.rs | 91 ---- clients/native/src/client/mod.rs | 40 +- .../src/client/real_messages_control/mod.rs | 170 ------ .../native/src/client/reply_key_storage.rs | 105 ---- clients/native/src/commands/init.rs | 6 +- clients/native/src/commands/mod.rs | 2 +- clients/native/src/commands/run.rs | 2 +- clients/native/src/lib.rs | 1 - clients/native/src/main.rs | 3 +- clients/native/src/websocket/handler.rs | 4 +- clients/socks5/Cargo.toml | 1 + .../socks5/src/client/cover_traffic_stream.rs | 186 ------- clients/socks5/src/client/mod.rs | 45 +- .../acknowledgement_listener.rs | 92 ---- .../input_message_listener.rs | 205 ------- .../acknowlegement_control/mod.rs | 230 -------- .../retransmission_request_listener.rs | 129 ----- .../sent_notification_listener.rs | 109 ---- .../real_traffic_stream.rs | 245 --------- clients/socks5/src/client/received_buffer.rs | 374 ------------- clients/socks5/src/client/topology_control.rs | 204 ------- clients/socks5/src/commands/init.rs | 6 +- clients/socks5/src/commands/mod.rs | 2 +- clients/socks5/src/commands/run.rs | 2 +- clients/socks5/src/config/mod.rs | 503 ------------------ .../src/config/persistence/key_pathfinder.rs | 76 --- clients/socks5/src/config/template.rs | 102 ---- clients/socks5/src/lib.rs | 1 - clients/socks5/src/main.rs | 1 - clients/socks5/src/socks/client.rs | 4 +- clients/socks5/src/socks/mixnet_responses.rs | 4 +- clients/socks5/src/socks/server.rs | 2 +- 60 files changed, 213 insertions(+), 3152 deletions(-) create mode 100644 clients/client-core/Cargo.toml create mode 100644 clients/client-core/build.rs rename clients/{socks5/src/config/persistence/mod.rs => client-core/src/built_info.rs} (85%) rename clients/{native => client-core}/src/client/cover_traffic_stream.rs (98%) rename clients/{socks5 => client-core}/src/client/inbound_messages.rs (58%) rename clients/{socks5 => client-core}/src/client/key_manager.rs (91%) rename clients/{socks5 => client-core}/src/client/mix_traffic.rs (85%) create mode 100644 clients/client-core/src/client/mod.rs rename clients/{native => client-core}/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs (100%) rename clients/{native => client-core}/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs (100%) rename clients/{native => client-core}/src/client/real_messages_control/acknowlegement_control/mod.rs (100%) rename clients/{native => client-core}/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs (100%) rename clients/{native => client-core}/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs (100%) rename clients/{socks5 => client-core}/src/client/real_messages_control/mod.rs (96%) rename clients/{native => client-core}/src/client/real_messages_control/real_traffic_stream.rs (100%) rename clients/{native => client-core}/src/client/received_buffer.rs (96%) rename clients/{socks5 => client-core}/src/client/reply_key_storage.rs (94%) rename clients/{native => client-core}/src/client/topology_control.rs (89%) rename clients/{native => client-core}/src/config/mod.rs (100%) rename clients/{native => client-core}/src/config/persistence/key_pathfinder.rs (100%) rename clients/{native => client-core}/src/config/persistence/mod.rs (100%) rename clients/{native => client-core}/src/config/template.rs (100%) create mode 100644 clients/client-core/src/lib.rs delete mode 100644 clients/native/src/client/inbound_messages.rs delete mode 100644 clients/native/src/client/key_manager.rs delete mode 100644 clients/native/src/client/mix_traffic.rs delete mode 100644 clients/native/src/client/real_messages_control/mod.rs delete mode 100644 clients/native/src/client/reply_key_storage.rs delete mode 100644 clients/socks5/src/client/cover_traffic_stream.rs delete mode 100644 clients/socks5/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs delete mode 100644 clients/socks5/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs delete mode 100644 clients/socks5/src/client/real_messages_control/acknowlegement_control/mod.rs delete mode 100644 clients/socks5/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs delete mode 100644 clients/socks5/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs delete mode 100644 clients/socks5/src/client/real_messages_control/real_traffic_stream.rs delete mode 100644 clients/socks5/src/client/received_buffer.rs delete mode 100644 clients/socks5/src/client/topology_control.rs delete mode 100644 clients/socks5/src/config/mod.rs delete mode 100644 clients/socks5/src/config/persistence/key_pathfinder.rs delete mode 100644 clients/socks5/src/config/template.rs diff --git a/Cargo.lock b/Cargo.lock index a217020fac..83d0e3050e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -423,6 +423,31 @@ dependencies = [ "cc", ] +[[package]] +name = "client-core" +version = "0.8.0-dev" +dependencies = [ + "built", + "config", + "crypto", + "directory-client", + "dirs", + "dotenv", + "futures 0.3.4", + "gateway-client", + "gateway-requests", + "log 0.4.8", + "nymsphinx", + "pemstore", + "rand 0.7.3", + "serde", + "sled 0.33.0", + "tempfile", + "tokio 0.2.16", + "topology", + "url 2.1.1", +] + [[package]] name = "cloudabi" version = "0.0.3" @@ -1838,6 +1863,7 @@ dependencies = [ "bs58", "built", "clap", + "client-core", "config", "crypto", "directory-client", @@ -3098,6 +3124,7 @@ dependencies = [ "bs58", "built", "clap", + "client-core", "config", "crypto", "curve25519-dalek", diff --git a/Cargo.toml b/Cargo.toml index 0b7b12a7e5..8dacbdec2e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "clients/native", "clients/socks5", "clients/webassembly", + "clients/client-core", "common/client-libs/directory-client", "common/client-libs/directory-client/models", "common/client-libs/gateway-client", diff --git a/clients/client-core/Cargo.toml b/clients/client-core/Cargo.toml new file mode 100644 index 0000000000..ff12ba02cf --- /dev/null +++ b/clients/client-core/Cargo.toml @@ -0,0 +1,35 @@ +[package] +build = "build.rs" +name = "client-core" +version = "0.8.0-dev" +authors = ["Dave Hrycyszyn "] +edition = "2018" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +dirs = "2.0.2" +dotenv = "0.15.0" +futures = "0.3.1" +log = "0.4" +rand = { version = "0.7.3", features = ["wasm-bindgen"] } +serde = { version = "1.0.104", features = ["derive"] } +sled = "0.33" +tokio = { version = "0.2", features = ["full"] } +url = "2.1" + +# internal +config = { path = "../../common/config" } +crypto = { path = "../../common/crypto" } +directory-client = { path = "../../common/client-libs/directory-client" } +gateway-client = { path = "../../common/client-libs/gateway-client" } +gateway-requests = { path = "../../gateway/gateway-requests" } +nymsphinx = { path = "../../common/nymsphinx" } +pemstore = { path = "../../common/pemstore" } +topology = { path = "../../common/topology" } + +[build-dependencies] +built = "0.3.2" + +[dev-dependencies] +tempfile = "3.1.0" \ No newline at end of file diff --git a/clients/client-core/build.rs b/clients/client-core/build.rs new file mode 100644 index 0000000000..56d753472c --- /dev/null +++ b/clients/client-core/build.rs @@ -0,0 +1,19 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use built; + +fn main() { + built::write_built_file().expect("Failed to acquire build-time information"); +} diff --git a/clients/socks5/src/config/persistence/mod.rs b/clients/client-core/src/built_info.rs similarity index 85% rename from clients/socks5/src/config/persistence/mod.rs rename to clients/client-core/src/built_info.rs index b87c0be15d..38b46d59e6 100644 --- a/clients/socks5/src/config/persistence/mod.rs +++ b/clients/client-core/src/built_info.rs @@ -12,4 +12,5 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub mod key_pathfinder; +// The file has been placed there by the build script. +include!(concat!(env!("OUT_DIR"), "/built.rs")); diff --git a/clients/native/src/client/cover_traffic_stream.rs b/clients/client-core/src/client/cover_traffic_stream.rs similarity index 98% rename from clients/native/src/client/cover_traffic_stream.rs rename to clients/client-core/src/client/cover_traffic_stream.rs index 53adb4a78a..0ba912a54d 100644 --- a/clients/native/src/client/cover_traffic_stream.rs +++ b/clients/client-core/src/client/cover_traffic_stream.rs @@ -28,7 +28,7 @@ use tokio::runtime::Handle; use tokio::task::JoinHandle; use tokio::time; -pub(crate) struct LoopCoverTrafficStream +pub struct LoopCoverTrafficStream where R: CryptoRng + Rng, { @@ -96,7 +96,7 @@ where // obviously when we finally make shared rng that is on 'higher' level, this should become // generic `R` impl LoopCoverTrafficStream { - pub(crate) fn new( + pub fn new( ack_key: Arc, average_ack_delay: time::Duration, average_packet_delay: time::Duration, @@ -178,7 +178,7 @@ impl LoopCoverTrafficStream { } } - pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { + pub fn start(mut self, handle: &Handle) -> JoinHandle<()> { handle.spawn(async move { self.run().await; }) diff --git a/clients/socks5/src/client/inbound_messages.rs b/clients/client-core/src/client/inbound_messages.rs similarity index 58% rename from clients/socks5/src/client/inbound_messages.rs rename to clients/client-core/src/client/inbound_messages.rs index ab4abe4445..079dea98cf 100644 --- a/clients/socks5/src/client/inbound_messages.rs +++ b/clients/client-core/src/client/inbound_messages.rs @@ -2,11 +2,11 @@ use futures::channel::mpsc; use nymsphinx::addressing::clients::Recipient; use nymsphinx::anonymous_replies::ReplySURB; -pub(crate) type InputMessageSender = mpsc::UnboundedSender; -pub(crate) type InputMessageReceiver = mpsc::UnboundedReceiver; +pub type InputMessageSender = mpsc::UnboundedSender; +pub type InputMessageReceiver = mpsc::UnboundedReceiver; #[derive(Debug)] -pub(crate) enum InputMessage { +pub enum InputMessage { Fresh { recipient: Recipient, data: Vec, @@ -19,7 +19,7 @@ pub(crate) enum InputMessage { } impl InputMessage { - pub(crate) fn new_fresh(recipient: Recipient, data: Vec, with_reply_surb: bool) -> Self { + pub fn new_fresh(recipient: Recipient, data: Vec, with_reply_surb: bool) -> Self { InputMessage::Fresh { recipient, data, @@ -27,7 +27,7 @@ impl InputMessage { } } - pub(crate) fn new_reply(reply_surb: ReplySURB, data: Vec) -> Self { + pub fn new_reply(reply_surb: ReplySURB, data: Vec) -> Self { InputMessage::Reply { reply_surb, data } } } diff --git a/clients/socks5/src/client/key_manager.rs b/clients/client-core/src/client/key_manager.rs similarity index 91% rename from clients/socks5/src/client/key_manager.rs rename to clients/client-core/src/client/key_manager.rs index 89cc2829d6..49a553c43d 100644 --- a/clients/socks5/src/client/key_manager.rs +++ b/clients/client-core/src/client/key_manager.rs @@ -28,7 +28,7 @@ use std::sync::Arc; // use the old key after new one was issued. // Remember that Arc has Deref implementation for T -pub(crate) struct KeyManager { +pub struct KeyManager { /// identity key associated with the client instance. identity_keypair: Arc, @@ -56,7 +56,7 @@ impl KeyManager { // I have absolutely no idea why the compiler insists it's unused. The call happens during client::init::execute #[allow(dead_code)] /// Creates new instance of a [`KeyManager`] - pub(crate) fn new(rng: &mut R) -> Self + pub fn new(rng: &mut R) -> Self where R: RngCore + CryptoRng, { @@ -72,12 +72,12 @@ impl KeyManager { // I have absolutely no idea why the compiler insists it's unused. The call happens during client::init::execute #[allow(dead_code)] /// After shared key with the gateway is derived, puts its ownership to this instance of a [`KeyManager`]. - pub(crate) fn insert_gateway_shared_key(&mut self, gateway_shared_key: SharedKeys) { + pub fn insert_gateway_shared_key(&mut self, gateway_shared_key: SharedKeys) { self.gateway_shared_key = Some(Arc::new(gateway_shared_key)) } /// Loads previously stored keys from the disk. - pub(crate) fn load_keys(client_pathfinder: &ClientKeyPathfinder) -> io::Result { + pub fn load_keys(client_pathfinder: &ClientKeyPathfinder) -> io::Result { let identity_keypair: identity::KeyPair = pemstore::load_keypair(&pemstore::KeyPairPath::new( client_pathfinder.private_identity_key().to_owned(), @@ -111,7 +111,7 @@ impl KeyManager { // While perhaps there is no much point in storing the `AckKey` on the disk, // it is done so for the consistency sake so that you wouldn't require an rng instance // during `load_keys` to generate the said key. - pub(crate) fn store_keys(&self, client_pathfinder: &ClientKeyPathfinder) -> io::Result<()> { + pub fn store_keys(&self, client_pathfinder: &ClientKeyPathfinder) -> io::Result<()> { pemstore::store_keypair( self.identity_keypair.as_ref(), &pemstore::KeyPairPath::new( @@ -140,19 +140,19 @@ impl KeyManager { } /// Gets an atomically reference counted pointer to [`identity::KeyPair`]. - pub(crate) fn identity_keypair(&self) -> Arc { + pub fn identity_keypair(&self) -> Arc { Arc::clone(&self.identity_keypair) } /// Gets an atomically reference counted pointer to [`encryption::KeyPair`]. - pub(crate) fn encryption_keypair(&self) -> Arc { + pub fn encryption_keypair(&self) -> Arc { Arc::clone(&self.encryption_keypair) } /// Gets an atomically reference counted pointer to [`SharedKey`]. // since this function is not fully public, it is not expected to be used externally and // hence it's up to us to ensure it's called in correct context - pub(crate) fn gateway_shared_key(&self) -> Arc { + pub fn gateway_shared_key(&self) -> Arc { Arc::clone( &self .gateway_shared_key @@ -162,7 +162,7 @@ impl KeyManager { } /// Gets an atomically reference counted pointer to [`AckKey`]. - pub(crate) fn ack_key(&self) -> Arc { + pub fn ack_key(&self) -> Arc { Arc::clone(&self.ack_key) } } diff --git a/clients/socks5/src/client/mix_traffic.rs b/clients/client-core/src/client/mix_traffic.rs similarity index 85% rename from clients/socks5/src/client/mix_traffic.rs rename to clients/client-core/src/client/mix_traffic.rs index cd5b78fa6c..22dc96c787 100644 --- a/clients/socks5/src/client/mix_traffic.rs +++ b/clients/client-core/src/client/mix_traffic.rs @@ -20,19 +20,19 @@ use nymsphinx::{addressing::nodes::NymNodeRoutingAddress, SphinxPacket}; use tokio::runtime::Handle; use tokio::task::JoinHandle; -pub(crate) struct MixMessage(NymNodeRoutingAddress, SphinxPacket); -pub(crate) type MixMessageSender = mpsc::UnboundedSender; -pub(crate) type MixMessageReceiver = mpsc::UnboundedReceiver; +pub struct MixMessage(NymNodeRoutingAddress, SphinxPacket); +pub type MixMessageSender = mpsc::UnboundedSender; +pub type MixMessageReceiver = mpsc::UnboundedReceiver; impl MixMessage { - pub(crate) fn new(address: NymNodeRoutingAddress, packet: SphinxPacket) -> Self { + pub fn new(address: NymNodeRoutingAddress, packet: SphinxPacket) -> Self { MixMessage(address, packet) } } const MAX_FAILURE_COUNT: usize = 100; -pub(crate) struct MixTrafficController<'a> { +pub struct MixTrafficController<'a> { // TODO: most likely to be replaced by some higher level construct as // later on gateway_client will need to be accessible by other entities gateway_client: GatewayClient<'a, url::Url>, @@ -44,7 +44,7 @@ pub(crate) struct MixTrafficController<'a> { } impl<'a> MixTrafficController<'static> { - pub(crate) fn new( + pub fn new( mix_rx: MixMessageReceiver, gateway_client: GatewayClient<'a, url::Url>, ) -> MixTrafficController<'a> { @@ -77,13 +77,13 @@ impl<'a> MixTrafficController<'static> { } } - pub(crate) async fn run(&mut self) { + pub async fn run(&mut self) { while let Some(mix_message) = self.mix_rx.next().await { self.on_message(mix_message).await; } } - pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { + pub fn start(mut self, handle: &Handle) -> JoinHandle<()> { handle.spawn(async move { self.run().await; }) diff --git a/clients/client-core/src/client/mod.rs b/clients/client-core/src/client/mod.rs new file mode 100644 index 0000000000..510c072073 --- /dev/null +++ b/clients/client-core/src/client/mod.rs @@ -0,0 +1,8 @@ +pub mod cover_traffic_stream; +pub mod inbound_messages; +pub mod key_manager; +pub mod mix_traffic; +pub mod real_messages_control; +pub mod received_buffer; +pub mod reply_key_storage; +pub mod topology_control; diff --git a/clients/native/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs b/clients/client-core/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs similarity index 100% rename from clients/native/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs rename to clients/client-core/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs diff --git a/clients/native/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs b/clients/client-core/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs similarity index 100% rename from clients/native/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs rename to clients/client-core/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs diff --git a/clients/native/src/client/real_messages_control/acknowlegement_control/mod.rs b/clients/client-core/src/client/real_messages_control/acknowlegement_control/mod.rs similarity index 100% rename from clients/native/src/client/real_messages_control/acknowlegement_control/mod.rs rename to clients/client-core/src/client/real_messages_control/acknowlegement_control/mod.rs diff --git a/clients/native/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs b/clients/client-core/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs similarity index 100% rename from clients/native/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs rename to clients/client-core/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs diff --git a/clients/native/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs b/clients/client-core/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs similarity index 100% rename from clients/native/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs rename to clients/client-core/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs diff --git a/clients/socks5/src/client/real_messages_control/mod.rs b/clients/client-core/src/client/real_messages_control/mod.rs similarity index 96% rename from clients/socks5/src/client/real_messages_control/mod.rs rename to clients/client-core/src/client/real_messages_control/mod.rs index 3988475b41..7cefb33636 100644 --- a/clients/socks5/src/client/real_messages_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/mod.rs @@ -39,7 +39,7 @@ use tokio::task::JoinHandle; mod acknowlegement_control; mod real_traffic_stream; -pub(crate) struct Config { +pub struct Config { ack_key: Arc, ack_wait_multiplier: f64, ack_wait_addition: Duration, @@ -50,7 +50,7 @@ pub(crate) struct Config { } impl Config { - pub(crate) fn new( + pub fn new( ack_key: Arc, ack_wait_multiplier: f64, ack_wait_addition: Duration, @@ -71,7 +71,7 @@ impl Config { } } -pub(crate) struct RealMessagesController +pub struct RealMessagesController where R: CryptoRng + Rng, { @@ -82,7 +82,7 @@ where // obviously when we finally make shared rng that is on 'higher' level, this should become // generic `R` impl RealMessagesController { - pub(crate) fn new( + pub fn new( config: Config, ack_receiver: AcknowledgementReceiver, input_receiver: InputMessageReceiver, @@ -161,7 +161,7 @@ impl RealMessagesController { // &Handle is only passed for consistency sake with other client modules, but I think // when we get to refactoring, we should apply gateway approach and make it implicit - pub(super) fn start(mut self, handle: &Handle) -> JoinHandle { + pub fn start(mut self, handle: &Handle) -> JoinHandle { handle.spawn(async move { self.run().await; self diff --git a/clients/native/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs similarity index 100% rename from clients/native/src/client/real_messages_control/real_traffic_stream.rs rename to clients/client-core/src/client/real_messages_control/real_traffic_stream.rs diff --git a/clients/native/src/client/received_buffer.rs b/clients/client-core/src/client/received_buffer.rs similarity index 96% rename from clients/native/src/client/received_buffer.rs rename to clients/client-core/src/client/received_buffer.rs index c3bf2d2bb2..db67cc6592 100644 --- a/clients/native/src/client/received_buffer.rs +++ b/clients/client-core/src/client/received_buffer.rs @@ -31,12 +31,12 @@ use tokio::task::JoinHandle; // Buffer Requests to say "hey, send any reconstructed messages to this channel" // or to say "hey, I'm going offline, don't send anything more to me. Just buffer them instead" -pub(crate) type ReceivedBufferRequestSender = mpsc::UnboundedSender; -pub(crate) type ReceivedBufferRequestReceiver = mpsc::UnboundedReceiver; +pub type ReceivedBufferRequestSender = mpsc::UnboundedSender; +pub type ReceivedBufferRequestReceiver = mpsc::UnboundedReceiver; // The channel set for the above -pub(crate) type ReconstructedMessagesSender = mpsc::UnboundedSender>; -pub(crate) type ReconstructedMessagesReceiver = mpsc::UnboundedReceiver>; +pub type ReconstructedMessagesSender = mpsc::UnboundedSender>; +pub type ReconstructedMessagesReceiver = mpsc::UnboundedReceiver>; struct ReceivedMessagesBufferInner { messages: Vec, @@ -277,7 +277,7 @@ impl ReceivedMessagesBuffer { } } -pub(crate) enum ReceivedBufferMessage { +pub enum ReceivedBufferMessage { // Signals a websocket connection (or a native implementation) was established and we should stop buffering messages, // and instead send them directly to the received channel ReceiverAnnounce(ReconstructedMessagesSender), @@ -342,13 +342,13 @@ impl FragmentedMessageReceiver { } } -pub(crate) struct ReceivedMessagesBufferController { +pub struct ReceivedMessagesBufferController { fragmented_message_receiver: FragmentedMessageReceiver, request_receiver: RequestReceiver, } impl ReceivedMessagesBufferController { - pub(crate) fn new( + pub fn new( local_encryption_keypair: Arc, query_receiver: ReceivedBufferRequestReceiver, mixnet_packet_receiver: MixnetMessageReceiver, @@ -366,7 +366,7 @@ impl ReceivedMessagesBufferController { } } - pub(crate) fn start(self, handle: &Handle) { + pub fn start(self, handle: &Handle) { // TODO: should we do anything with JoinHandle(s) returned by start methods? self.fragmented_message_receiver.start(handle); self.request_receiver.start(handle); diff --git a/clients/socks5/src/client/reply_key_storage.rs b/clients/client-core/src/client/reply_key_storage.rs similarity index 94% rename from clients/socks5/src/client/reply_key_storage.rs rename to clients/client-core/src/client/reply_key_storage.rs index e1dc4a6b9a..5656ad850d 100644 --- a/clients/socks5/src/client/reply_key_storage.rs +++ b/clients/client-core/src/client/reply_key_storage.rs @@ -20,7 +20,7 @@ use nymsphinx::anonymous_replies::{ use std::path::Path; #[derive(Debug)] -pub(crate) enum ReplyKeyStorageError { +pub enum ReplyKeyStorageError { DbReadError(sled::Error), DbWriteError(sled::Error), DbOpenError(sled::Error), @@ -40,7 +40,7 @@ pub struct ReplyKeyStorage { } impl ReplyKeyStorage { - pub(crate) fn load>(path: P) -> Result { + pub fn load>(path: P) -> Result { let db = match sled::open(path) { Err(e) => return Err(ReplyKeyStorageError::DbOpenError(e)), Ok(db) => db, @@ -64,7 +64,7 @@ impl ReplyKeyStorage { } // TOOD: perhaps we could also store some part of original message here too? - pub(crate) fn insert_encryption_key( + pub fn insert_encryption_key( &mut self, encryption_key: SURBEncryptionKey, ) -> Result<(), ReplyKeyStorageError> { @@ -86,7 +86,7 @@ impl ReplyKeyStorage { } // Once we use key once, we do not expect to use it again - pub(crate) fn get_and_remove_encryption_key( + pub fn get_and_remove_encryption_key( &self, key_digest: EncryptionKeyDigest, ) -> Result, ReplyKeyStorageError> { diff --git a/clients/native/src/client/topology_control.rs b/clients/client-core/src/client/topology_control.rs similarity index 89% rename from clients/native/src/client/topology_control.rs rename to clients/client-core/src/client/topology_control.rs index 366c1844f2..b39297bc17 100644 --- a/clients/native/src/client/topology_control.rs +++ b/clients/client-core/src/client/topology_control.rs @@ -29,7 +29,7 @@ use topology::NymTopology; // I'm extremely curious why compiler NEVER complained about lack of Debug here before #[derive(Debug)] -pub(super) struct TopologyAccessorInner(Option); +pub struct TopologyAccessorInner(Option); impl AsRef> for TopologyAccessorInner { fn as_ref(&self) -> &Option { @@ -47,7 +47,7 @@ impl TopologyAccessorInner { } } -pub(super) struct TopologyReadPermit<'a> { +pub struct TopologyReadPermit<'a> { permit: RwLockReadGuard<'a, TopologyAccessorInner>, } @@ -99,7 +99,7 @@ impl<'a> From> for TopologyReadPermit } #[derive(Clone, Debug)] -pub(crate) struct TopologyAccessor { +pub struct TopologyAccessor { // `RwLock` *seems to* be the better approach for this as write access is only requested every // few seconds, while reads are needed every single packet generated. // However, proper benchmarks will be needed to determine if `RwLock` is indeed a better @@ -108,13 +108,13 @@ pub(crate) struct TopologyAccessor { } impl TopologyAccessor { - pub(crate) fn new() -> Self { + pub fn new() -> Self { TopologyAccessor { inner: Arc::new(RwLock::new(TopologyAccessorInner::new())), } } - pub(super) async fn get_read_permit(&self) -> TopologyReadPermit<'_> { + pub async fn get_read_permit(&self) -> TopologyReadPermit<'_> { self.inner.read().await.into() } @@ -124,7 +124,7 @@ impl TopologyAccessor { // only used by the client at startup to get a slightly more reasonable error message // (currently displays as unused because health checker is disabled due to required changes) - pub(crate) async fn is_routable(&self) -> bool { + pub async fn is_routable(&self) -> bool { match &self.inner.read().await.0 { None => false, Some(ref topology) => topology.can_construct_path_through(DEFAULT_NUM_MIX_HOPS), @@ -132,13 +132,13 @@ impl TopologyAccessor { } } -pub(crate) struct TopologyRefresherConfig { +pub struct TopologyRefresherConfig { directory_server: String, refresh_rate: time::Duration, } impl TopologyRefresherConfig { - pub(crate) fn new(directory_server: String, refresh_rate: time::Duration) -> Self { + pub fn new(directory_server: String, refresh_rate: time::Duration) -> Self { TopologyRefresherConfig { directory_server, refresh_rate, @@ -146,14 +146,14 @@ impl TopologyRefresherConfig { } } -pub(crate) struct TopologyRefresher { +pub struct TopologyRefresher { directory_client: directory_client::Client, topology_accessor: TopologyAccessor, refresh_rate: Duration, } impl TopologyRefresher { - pub(crate) fn new_directory_client( + pub fn new_directory_client( cfg: TopologyRefresherConfig, topology_accessor: TopologyAccessor, ) -> Self { @@ -180,7 +180,7 @@ impl TopologyRefresher { } } - pub(crate) async fn refresh(&mut self) { + pub async fn refresh(&mut self) { trace!("Refreshing the topology"); let new_topology = self.get_current_compatible_topology().await; @@ -189,11 +189,11 @@ impl TopologyRefresher { .await; } - pub(crate) async fn is_topology_routable(&self) -> bool { + pub async fn is_topology_routable(&self) -> bool { self.topology_accessor.is_routable().await } - pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { + pub fn start(mut self, handle: &Handle) -> JoinHandle<()> { handle.spawn(async move { loop { tokio::time::delay_for(self.refresh_rate).await; diff --git a/clients/native/src/config/mod.rs b/clients/client-core/src/config/mod.rs similarity index 100% rename from clients/native/src/config/mod.rs rename to clients/client-core/src/config/mod.rs diff --git a/clients/native/src/config/persistence/key_pathfinder.rs b/clients/client-core/src/config/persistence/key_pathfinder.rs similarity index 100% rename from clients/native/src/config/persistence/key_pathfinder.rs rename to clients/client-core/src/config/persistence/key_pathfinder.rs diff --git a/clients/native/src/config/persistence/mod.rs b/clients/client-core/src/config/persistence/mod.rs similarity index 100% rename from clients/native/src/config/persistence/mod.rs rename to clients/client-core/src/config/persistence/mod.rs diff --git a/clients/native/src/config/template.rs b/clients/client-core/src/config/template.rs similarity index 100% rename from clients/native/src/config/template.rs rename to clients/client-core/src/config/template.rs diff --git a/clients/client-core/src/lib.rs b/clients/client-core/src/lib.rs new file mode 100644 index 0000000000..a557c4a3b9 --- /dev/null +++ b/clients/client-core/src/lib.rs @@ -0,0 +1,3 @@ +pub mod built_info; +pub mod client; +pub mod config; diff --git a/clients/native/Cargo.toml b/clients/native/Cargo.toml index 577d5abd02..847474c1f5 100644 --- a/clients/native/Cargo.toml +++ b/clients/native/Cargo.toml @@ -20,7 +20,7 @@ futures = "0.3.1" log = "0.4" pem = "0.7.0" pretty_env_logger = "0.3" -rand = {version = "0.7.3", features = ["wasm-bindgen"]} +rand = { version = "0.7.3", features = ["wasm-bindgen"] } reqwest = "0.9.22" serde = { version = "1.0.104", features = ["derive"] } serde_json = "1.0.44" @@ -30,15 +30,16 @@ tokio-tungstenite = "0.10.1" url = "2.1" ## internal -config = {path = "../../common/config"} -crypto = {path = "../../common/crypto"} +client-core = { path = "../client-core" } +config = { path = "../../common/config" } +crypto = { path = "../../common/crypto" } directory-client = { path = "../../common/client-libs/directory-client" } gateway-client = { path = "../../common/client-libs/gateway-client" } gateway-requests = { path = "../../gateway/gateway-requests" } mixnet-client = { path = "../../common/client-libs/mixnet-client" } nymsphinx = { path = "../../common/nymsphinx" } -pemstore = {path = "../../common/pemstore"} -topology = {path = "../../common/topology" } +pemstore = { path = "../../common/pemstore" } +topology = { path = "../../common/topology" } [build-dependencies] built = "0.3.2" diff --git a/clients/native/src/client/inbound_messages.rs b/clients/native/src/client/inbound_messages.rs deleted file mode 100644 index ab4abe4445..0000000000 --- a/clients/native/src/client/inbound_messages.rs +++ /dev/null @@ -1,33 +0,0 @@ -use futures::channel::mpsc; -use nymsphinx::addressing::clients::Recipient; -use nymsphinx::anonymous_replies::ReplySURB; - -pub(crate) type InputMessageSender = mpsc::UnboundedSender; -pub(crate) type InputMessageReceiver = mpsc::UnboundedReceiver; - -#[derive(Debug)] -pub(crate) enum InputMessage { - Fresh { - recipient: Recipient, - data: Vec, - with_reply_surb: bool, - }, - Reply { - reply_surb: ReplySURB, - data: Vec, - }, -} - -impl InputMessage { - pub(crate) fn new_fresh(recipient: Recipient, data: Vec, with_reply_surb: bool) -> Self { - InputMessage::Fresh { - recipient, - data, - with_reply_surb, - } - } - - pub(crate) fn new_reply(reply_surb: ReplySURB, data: Vec) -> Self { - InputMessage::Reply { reply_surb, data } - } -} diff --git a/clients/native/src/client/key_manager.rs b/clients/native/src/client/key_manager.rs deleted file mode 100644 index 89cc2829d6..0000000000 --- a/clients/native/src/client/key_manager.rs +++ /dev/null @@ -1,168 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::config::persistence::key_pathfinder::ClientKeyPathfinder; -use crypto::asymmetric::{encryption, identity}; -use gateway_requests::registration::handshake::SharedKeys; -use log::*; -use nymsphinx::acknowledgements::AckKey; -use rand::{CryptoRng, RngCore}; -use std::io; -use std::sync::Arc; - -// Note: to support key rotation in the future, all keys will require adding an extra smart pointer, -// most likely an AtomicCell, or if it doesn't work as I think it does, a Mutex. Although I think -// AtomicCell includes a Mutex implicitly if the underlying type does not work atomically. -// And I guess there will need to be some mechanism for a grace period when you can still -// use the old key after new one was issued. - -// Remember that Arc has Deref implementation for T -pub(crate) struct KeyManager { - /// identity key associated with the client instance. - identity_keypair: Arc, - - /// encryption key associated with the client instance. - encryption_keypair: Arc, - - /// shared key derived with the gateway during "registration handshake" - gateway_shared_key: Option>, - - /// key used for producing and processing acknowledgement packets. - ack_key: Arc, -} - -// The expected flow of a KeyManager "lifetime" is as follows: -/* - 1. ::new() is called during client-init - 2. after gateway registration is completed [in init] ::insert_gateway_shared_key() is called - 3. ::store_keys() is called before init finishes execution. - 4. ::load_keys() is called at the beginning of each subsequent client-run - 5. [not implemented] ::rotate_keys() is called periodically during client-run I presume? -*/ - -impl KeyManager { - // this is actually **NOT** dead code - // I have absolutely no idea why the compiler insists it's unused. The call happens during client::init::execute - #[allow(dead_code)] - /// Creates new instance of a [`KeyManager`] - pub(crate) fn new(rng: &mut R) -> Self - where - R: RngCore + CryptoRng, - { - KeyManager { - identity_keypair: Arc::new(identity::KeyPair::new_with_rng(rng)), - encryption_keypair: Arc::new(encryption::KeyPair::new_with_rng(rng)), - gateway_shared_key: None, - ack_key: Arc::new(AckKey::new(rng)), - } - } - - // this is actually **NOT** dead code - // I have absolutely no idea why the compiler insists it's unused. The call happens during client::init::execute - #[allow(dead_code)] - /// After shared key with the gateway is derived, puts its ownership to this instance of a [`KeyManager`]. - pub(crate) fn insert_gateway_shared_key(&mut self, gateway_shared_key: SharedKeys) { - self.gateway_shared_key = Some(Arc::new(gateway_shared_key)) - } - - /// Loads previously stored keys from the disk. - pub(crate) fn load_keys(client_pathfinder: &ClientKeyPathfinder) -> io::Result { - let identity_keypair: identity::KeyPair = - pemstore::load_keypair(&pemstore::KeyPairPath::new( - client_pathfinder.private_identity_key().to_owned(), - client_pathfinder.public_identity_key().to_owned(), - ))?; - let encryption_keypair: encryption::KeyPair = - pemstore::load_keypair(&pemstore::KeyPairPath::new( - client_pathfinder.private_encryption_key().to_owned(), - client_pathfinder.public_encryption_key().to_owned(), - ))?; - - let gateway_shared_key: SharedKeys = - pemstore::load_key(&client_pathfinder.gateway_shared_key().to_owned())?; - - let ack_key: AckKey = pemstore::load_key(&client_pathfinder.ack_key().to_owned())?; - - // TODO: ack key is never stored so it is generated now. But perhaps it should be stored - // after all for consistency sake? - Ok(KeyManager { - identity_keypair: Arc::new(identity_keypair), - encryption_keypair: Arc::new(encryption_keypair), - gateway_shared_key: Some(Arc::new(gateway_shared_key)), - ack_key: Arc::new(ack_key), - }) - } - - // this is actually **NOT** dead code - // I have absolutely no idea why the compiler insists it's unused. The call happens during client::init::execute - #[allow(dead_code)] - /// Stores all available keys on the disk. - // While perhaps there is no much point in storing the `AckKey` on the disk, - // it is done so for the consistency sake so that you wouldn't require an rng instance - // during `load_keys` to generate the said key. - pub(crate) fn store_keys(&self, client_pathfinder: &ClientKeyPathfinder) -> io::Result<()> { - pemstore::store_keypair( - self.identity_keypair.as_ref(), - &pemstore::KeyPairPath::new( - client_pathfinder.private_identity_key().to_owned(), - client_pathfinder.public_identity_key().to_owned(), - ), - )?; - pemstore::store_keypair( - self.encryption_keypair.as_ref(), - &pemstore::KeyPairPath::new( - client_pathfinder.private_encryption_key().to_owned(), - client_pathfinder.public_encryption_key().to_owned(), - ), - )?; - - pemstore::store_key(self.ack_key.as_ref(), &client_pathfinder.ack_key())?; - - match self.gateway_shared_key.as_ref() { - None => warn!("No gateway shared key available to store!"), - Some(gate_key) => { - pemstore::store_key(gate_key.as_ref(), &client_pathfinder.gateway_shared_key())? - } - } - - Ok(()) - } - - /// Gets an atomically reference counted pointer to [`identity::KeyPair`]. - pub(crate) fn identity_keypair(&self) -> Arc { - Arc::clone(&self.identity_keypair) - } - - /// Gets an atomically reference counted pointer to [`encryption::KeyPair`]. - pub(crate) fn encryption_keypair(&self) -> Arc { - Arc::clone(&self.encryption_keypair) - } - - /// Gets an atomically reference counted pointer to [`SharedKey`]. - // since this function is not fully public, it is not expected to be used externally and - // hence it's up to us to ensure it's called in correct context - pub(crate) fn gateway_shared_key(&self) -> Arc { - Arc::clone( - &self - .gateway_shared_key - .as_ref() - .expect("tried to unwrap empty gateway key!"), - ) - } - - /// Gets an atomically reference counted pointer to [`AckKey`]. - pub(crate) fn ack_key(&self) -> Arc { - Arc::clone(&self.ack_key) - } -} diff --git a/clients/native/src/client/mix_traffic.rs b/clients/native/src/client/mix_traffic.rs deleted file mode 100644 index cd5b78fa6c..0000000000 --- a/clients/native/src/client/mix_traffic.rs +++ /dev/null @@ -1,91 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use futures::channel::mpsc; -use futures::StreamExt; -use gateway_client::GatewayClient; -use log::*; -use nymsphinx::{addressing::nodes::NymNodeRoutingAddress, SphinxPacket}; -use tokio::runtime::Handle; -use tokio::task::JoinHandle; - -pub(crate) struct MixMessage(NymNodeRoutingAddress, SphinxPacket); -pub(crate) type MixMessageSender = mpsc::UnboundedSender; -pub(crate) type MixMessageReceiver = mpsc::UnboundedReceiver; - -impl MixMessage { - pub(crate) fn new(address: NymNodeRoutingAddress, packet: SphinxPacket) -> Self { - MixMessage(address, packet) - } -} - -const MAX_FAILURE_COUNT: usize = 100; - -pub(crate) struct MixTrafficController<'a> { - // TODO: most likely to be replaced by some higher level construct as - // later on gateway_client will need to be accessible by other entities - gateway_client: GatewayClient<'a, url::Url>, - mix_rx: MixMessageReceiver, - - // TODO: this is temporary work-around. - // in long run `gateway_client` will be moved away from `MixTrafficController` anyway. - consecutive_gateway_failure_count: usize, -} - -impl<'a> MixTrafficController<'static> { - pub(crate) fn new( - mix_rx: MixMessageReceiver, - gateway_client: GatewayClient<'a, url::Url>, - ) -> MixTrafficController<'a> { - MixTrafficController { - gateway_client, - mix_rx, - consecutive_gateway_failure_count: 0, - } - } - - async fn on_message(&mut self, mix_message: MixMessage) { - debug!("Got a mix_message for {:?}", mix_message.0); - match self - .gateway_client - .send_sphinx_packet(mix_message.0, mix_message.1) - .await - { - Err(e) => { - error!("Failed to send sphinx packet to the gateway! - {:?}", e); - self.consecutive_gateway_failure_count += 1; - if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT { - // todo: in the future this should initiate a 'graceful' shutdown - panic!("failed to send sphinx packet to the gateway {} times in a row - assuming the gateway is dead. Can't do anything about it yet :(", MAX_FAILURE_COUNT) - } - } - Ok(_) => { - trace!("We *might* have managed to forward sphinx packet to the gateway!"); - self.consecutive_gateway_failure_count = 0; - } - } - } - - pub(crate) async fn run(&mut self) { - while let Some(mix_message) = self.mix_rx.next().await { - self.on_message(mix_message).await; - } - } - - pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { - handle.spawn(async move { - self.run().await; - }) - } -} diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index 5f53829dd4..a5be64e596 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -12,21 +12,27 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::client::cover_traffic_stream::LoopCoverTrafficStream; -use crate::client::inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender}; -use crate::client::key_manager::KeyManager; -use crate::client::mix_traffic::{MixMessageReceiver, MixMessageSender, MixTrafficController}; -use crate::client::real_messages_control::RealMessagesController; -use crate::client::received_buffer::{ - ReceivedBufferRequestReceiver, ReceivedBufferRequestSender, ReceivedMessagesBufferController, +use crate::websocket; +use client_core::client::cover_traffic_stream::LoopCoverTrafficStream; +use client_core::client::inbound_messages::{ + InputMessage, InputMessageReceiver, InputMessageSender, }; -use crate::client::reply_key_storage::ReplyKeyStorage; -use crate::client::topology_control::{ +use client_core::client::key_manager::KeyManager; +use client_core::client::mix_traffic::{ + MixMessageReceiver, MixMessageSender, MixTrafficController, +}; +use client_core::client::real_messages_control; +use client_core::client::real_messages_control::RealMessagesController; +use client_core::client::received_buffer::{ + ReceivedBufferMessage, ReceivedBufferRequestReceiver, ReceivedBufferRequestSender, + ReceivedMessagesBufferController, ReconstructedMessagesReceiver, +}; +use client_core::client::reply_key_storage::ReplyKeyStorage; +use client_core::client::topology_control::{ TopologyAccessor, TopologyRefresher, TopologyRefresherConfig, }; -use crate::config::persistence::key_pathfinder::ClientKeyPathfinder; -use crate::config::{Config, SocketType}; -use crate::websocket; +use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder; +use client_core::config::{Config, SocketType}; use crypto::asymmetric::identity; use futures::channel::mpsc; use gateway_client::{ @@ -38,18 +44,8 @@ use nymsphinx::addressing::clients::Recipient; use nymsphinx::addressing::nodes::NodeIdentity; use nymsphinx::anonymous_replies::ReplySURB; use nymsphinx::receiver::ReconstructedMessage; -use received_buffer::{ReceivedBufferMessage, ReconstructedMessagesReceiver}; use tokio::runtime::Runtime; -mod cover_traffic_stream; -pub(crate) mod inbound_messages; -pub(crate) mod key_manager; -mod mix_traffic; -pub(crate) mod real_messages_control; -pub(crate) mod received_buffer; -mod reply_key_storage; -pub(crate) mod topology_control; - pub struct NymClient { /// Client configuration options, including, among other things, packet sending rates, /// key filepaths, etc. diff --git a/clients/native/src/client/real_messages_control/mod.rs b/clients/native/src/client/real_messages_control/mod.rs deleted file mode 100644 index 3988475b41..0000000000 --- a/clients/native/src/client/real_messages_control/mod.rs +++ /dev/null @@ -1,170 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -// INPUT: InputMessage from user -// INPUT2: Acks from mix -// OUTPUT: MixMessage to mix traffic - -use self::{ - acknowlegement_control::AcknowledgementController, real_traffic_stream::OutQueueControl, -}; -use crate::client::real_messages_control::acknowlegement_control::AcknowledgementControllerConnectors; -use crate::client::reply_key_storage::ReplyKeyStorage; -use crate::client::{ - inbound_messages::InputMessageReceiver, mix_traffic::MixMessageSender, - topology_control::TopologyAccessor, -}; -use futures::channel::mpsc; -use gateway_client::AcknowledgementReceiver; -use log::*; -use nymsphinx::acknowledgements::AckKey; -use nymsphinx::addressing::clients::Recipient; -use rand::{rngs::OsRng, CryptoRng, Rng}; -use std::sync::Arc; -use std::time::Duration; -use tokio::runtime::Handle; -use tokio::task::JoinHandle; - -mod acknowlegement_control; -mod real_traffic_stream; - -pub(crate) struct Config { - ack_key: Arc, - ack_wait_multiplier: f64, - ack_wait_addition: Duration, - self_recipient: Recipient, - average_packet_delay_duration: Duration, - average_ack_delay_duration: Duration, - average_message_sending_delay: Duration, -} - -impl Config { - pub(crate) fn new( - ack_key: Arc, - ack_wait_multiplier: f64, - ack_wait_addition: Duration, - average_ack_delay_duration: Duration, - average_message_sending_delay: Duration, - average_packet_delay_duration: Duration, - self_recipient: Recipient, - ) -> Self { - Config { - ack_key, - self_recipient, - average_packet_delay_duration, - average_ack_delay_duration, - average_message_sending_delay, - ack_wait_multiplier, - ack_wait_addition, - } - } -} - -pub(crate) struct RealMessagesController -where - R: CryptoRng + Rng, -{ - out_queue_control: Option>, - ack_control: Option>, -} - -// obviously when we finally make shared rng that is on 'higher' level, this should become -// generic `R` -impl RealMessagesController { - pub(crate) fn new( - config: Config, - ack_receiver: AcknowledgementReceiver, - input_receiver: InputMessageReceiver, - mix_sender: MixMessageSender, - topology_access: TopologyAccessor, - reply_key_storage: ReplyKeyStorage, - ) -> Self { - let rng = OsRng; - - let (real_message_sender, real_message_receiver) = mpsc::unbounded(); - let (sent_notifier_tx, sent_notifier_rx) = mpsc::unbounded(); - - let ack_controller_connectors = AcknowledgementControllerConnectors::new( - real_message_sender, - input_receiver, - sent_notifier_rx, - ack_receiver, - ); - - let ack_control = AcknowledgementController::new( - rng, - topology_access.clone(), - Arc::clone(&config.ack_key), - config.self_recipient.clone(), - reply_key_storage, - config.average_packet_delay_duration, - config.average_ack_delay_duration, - config.ack_wait_multiplier, - config.ack_wait_addition, - ack_controller_connectors, - ); - - let out_queue_control = OutQueueControl::new( - Arc::clone(&config.ack_key), - config.average_ack_delay_duration, - config.average_packet_delay_duration, - config.average_message_sending_delay, - sent_notifier_tx, - mix_sender, - real_message_receiver, - rng, - config.self_recipient, - topology_access, - ); - - RealMessagesController { - out_queue_control: Some(out_queue_control), - ack_control: Some(ack_control), - } - } - - pub(super) async fn run(&mut self) { - let mut out_queue_control = self.out_queue_control.take().unwrap(); - let mut ack_control = self.ack_control.take().unwrap(); - - // the below are log messages are errors as at the current stage we do not expect any of - // the task to ever finish. This will of course change once we introduce - // graceful shutdowns. - let out_queue_control_fut = tokio::spawn(async move { - out_queue_control.run_out_queue_control().await; - error!("The out queue controller has finished execution!"); - out_queue_control - }); - let ack_control_fut = tokio::spawn(async move { - ack_control.run().await; - error!("The acknowledgement controller has finished execution!"); - ack_control - }); - - // technically we don't have to bring `RealMessagesController` back to a valid state - // but we can do it, so why not? Perhaps it might be useful if we wanted to allow - // for restarts of certain modules without killing the entire process. - self.out_queue_control = Some(out_queue_control_fut.await.unwrap()); - self.ack_control = Some(ack_control_fut.await.unwrap()); - } - - // &Handle is only passed for consistency sake with other client modules, but I think - // when we get to refactoring, we should apply gateway approach and make it implicit - pub(super) fn start(mut self, handle: &Handle) -> JoinHandle { - handle.spawn(async move { - self.run().await; - self - }) - } -} diff --git a/clients/native/src/client/reply_key_storage.rs b/clients/native/src/client/reply_key_storage.rs deleted file mode 100644 index e1dc4a6b9a..0000000000 --- a/clients/native/src/client/reply_key_storage.rs +++ /dev/null @@ -1,105 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use log::*; -use nymsphinx::anonymous_replies::{ - encryption_key::EncryptionKeyDigest, encryption_key::Unsigned, SURBEncryptionKey, - SURBEncryptionKeySize, -}; -use std::path::Path; - -#[derive(Debug)] -pub(crate) enum ReplyKeyStorageError { - DbReadError(sled::Error), - DbWriteError(sled::Error), - DbOpenError(sled::Error), -} - -/// Permanent storage for keys in all sent [`ReplySURB`] -/// -/// Each sent out [`ReplySURB`] has a new key associated with it that is going to be used for -/// payload encryption. In order to decrypt whatever reply we receive, we need to know which -/// key to use for that purpose. We do it based on received `H(t)` which has to be included -/// with each reply. -/// Moreover, there is no restriction when the [`ReplySURB`] might get used so we need to -/// have a permanent storage for all the keys that we might ever see in the future. -#[derive(Debug, Clone)] -pub struct ReplyKeyStorage { - db: sled::Db, -} - -impl ReplyKeyStorage { - pub(crate) fn load>(path: P) -> Result { - let db = match sled::open(path) { - Err(e) => return Err(ReplyKeyStorageError::DbOpenError(e)), - Ok(db) => db, - }; - - Ok(ReplyKeyStorage { db }) - } - - fn read_encryption_key(&self, raw_key: sled::IVec) -> SURBEncryptionKey { - let key_bytes_ref = raw_key.as_ref(); - // if this fails it means we have some database corruption and we - // absolutely can't continue - - if key_bytes_ref.len() != SURBEncryptionKeySize::to_usize() { - error!("REPLY KEY STORAGE DATA CORRUPTION - ENCRYPTION KEY HAS INVALID LENGTH"); - panic!("REPLY KEY STORAGE DATA CORRUPTION - ENCRYPTION KEY HAS INVALID LENGTH"); - } - - // this can only fail if the bytes have invalid length but we already asserted it - SURBEncryptionKey::try_from_bytes(key_bytes_ref).unwrap() - } - - // TOOD: perhaps we could also store some part of original message here too? - pub(crate) fn insert_encryption_key( - &mut self, - encryption_key: SURBEncryptionKey, - ) -> Result<(), ReplyKeyStorageError> { - let digest = encryption_key.compute_digest(); - - let insertion_result = match self.db.insert(digest.to_vec(), encryption_key.to_bytes()) { - Err(e) => Err(ReplyKeyStorageError::DbWriteError(e)), - Ok(existing_key) => { - if existing_key.is_some() { - panic!("HASH COLLISION DETECTED") - }; - Ok(()) - } - }; - - // TODO: perhaps we could implement some batching mechanism to avoid frequent flushes? - self.db.flush().unwrap(); - insertion_result - } - - // Once we use key once, we do not expect to use it again - pub(crate) fn get_and_remove_encryption_key( - &self, - key_digest: EncryptionKeyDigest, - ) -> Result, ReplyKeyStorageError> { - let removal_result = match self.db.remove(&key_digest.to_vec()) { - Err(e) => Err(ReplyKeyStorageError::DbReadError(e)), - Ok(existing_key) => { - Ok(existing_key.map(|existing_key| self.read_encryption_key(existing_key))) - } - }; - - // TODO: not sure how to feel about flushing it every single time here... - // same with insertion - self.db.flush().unwrap(); - removal_result - } -} diff --git a/clients/native/src/commands/init.rs b/clients/native/src/commands/init.rs index f6f7da0c9b..4200a9b1ba 100644 --- a/clients/native/src/commands/init.rs +++ b/clients/native/src/commands/init.rs @@ -13,10 +13,10 @@ // limitations under the License. use crate::built_info; -use crate::client::key_manager::KeyManager; use crate::commands::override_config; -use crate::config::persistence::key_pathfinder::ClientKeyPathfinder; use clap::{App, Arg, ArgMatches}; +use client_core::client::key_manager::KeyManager; +use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder; use config::NymConfig; use crypto::asymmetric::identity; use directory_client::DirectoryClient; @@ -149,7 +149,7 @@ pub fn execute(matches: &ArgMatches) { println!("Initialising client..."); let id = matches.value_of("id").unwrap(); // required for now - let mut config = crate::config::Config::new(id); + let mut config = client_core::config::Config::new(id); let mut rng = OsRng; config = override_config(config, matches); diff --git a/clients/native/src/commands/mod.rs b/clients/native/src/commands/mod.rs index f9bb333966..4eba736317 100644 --- a/clients/native/src/commands/mod.rs +++ b/clients/native/src/commands/mod.rs @@ -12,8 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::config::{Config, SocketType}; use clap::ArgMatches; +use client_core::config::{Config, SocketType}; pub mod init; pub mod run; diff --git a/clients/native/src/commands/run.rs b/clients/native/src/commands/run.rs index 83af38944e..464b3d962c 100644 --- a/clients/native/src/commands/run.rs +++ b/clients/native/src/commands/run.rs @@ -14,8 +14,8 @@ use crate::client::NymClient; use crate::commands::override_config; -use crate::config::Config; use clap::{App, Arg, ArgMatches}; +use client_core::config::Config; use config::NymConfig; pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> { diff --git a/clients/native/src/lib.rs b/clients/native/src/lib.rs index a1716e3ab7..0934a2e435 100644 --- a/clients/native/src/lib.rs +++ b/clients/native/src/lib.rs @@ -14,5 +14,4 @@ pub mod built_info; pub mod client; -pub mod config; pub mod websocket; diff --git a/clients/native/src/main.rs b/clients/native/src/main.rs index 2c033ce245..04e74a88c3 100644 --- a/clients/native/src/main.rs +++ b/clients/native/src/main.rs @@ -16,8 +16,7 @@ use clap::{App, ArgMatches}; pub mod built_info; pub mod client; -mod commands; -pub mod config; +pub mod commands; pub mod websocket; fn main() { diff --git a/clients/native/src/websocket/handler.rs b/clients/native/src/websocket/handler.rs index 8b90f1da54..df0d8cbaeb 100644 --- a/clients/native/src/websocket/handler.rs +++ b/clients/native/src/websocket/handler.rs @@ -13,13 +13,13 @@ // limitations under the License. use super::types::{BinaryClientRequest, ClientTextRequest, ServerTextResponse}; -use crate::client::{ +use crate::websocket::types::ReceivedTextMessage; +use client_core::client::{ inbound_messages::{InputMessage, InputMessageSender}, received_buffer::{ ReceivedBufferMessage, ReceivedBufferRequestSender, ReconstructedMessagesReceiver, }, }; -use crate::websocket::types::ReceivedTextMessage; use futures::channel::mpsc; use futures::{SinkExt, StreamExt}; use log::*; diff --git a/clients/socks5/Cargo.toml b/clients/socks5/Cargo.toml index 773a765d61..e69a1f8dd6 100644 --- a/clients/socks5/Cargo.toml +++ b/clients/socks5/Cargo.toml @@ -29,6 +29,7 @@ tokio-tungstenite = "0.10.1" url = "2.1" ## internal +client-core = { path = "../client-core" } config = {path = "../../common/config"} crypto = {path = "../../common/crypto"} directory-client = { path = "../../common/client-libs/directory-client" } diff --git a/clients/socks5/src/client/cover_traffic_stream.rs b/clients/socks5/src/client/cover_traffic_stream.rs deleted file mode 100644 index 53adb4a78a..0000000000 --- a/clients/socks5/src/client/cover_traffic_stream.rs +++ /dev/null @@ -1,186 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::client::mix_traffic::{MixMessage, MixMessageSender}; -use crate::client::topology_control::TopologyAccessor; -use futures::task::{Context, Poll}; -use futures::{Future, Stream, StreamExt}; -use log::*; -use nymsphinx::acknowledgements::AckKey; -use nymsphinx::addressing::clients::Recipient; -use nymsphinx::cover::generate_loop_cover_packet; -use nymsphinx::utils::sample_poisson_duration; -use rand::{rngs::OsRng, CryptoRng, Rng}; -use std::pin::Pin; -use std::sync::Arc; -use tokio::runtime::Handle; -use tokio::task::JoinHandle; -use tokio::time; - -pub(crate) struct LoopCoverTrafficStream -where - R: CryptoRng + Rng, -{ - /// Key used to encrypt and decrypt content of an ACK packet. - ack_key: Arc, - - /// Average delay an acknowledgement packet is going to get delay at a single mixnode. - average_ack_delay: time::Duration, - - /// Average delay a data packet is going to get delay at a single mixnode. - average_packet_delay: time::Duration, - - /// Average delay between sending subsequent cover packets. - average_cover_message_sending_delay: time::Duration, - - /// Internal state, determined by `average_message_sending_delay`, - /// used to keep track of when a next packet should be sent out. - next_delay: time::Delay, - - /// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them - /// out to the network without any further delays. - mix_tx: MixMessageSender, - - /// Represents full address of this client. - our_full_destination: Recipient, - - /// Instance of a cryptographically secure random number generator. - rng: R, - - /// Accessor to the common instance of network topology. - topology_access: TopologyAccessor, -} - -impl Stream for LoopCoverTrafficStream -where - R: CryptoRng + Rng + Unpin, -{ - // Item is only used to indicate we should create a new message rather than actual cover message - // reason being to not introduce unnecessary complexity by having to keep state of topology - // mutex when trying to acquire it. So right now the Stream trait serves as a glorified timer. - // Perhaps this should be changed in the future. - type Item = (); - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // it is not yet time to return a message - if Pin::new(&mut self.next_delay).poll(cx).is_pending() { - return Poll::Pending; - }; - - // we know it's time to send a message, so let's prepare delay for the next one - // Get the `now` by looking at the current `delay` deadline - let avg_delay = self.average_cover_message_sending_delay; - let now = self.next_delay.deadline(); - let next_poisson_delay = sample_poisson_duration(&mut self.rng, avg_delay); - - // The next interval value is `next_poisson_delay` after the one that just - // yielded. - let next = now + next_poisson_delay; - self.next_delay.reset(next); - - Poll::Ready(Some(())) - } -} - -// obviously when we finally make shared rng that is on 'higher' level, this should become -// generic `R` -impl LoopCoverTrafficStream { - pub(crate) fn new( - ack_key: Arc, - average_ack_delay: time::Duration, - average_packet_delay: time::Duration, - average_cover_message_sending_delay: time::Duration, - mix_tx: MixMessageSender, - our_full_destination: Recipient, - topology_access: TopologyAccessor, - ) -> Self { - let rng = OsRng; - - LoopCoverTrafficStream { - ack_key, - average_ack_delay, - average_packet_delay, - average_cover_message_sending_delay, - next_delay: time::delay_for(Default::default()), - mix_tx, - our_full_destination, - rng, - topology_access, - } - } - - async fn on_new_message(&mut self) { - trace!("next cover message!"); - - // TODO for way down the line: in very rare cases (during topology update) we might have - // to wait a really tiny bit before actually obtaining the permit hence messing with our - // poisson delay, but is it really a problem? - let topology_permit = self.topology_access.get_read_permit().await; - // the ack is sent back to ourselves (and then ignored) - let topology_ref_option = topology_permit.try_get_valid_topology_ref( - &self.our_full_destination, - Some(&self.our_full_destination), - ); - if topology_ref_option.is_none() { - warn!("No valid topology detected - won't send any loop cover message this time"); - return; - } - let topology_ref = topology_ref_option.unwrap(); - - let cover_message = generate_loop_cover_packet( - &mut self.rng, - topology_ref, - &*self.ack_key, - &self.our_full_destination, - self.average_ack_delay, - self.average_packet_delay, - ) - .expect("Somehow failed to generate a loop cover message with a valid topology"); - - // if this one fails, there's no retrying because it means that either: - // - we run out of memory - // - the receiver channel is closed - // in either case there's no recovery and we can only panic - self.mix_tx - .unbounded_send(MixMessage::new(cover_message.0, cover_message.1)) - .unwrap(); - - // TODO: I'm not entirely sure whether this is really required, because I'm not 100% - // sure how `yield_now()` works - whether it just notifies the scheduler or whether it - // properly blocks. So to play it on the safe side, just explicitly drop the read permit - drop(topology_permit); - - // JS: due to identical logical structure to OutQueueControl::on_message(), this is also - // presumably required to prevent bugs in the future. Exact reason is still unknown to me. - tokio::task::yield_now().await; - } - - async fn run(&mut self) { - // we should set initial delay only when we actually start the stream - self.next_delay = time::delay_for(sample_poisson_duration( - &mut self.rng, - self.average_cover_message_sending_delay, - )); - - while let Some(_) = self.next().await { - self.on_new_message().await; - } - } - - pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { - handle.spawn(async move { - self.run().await; - }) - } -} diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index 0693c569e2..24378f1fd5 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -12,24 +12,29 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::client::cover_traffic_stream::LoopCoverTrafficStream; -use crate::client::inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender}; -use crate::client::key_manager::KeyManager; -use crate::client::mix_traffic::{MixMessageReceiver, MixMessageSender, MixTrafficController}; -use crate::client::real_messages_control::RealMessagesController; -use crate::client::received_buffer::{ - ReceivedBufferRequestReceiver, ReceivedBufferRequestSender, ReceivedMessagesBufferController, -}; -use crate::client::reply_key_storage::ReplyKeyStorage; -use crate::client::topology_control::{ - TopologyAccessor, TopologyRefresher, TopologyRefresherConfig, -}; -use crate::config::persistence::key_pathfinder::ClientKeyPathfinder; -use crate::config::{Config, SocketType}; use crate::socks::{ authentication::{AuthenticationMethods, Authenticator, User}, server::SphinxSocksServer, }; +use client_core::client::cover_traffic_stream::LoopCoverTrafficStream; +use client_core::client::inbound_messages::{ + InputMessage, InputMessageReceiver, InputMessageSender, +}; +use client_core::client::key_manager::KeyManager; +use client_core::client::mix_traffic::{ + MixMessageReceiver, MixMessageSender, MixTrafficController, +}; +use client_core::client::real_messages_control::RealMessagesController; +use client_core::client::received_buffer::{ReceivedBufferMessage, ReconstructedMessagesReceiver}; +use client_core::client::received_buffer::{ + ReceivedBufferRequestReceiver, ReceivedBufferRequestSender, ReceivedMessagesBufferController, +}; +use client_core::client::reply_key_storage::ReplyKeyStorage; +use client_core::client::topology_control::{ + TopologyAccessor, TopologyRefresher, TopologyRefresherConfig, +}; +use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder; +use client_core::config::{Config, SocketType}; use crypto::asymmetric::identity; use futures::channel::mpsc; use gateway_client::{ @@ -41,18 +46,8 @@ use nymsphinx::addressing::clients::{ClientEncryptionKey, ClientIdentity, Recipi use nymsphinx::addressing::nodes::NodeIdentity; use nymsphinx::anonymous_replies::ReplySURB; use nymsphinx::receiver::ReconstructedMessage; -use received_buffer::{ReceivedBufferMessage, ReconstructedMessagesReceiver}; use tokio::runtime::Runtime; -mod cover_traffic_stream; -pub(crate) mod inbound_messages; -pub(crate) mod key_manager; -mod mix_traffic; -pub(crate) mod real_messages_control; -pub(crate) mod received_buffer; -mod reply_key_storage; -pub(crate) mod topology_control; - pub struct NymClient { /// Client configuration options, including, among other things, packet sending rates, /// key filepaths, etc. @@ -133,7 +128,7 @@ impl NymClient { input_receiver: InputMessageReceiver, mix_sender: MixMessageSender, ) { - let controller_config = real_messages_control::Config::new( + let controller_config = client_core::client::real_messages_control::Config::new( self.key_manager.ack_key(), self.config.get_ack_wait_multiplier(), self.config.get_ack_wait_addition(), diff --git a/clients/socks5/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs b/clients/socks5/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs deleted file mode 100644 index 9c6bb539ff..0000000000 --- a/clients/socks5/src/client/real_messages_control/acknowlegement_control/acknowledgement_listener.rs +++ /dev/null @@ -1,92 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use super::PendingAcksMap; -use futures::StreamExt; -use gateway_client::AcknowledgementReceiver; -use log::*; -use nymsphinx::{ - acknowledgements::{identifier::recover_identifier, AckKey}, - chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID}, -}; -use std::sync::Arc; - -// responsible for cancelling retransmission timers and removed entries from the map -pub(super) struct AcknowledgementListener { - ack_key: Arc, - ack_receiver: AcknowledgementReceiver, - pending_acks: PendingAcksMap, -} - -impl AcknowledgementListener { - pub(super) fn new( - ack_key: Arc, - ack_receiver: AcknowledgementReceiver, - pending_acks: PendingAcksMap, - ) -> Self { - AcknowledgementListener { - ack_key, - ack_receiver, - pending_acks, - } - } - - async fn on_ack(&mut self, ack_content: Vec) { - debug!("Received an ack"); - let frag_id = match recover_identifier(&self.ack_key, &ack_content) { - None => { - warn!("Received invalid ACK!"); // should we do anything else about that? - return; - } - Some(frag_id_bytes) => match FragmentIdentifier::try_from_bytes(frag_id_bytes) { - Ok(frag_id) => frag_id, - Err(err) => { - warn!("Received invalid ACK! - {:?}", err); // should we do anything else about that? - return; - } - }, - }; - - if frag_id == COVER_FRAG_ID { - trace!("Received an ack for a cover message - no need to do anything"); - return; - } else if frag_id.is_reply() { - debug!("Received an ack for a reply message - no need to do anything!"); - // TODO: probably there will need to be some extra procedure here, something to notify - // user that his reply reached the recipient (since we got an ack) - info!("We received an ack for one of the replies we sent!"); - return; - } - - if let Some(pending_ack) = self.pending_acks.write().await.remove(&frag_id) { - // cancel the retransmission future - pending_ack.retransmission_cancel.notify(); - } else { - warn!("received ACK for packet we haven't stored! - {:?}", frag_id); - } - } - - pub(super) async fn run(&mut self) { - debug!("Started AcknowledgementListener"); - while let Some(acks) = self.ack_receiver.next().await { - // realistically we would only be getting one ack at the time, but if we managed to - // introduce batching in gateway client, this call should be improved to not re-acquire - // write permit on the map every loop iteration - for ack in acks { - self.on_ack(ack).await; - } - } - error!("TODO: error msg. Or maybe panic?") - } -} diff --git a/clients/socks5/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs b/clients/socks5/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs deleted file mode 100644 index 46bc94b991..0000000000 --- a/clients/socks5/src/client/real_messages_control/acknowlegement_control/input_message_listener.rs +++ /dev/null @@ -1,205 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use super::{PendingAcknowledgement, PendingAcksMap}; -use crate::client::reply_key_storage::ReplyKeyStorage; -use crate::client::{ - inbound_messages::{InputMessage, InputMessageReceiver}, - real_messages_control::real_traffic_stream::{RealMessage, RealMessageSender}, - topology_control::TopologyAccessor, -}; -use futures::StreamExt; -use log::*; -use nymsphinx::anonymous_replies::ReplySURB; -use nymsphinx::preparer::MessagePreparer; -use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient}; -use rand::{CryptoRng, Rng}; -use std::sync::Arc; - -// responsible for splitting received message and initial sending attempt -pub(super) struct InputMessageListener -where - R: CryptoRng + Rng, -{ - ack_key: Arc, - ack_recipient: Recipient, - input_receiver: InputMessageReceiver, - message_preparer: MessagePreparer, - pending_acks: PendingAcksMap, - real_message_sender: RealMessageSender, - topology_access: TopologyAccessor, - reply_key_storage: ReplyKeyStorage, -} - -impl InputMessageListener -where - R: CryptoRng + Rng, -{ - pub(super) fn new( - ack_key: Arc, - ack_recipient: Recipient, - input_receiver: InputMessageReceiver, - message_preparer: MessagePreparer, - pending_acks: PendingAcksMap, - real_message_sender: RealMessageSender, - topology_access: TopologyAccessor, - reply_key_storage: ReplyKeyStorage, - ) -> Self { - InputMessageListener { - ack_key, - ack_recipient, - input_receiver, - message_preparer, - pending_acks, - real_message_sender, - topology_access, - reply_key_storage, - } - } - - async fn handle_reply(&mut self, reply_surb: ReplySURB, data: Vec) -> Option { - let topology_permit = self.topology_access.get_read_permit().await; - let topology_ref_option = - topology_permit.try_get_valid_topology_ref(&self.ack_recipient, None); - if topology_ref_option.is_none() { - warn!("Could not process the message - the network topology is invalid"); - return None; - } - let topology = topology_ref_option.unwrap(); - - match self - .message_preparer - .prepare_reply_for_use(data, reply_surb, topology, &self.ack_key) - { - Ok((reply_id, sphinx_packet, first_hop)) => { - // TODO: later probably write pending ack here - // and deal with them.... - // ... somehow - - Some(RealMessage::new(first_hop, sphinx_packet, reply_id)) - } - Err(err) => { - // TODO: should we have some mechanism to indicate to the user that the `reply_surb` - // could be reused since technically it wasn't used up here? - warn!("failed to deal with received reply surb - {:?}", err); - None - } - } - } - - async fn handle_fresh_message( - &mut self, - recipient: Recipient, - content: Vec, - with_reply_surb: bool, - ) -> Vec { - let topology_permit = self.topology_access.get_read_permit().await; - let topology_ref_option = - topology_permit.try_get_valid_topology_ref(&self.ack_recipient, Some(&recipient)); - if topology_ref_option.is_none() { - warn!("Could not process the message - the network topology is invalid"); - return Vec::new(); - } - let topology = topology_ref_option.unwrap(); - - // split the message, attach optional reply surb - let (split_message, reply_key) = self - .message_preparer - .prepare_and_split_message(content, with_reply_surb, topology) - .expect("somehow the topology was invalid after all!"); - - if let Some(reply_key) = reply_key { - self.reply_key_storage - .insert_encryption_key(reply_key) - .expect("Failed to insert surb reply key to the store!") - } - - // encrypt chunks, put them inside sphinx packets and generate acks - let mut pending_acks = Vec::with_capacity(split_message.len()); - let mut real_messages = Vec::with_capacity(split_message.len()); - for message_chunk in split_message { - // since the paths can be constructed, this CAN'T fail, if it does, there's a bug somewhere - let frag_id = message_chunk.fragment_identifier(); - // we need to clone it because we need to keep it in memory in case we had to retransmit - // it. And then we'd need to recreate entire ACK again. - let chunk_clone = message_chunk.clone(); - let prepared_fragment = self - .message_preparer - .prepare_chunk_for_sending(chunk_clone, topology, &self.ack_key, &recipient) - .unwrap(); - - real_messages.push(RealMessage::new( - prepared_fragment.first_hop_address, - prepared_fragment.sphinx_packet, - frag_id, - )); - - let pending_ack = PendingAcknowledgement::new( - message_chunk, - prepared_fragment.total_delay, - recipient.clone(), - ); - - pending_acks.push((frag_id, pending_ack)); - } - - // first insert pending_acks only then request fragments to be sent, otherwise you might get - // some very nasty (and time-consuming to figure out...) race condition. - let mut pending_acks_map_write_guard = self.pending_acks.write().await; - for (frag_id, pending_ack) in pending_acks.into_iter() { - if pending_acks_map_write_guard - .insert(frag_id, pending_ack) - .is_some() - { - panic!("Tried to insert duplicate pending ack") - } - } - - real_messages - } - - async fn on_input_message(&mut self, msg: InputMessage) { - let real_messages = match msg { - InputMessage::Fresh { - recipient, - data, - with_reply_surb, - } => { - self.handle_fresh_message(recipient, data, with_reply_surb) - .await - } - InputMessage::Reply { reply_surb, data } => { - if let Some(real_message) = self.handle_reply(reply_surb, data).await { - vec![real_message] - } else { - return; - } - } - }; - - for real_message in real_messages { - self.real_message_sender - .unbounded_send(real_message) - .unwrap(); - } - } - - pub(super) async fn run(&mut self) { - debug!("Started InputMessageListener"); - while let Some(input_msg) = self.input_receiver.next().await { - self.on_input_message(input_msg).await; - } - error!("TODO: error msg. Or maybe panic?") - } -} diff --git a/clients/socks5/src/client/real_messages_control/acknowlegement_control/mod.rs b/clients/socks5/src/client/real_messages_control/acknowlegement_control/mod.rs deleted file mode 100644 index 6ed812f2bd..0000000000 --- a/clients/socks5/src/client/real_messages_control/acknowlegement_control/mod.rs +++ /dev/null @@ -1,230 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use self::{ - acknowledgement_listener::AcknowledgementListener, - input_message_listener::InputMessageListener, - retransmission_request_listener::RetransmissionRequestListener, - sent_notification_listener::SentNotificationListener, -}; -use super::real_traffic_stream::RealMessageSender; -use crate::client::reply_key_storage::ReplyKeyStorage; -use crate::client::{inbound_messages::InputMessageReceiver, topology_control::TopologyAccessor}; -use futures::channel::mpsc; -use gateway_client::AcknowledgementReceiver; -use log::*; -use nymsphinx::preparer::MessagePreparer; -use nymsphinx::{ - acknowledgements::AckKey, - addressing::clients::Recipient, - chunking::fragment::{Fragment, FragmentIdentifier}, - Delay, -}; -use rand::{CryptoRng, Rng}; -use std::{collections::HashMap, sync::Arc, time::Duration}; -use tokio::{ - sync::{Notify, RwLock}, - task::JoinHandle, -}; - -mod acknowledgement_listener; -mod input_message_listener; -mod retransmission_request_listener; -mod sent_notification_listener; - -type RetransmissionRequestSender = mpsc::UnboundedSender; -type RetransmissionRequestReceiver = mpsc::UnboundedReceiver; - -pub(super) type SentPacketNotificationSender = mpsc::UnboundedSender; -type SentPacketNotificationReceiver = mpsc::UnboundedReceiver; - -type PendingAcksMap = Arc>>; - -struct PendingAcknowledgement { - message_chunk: Fragment, - delay: Delay, - recipient: Recipient, - retransmission_cancel: Arc, -} - -impl PendingAcknowledgement { - fn new(message_chunk: Fragment, delay: Delay, recipient: Recipient) -> Self { - PendingAcknowledgement { - message_chunk, - delay, - retransmission_cancel: Arc::new(Notify::new()), - recipient, - } - } - - fn update_delay(&mut self, new_delay: Delay) { - self.delay = new_delay; - } -} - -pub(super) struct AcknowledgementControllerConnectors { - real_message_sender: RealMessageSender, - input_receiver: InputMessageReceiver, - sent_notifier: SentPacketNotificationReceiver, - ack_receiver: AcknowledgementReceiver, -} - -impl AcknowledgementControllerConnectors { - pub(super) fn new( - real_message_sender: RealMessageSender, - input_receiver: InputMessageReceiver, - sent_notifier: SentPacketNotificationReceiver, - ack_receiver: AcknowledgementReceiver, - ) -> Self { - AcknowledgementControllerConnectors { - real_message_sender, - input_receiver, - sent_notifier, - ack_receiver, - } - } -} - -pub(super) struct AcknowledgementController -where - R: CryptoRng + Rng, -{ - acknowledgement_listener: Option, - input_message_listener: Option>, - retransmission_request_listener: Option>, - sent_notification_listener: Option, -} - -impl AcknowledgementController -where - R: 'static + CryptoRng + Rng + Clone + Send, -{ - pub(super) fn new( - rng: R, - topology_access: TopologyAccessor, - ack_key: Arc, - ack_recipient: Recipient, - reply_key_storage: ReplyKeyStorage, - average_packet_delay: Duration, - average_ack_delay: Duration, - ack_wait_multiplier: f64, - ack_wait_addition: Duration, - connectors: AcknowledgementControllerConnectors, - ) -> Self { - let pending_acks = Arc::new(RwLock::new(HashMap::new())); - let message_preparer = MessagePreparer::new( - rng, - ack_recipient.clone(), - average_packet_delay, - average_ack_delay, - ); - - let acknowledgement_listener = AcknowledgementListener::new( - Arc::clone(&ack_key), - connectors.ack_receiver, - Arc::clone(&pending_acks), - ); - - let input_message_listener = InputMessageListener::new( - Arc::clone(&ack_key), - ack_recipient.clone(), - connectors.input_receiver, - message_preparer.clone(), - Arc::clone(&pending_acks), - connectors.real_message_sender.clone(), - topology_access.clone(), - reply_key_storage, - ); - - let (retransmission_tx, retransmission_rx) = mpsc::unbounded(); - - let retransmission_request_listener = RetransmissionRequestListener::new( - Arc::clone(&ack_key), - ack_recipient, - message_preparer, - Arc::clone(&pending_acks), - connectors.real_message_sender, - retransmission_rx, - topology_access, - ); - - let sent_notification_listener = SentNotificationListener::new( - ack_wait_multiplier, - ack_wait_addition, - connectors.sent_notifier, - pending_acks, - retransmission_tx, - ); - - AcknowledgementController { - acknowledgement_listener: Some(acknowledgement_listener), - input_message_listener: Some(input_message_listener), - retransmission_request_listener: Some(retransmission_request_listener), - sent_notification_listener: Some(sent_notification_listener), - } - } - - pub(super) async fn run(&mut self) { - let mut acknowledgement_listener = self.acknowledgement_listener.take().unwrap(); - let mut input_message_listener = self.input_message_listener.take().unwrap(); - let mut retransmission_request_listener = - self.retransmission_request_listener.take().unwrap(); - let mut sent_notification_listener = self.sent_notification_listener.take().unwrap(); - - // TODO: perhaps an extra 'DEBUG' task that would periodically check for stale entries in - // pending acks map? - // It would only be 'DEBUG' as I don't expect any stale entries to exist there to begin with, - // but when can bugs be expected to begin with? - - // the below are log messages are errors as at the current stage we do not expect any of - // the task to ever finish. This will of course change once we introduce - // graceful shutdowns. - let ack_listener_fut = tokio::spawn(async move { - acknowledgement_listener.run().await; - error!("The acknowledgement listener has finished execution!"); - acknowledgement_listener - }); - let input_listener_fut = tokio::spawn(async move { - input_message_listener.run().await; - error!("The input listener has finished execution!"); - input_message_listener - }); - let retransmission_req_fut = tokio::spawn(async move { - retransmission_request_listener.run().await; - error!("The retransmission request listener has finished execution!"); - retransmission_request_listener - }); - let sent_notification_fut = tokio::spawn(async move { - sent_notification_listener.run().await; - error!("The sent notification listener has finished execution!"); - sent_notification_listener - }); - - // technically we don't have to bring `AcknowledgementController` back to a valid state - // but we can do it, so why not? Perhaps it might be useful if we wanted to allow - // for restarts of certain modules without killing the entire process. - self.acknowledgement_listener = Some(ack_listener_fut.await.unwrap()); - self.input_message_listener = Some(input_listener_fut.await.unwrap()); - self.retransmission_request_listener = Some(retransmission_req_fut.await.unwrap()); - self.sent_notification_listener = Some(sent_notification_fut.await.unwrap()); - } - - #[allow(dead_code)] - pub(super) fn start(mut self) -> JoinHandle { - tokio::spawn(async move { - self.run().await; - self - }) - } -} diff --git a/clients/socks5/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs b/clients/socks5/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs deleted file mode 100644 index ed25032438..0000000000 --- a/clients/socks5/src/client/real_messages_control/acknowlegement_control/retransmission_request_listener.rs +++ /dev/null @@ -1,129 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use super::{PendingAcksMap, RetransmissionRequestReceiver}; -use crate::client::{ - real_messages_control::real_traffic_stream::{RealMessage, RealMessageSender}, - topology_control::TopologyAccessor, -}; -use futures::StreamExt; -use log::*; -use nymsphinx::preparer::MessagePreparer; -use nymsphinx::{ - acknowledgements::AckKey, addressing::clients::Recipient, - chunking::fragment::FragmentIdentifier, -}; -use rand::{CryptoRng, Rng}; -use std::sync::Arc; - -// responsible for packet retransmission upon fired timer -pub(super) struct RetransmissionRequestListener -where - R: CryptoRng + Rng, -{ - ack_key: Arc, - ack_recipient: Recipient, - message_preparer: MessagePreparer, - pending_acks: PendingAcksMap, - real_message_sender: RealMessageSender, - request_receiver: RetransmissionRequestReceiver, - topology_access: TopologyAccessor, -} - -impl RetransmissionRequestListener -where - R: CryptoRng + Rng, -{ - pub(super) fn new( - ack_key: Arc, - ack_recipient: Recipient, - message_preparer: MessagePreparer, - pending_acks: PendingAcksMap, - real_message_sender: RealMessageSender, - request_receiver: RetransmissionRequestReceiver, - topology_access: TopologyAccessor, - ) -> Self { - RetransmissionRequestListener { - ack_key, - ack_recipient, - message_preparer, - pending_acks, - real_message_sender, - request_receiver, - topology_access, - } - } - - async fn on_retransmission_request(&mut self, frag_id: FragmentIdentifier) { - let pending_acks_map_read_guard = self.pending_acks.read().await; - // if the unwrap failed here, we have some weird bug somewhere - honestly, I'm not sure - // if it's even possible for it to happen - let unreceived_ack_fragment = pending_acks_map_read_guard - .get(&frag_id) - .expect("wanted to retransmit ack'd fragment"); - - let packet_recipient = unreceived_ack_fragment.recipient.clone(); - let chunk_clone = unreceived_ack_fragment.message_chunk.clone(); - let frag_id = unreceived_ack_fragment.message_chunk.fragment_identifier(); - - // TODO: we need some proper benchmarking here to determine whether it could - // be more efficient to just get write lock and keep it while doing sphinx computation, - // but my gut feeling tells me we should re-acquire it. - drop(pending_acks_map_read_guard); - - let topology_permit = self.topology_access.get_read_permit().await; - let topology_ref_option = topology_permit - .try_get_valid_topology_ref(&self.ack_recipient, Some(&packet_recipient)); - if topology_ref_option.is_none() { - warn!("Could not retransmit the packet - the network topology is invalid"); - // TODO: perhaps put back into pending acks and reset the timer? - return; - } - let topology_ref = topology_ref_option.unwrap(); - - let prepared_fragment = self - .message_preparer - .prepare_chunk_for_sending(chunk_clone, topology_ref, &self.ack_key, &packet_recipient) - .unwrap(); - - // minor optimization to not hold the permit while we no longer need it and might have to block - // waiting for the write lock on `pending_acks` - drop(topology_permit); - - self.pending_acks - .write() - .await - .get_mut(&frag_id) - .expect( - "on_retransmission_request: somehow we already received an ack for this packet?", - ) - .update_delay(prepared_fragment.total_delay); - - self.real_message_sender - .unbounded_send(RealMessage::new( - prepared_fragment.first_hop_address, - prepared_fragment.sphinx_packet, - frag_id, - )) - .unwrap(); - } - - pub(super) async fn run(&mut self) { - debug!("Started RetransmissionRequestListener"); - while let Some(frag_id) = self.request_receiver.next().await { - self.on_retransmission_request(frag_id).await; - } - error!("TODO: error msg. Or maybe panic?") - } -} diff --git a/clients/socks5/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs b/clients/socks5/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs deleted file mode 100644 index 30efb7cc81..0000000000 --- a/clients/socks5/src/client/real_messages_control/acknowlegement_control/sent_notification_listener.rs +++ /dev/null @@ -1,109 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use super::{PendingAcksMap, RetransmissionRequestSender, SentPacketNotificationReceiver}; -use futures::StreamExt; -use log::*; -use nymsphinx::chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID}; -use std::sync::Arc; -use std::time::Duration; - -// responsible for starting and controlling retransmission timers -// it is required because when we send our packet to the `real traffic stream` controlled -// with poisson timer, there's no guarantee the message will be sent immediately, so we might -// accidentally fire retransmission way quicker than we would have wanted. -pub(super) struct SentNotificationListener { - ack_wait_multiplier: f64, - ack_wait_addition: Duration, - sent_notifier: SentPacketNotificationReceiver, - pending_acks: PendingAcksMap, - retransmission_sender: RetransmissionRequestSender, -} - -impl SentNotificationListener { - pub(super) fn new( - ack_wait_multiplier: f64, - ack_wait_addition: Duration, - sent_notifier: SentPacketNotificationReceiver, - pending_acks: PendingAcksMap, - retransmission_sender: RetransmissionRequestSender, - ) -> Self { - SentNotificationListener { - ack_wait_multiplier, - ack_wait_addition, - sent_notifier, - pending_acks, - retransmission_sender, - } - } - - async fn on_sent_message(&mut self, frag_id: FragmentIdentifier) { - if frag_id == COVER_FRAG_ID { - trace!("sent off a cover message - no need to start retransmission timer!"); - return; - } else if frag_id.is_reply() { - debug!("sent off a reply message - no need to start retransmission timer!"); - // TODO: probably there will need to be some extra procedure here, like it would - // be nice to know that our reply actually reached the recipient (i.e. we got the ack) - return; - } - - let pending_acks_map_read_guard = self.pending_acks.read().await; - // if the unwrap failed here, we have some weird bug somewhere - // although when I think about it, it *theoretically* could happen under extremely heavy client - // load that `on_sent_message()` is not called (and we do not receive the read permit) - // until we already received and processed an ack for the packet - // but this seems extremely unrealistic, but perhaps we should guard against that? - let pending_ack_data = pending_acks_map_read_guard - .get(&frag_id) - .expect("on_sent_message: somehow we already received an ack for this packet?"); - - // if this assertion ever fails, we have some bug due to some unintended leak. - // the only reason I see it could happen if the `tokio::select` in the spawned - // task below somehow did not drop it - debug_assert_eq!( - Arc::strong_count(&pending_ack_data.retransmission_cancel), - 1 - ); - - // TODO: read more about Arc::downgrade. it could be useful here - let retransmission_cancel = Arc::clone(&pending_ack_data.retransmission_cancel); - - let retransmission_timeout = tokio::time::delay_for( - (pending_ack_data.delay.clone() * self.ack_wait_multiplier).to_duration() - + self.ack_wait_addition, - ); - - let retransmission_sender = self.retransmission_sender.clone(); - tokio::spawn(async move { - tokio::select! { - _ = retransmission_cancel.notified() => { - trace!("received ack for the fragment. Cancelling retransmission future"); - } - _ = retransmission_timeout => { - trace!("did not receive an ack - will retransmit the packet"); - retransmission_sender.unbounded_send(frag_id).unwrap(); - } - } - }); - } - - pub(super) async fn run(&mut self) { - debug!("Started SentNotificationListener"); - while let Some(frag_id) = self.sent_notifier.next().await { - self.on_sent_message(frag_id).await; - } - error!("TODO: error msg. Or maybe panic?") - } -} diff --git a/clients/socks5/src/client/real_messages_control/real_traffic_stream.rs b/clients/socks5/src/client/real_messages_control/real_traffic_stream.rs deleted file mode 100644 index 539370a4bb..0000000000 --- a/clients/socks5/src/client/real_messages_control/real_traffic_stream.rs +++ /dev/null @@ -1,245 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::client::mix_traffic::{MixMessage, MixMessageSender}; -use crate::client::real_messages_control::acknowlegement_control::SentPacketNotificationSender; -use crate::client::topology_control::TopologyAccessor; -use futures::channel::mpsc; -use futures::task::{Context, Poll}; -use futures::{Future, Stream, StreamExt}; -use log::*; -use nymsphinx::acknowledgements::AckKey; -use nymsphinx::addressing::{clients::Recipient, nodes::NymNodeRoutingAddress}; -use nymsphinx::chunking::fragment::FragmentIdentifier; -use nymsphinx::cover::generate_loop_cover_packet; -use nymsphinx::utils::sample_poisson_duration; -use nymsphinx::SphinxPacket; -use rand::{CryptoRng, Rng}; -use std::pin::Pin; -use std::sync::Arc; -use std::time::Duration; -use tokio::time; - -pub(crate) struct OutQueueControl -where - R: CryptoRng + Rng, -{ - /// Key used to encrypt and decrypt content of an ACK packet. - ack_key: Arc, - - /// Average delay an acknowledgement packet is going to get delay at a single mixnode. - average_ack_delay: Duration, - - /// Average delay a data packet is going to get delay at a single mixnode. - average_packet_delay: Duration, - - /// Average delay between sending subsequent packets. - average_message_sending_delay: Duration, - - /// Channel used for notifying of a real packet being sent out. Used to start up retransmission timer. - sent_notifier: SentPacketNotificationSender, - - /// Internal state, determined by `average_message_sending_delay`, - /// used to keep track of when a next packet should be sent out. - next_delay: time::Delay, - - /// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them - /// out to the network without any further delays. - mix_tx: MixMessageSender, - - /// Channel used for receiving real, prepared, messages that must be first sufficiently delayed - /// before being sent out into the network. - real_receiver: RealMessageReceiver, - - /// Represents full address of this client. - our_full_destination: Recipient, - - /// Instance of a cryptographically secure random number generator. - rng: R, - - /// Accessor to the common instance of network topology. - topology_access: TopologyAccessor, -} - -pub(crate) struct RealMessage { - first_hop_address: NymNodeRoutingAddress, - packet: SphinxPacket, - fragment_id: FragmentIdentifier, -} - -impl RealMessage { - pub(crate) fn new( - first_hop_address: NymNodeRoutingAddress, - packet: SphinxPacket, - fragment_id: FragmentIdentifier, - ) -> Self { - RealMessage { - first_hop_address, - packet, - fragment_id, - } - } -} - -// messages are already prepared, etc. the real point of it is to forward it to mix_traffic -// after sufficient delay -pub(crate) type RealMessageSender = mpsc::UnboundedSender; -type RealMessageReceiver = mpsc::UnboundedReceiver; - -pub(crate) enum StreamMessage { - Cover, - Real(RealMessage), -} - -impl Stream for OutQueueControl -where - R: CryptoRng + Rng + Unpin, -{ - type Item = StreamMessage; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // it is not yet time to return a message - if Pin::new(&mut self.next_delay).poll(cx).is_pending() { - return Poll::Pending; - }; - - // we know it's time to send a message, so let's prepare delay for the next one - // Get the `now` by looking at the current `delay` deadline - let avg_delay = self.average_message_sending_delay; - let now = self.next_delay.deadline(); - let next_poisson_delay = sample_poisson_duration(&mut self.rng, avg_delay); - - // The next interval value is `next_poisson_delay` after the one that just - // yielded. - let next = now + next_poisson_delay; - self.next_delay.reset(next); - - // decide what kind of message to send - match Pin::new(&mut self.real_receiver).poll_next(cx) { - // in the case our real message channel stream was closed, we should also indicate we are closed - // (and whoever is using the stream should panic) - Poll::Ready(None) => Poll::Ready(None), - - // if there's an actual message - return it - Poll::Ready(Some(real_message)) => Poll::Ready(Some(StreamMessage::Real(real_message))), - - // otherwise construct a dummy one - Poll::Pending => Poll::Ready(Some(StreamMessage::Cover)), - } - } -} - -impl OutQueueControl -where - R: CryptoRng + Rng + Unpin, -{ - pub(crate) fn new( - ack_key: Arc, - average_ack_delay: Duration, - average_packet_delay: Duration, - average_message_sending_delay: Duration, - sent_notifier: SentPacketNotificationSender, - mix_tx: MixMessageSender, - real_receiver: RealMessageReceiver, - rng: R, - our_full_destination: Recipient, - topology_access: TopologyAccessor, - ) -> Self { - OutQueueControl { - ack_key, - average_ack_delay, - average_packet_delay, - average_message_sending_delay, - sent_notifier, - next_delay: time::delay_for(Default::default()), - mix_tx, - real_receiver, - our_full_destination, - rng, - topology_access, - } - } - - async fn on_message(&mut self, next_message: StreamMessage) { - trace!("created new message"); - - let next_message = match next_message { - StreamMessage::Cover => { - // TODO for way down the line: in very rare cases (during topology update) we might have - // to wait a really tiny bit before actually obtaining the permit hence messing with our - // poisson delay, but is it really a problem? - let topology_permit = self.topology_access.get_read_permit().await; - // the ack is sent back to ourselves (and then ignored) - let topology_ref_option = topology_permit.try_get_valid_topology_ref( - &self.our_full_destination, - Some(&self.our_full_destination), - ); - if topology_ref_option.is_none() { - warn!( - "No valid topology detected - won't send any loop cover message this time" - ); - return; - } - let topology_ref = topology_ref_option.unwrap(); - - let cover_message = generate_loop_cover_packet( - &mut self.rng, - topology_ref, - &*self.ack_key, - &self.our_full_destination, - self.average_ack_delay, - self.average_packet_delay, - ) - .expect("Somehow failed to generate a loop cover message with a valid topology"); - - MixMessage::new(cover_message.0, cover_message.1) - } - StreamMessage::Real(real_message) => { - // well technically the message was not sent just yet, but now it's up to internal - // queues and client load rather than the required delay. So realistically we can treat - // whatever is about to happen as negligible additional delay. - self.sent_notifier - .unbounded_send(real_message.fragment_id) - .unwrap(); - MixMessage::new(real_message.first_hop_address, real_message.packet) - } - }; - - // if this one fails, there's no retrying because it means that either: - // - we run out of memory - // - the receiver channel is closed - // in either case there's no recovery and we can only panic - self.mix_tx.unbounded_send(next_message).unwrap(); - - // JS: Not entirely sure why or how it fixes stuff, but without the yield call, - // the UnboundedReceiver [of mix_rx] will not get a chance to read anything - // JS2: Basically it was the case that with high enough rate, the stream had already a next value - // ready and hence was immediately re-scheduled causing other tasks to be starved; - // yield makes it go back the scheduling queue regardless of its value availability - tokio::task::yield_now().await; - } - - pub(crate) async fn run_out_queue_control(&mut self) { - // we should set initial delay only when we actually start the stream - self.next_delay = time::delay_for(sample_poisson_duration( - &mut self.rng, - self.average_message_sending_delay, - )); - - info!("Starting out queue controller..."); - while let Some(next_message) = self.next().await { - self.on_message(next_message).await; - } - } -} diff --git a/clients/socks5/src/client/received_buffer.rs b/clients/socks5/src/client/received_buffer.rs deleted file mode 100644 index c3bf2d2bb2..0000000000 --- a/clients/socks5/src/client/received_buffer.rs +++ /dev/null @@ -1,374 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::client::reply_key_storage::ReplyKeyStorage; -use crypto::asymmetric::encryption; -use crypto::symmetric::stream_cipher; -use crypto::Digest; -use futures::channel::mpsc; -use futures::lock::Mutex; -use futures::StreamExt; -use gateway_client::MixnetMessageReceiver; -use log::*; -use nymsphinx::anonymous_replies::{encryption_key::EncryptionKeyDigest, SURBEncryptionKey}; -use nymsphinx::params::{ReplySURBEncryptionAlgorithm, ReplySURBKeyDigestAlgorithm}; -use nymsphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage}; -use std::collections::HashSet; -use std::sync::Arc; -use tokio::runtime::Handle; -use tokio::task::JoinHandle; - -// Buffer Requests to say "hey, send any reconstructed messages to this channel" -// or to say "hey, I'm going offline, don't send anything more to me. Just buffer them instead" -pub(crate) type ReceivedBufferRequestSender = mpsc::UnboundedSender; -pub(crate) type ReceivedBufferRequestReceiver = mpsc::UnboundedReceiver; - -// The channel set for the above -pub(crate) type ReconstructedMessagesSender = mpsc::UnboundedSender>; -pub(crate) type ReconstructedMessagesReceiver = mpsc::UnboundedReceiver>; - -struct ReceivedMessagesBufferInner { - messages: Vec, - local_encryption_keypair: Arc, - - // TODO: looking how it 'looks' here, perhaps `MessageReceiver` should be renamed to something - // else instead. - message_receiver: MessageReceiver, - message_sender: Option, - - // TODO: this will get cleared upon re-running the client - // but perhaps it should be changed to include timestamps of when the message was reconstructed - // and every now and then remove ids older than X - recently_reconstructed: HashSet, -} - -impl ReceivedMessagesBufferInner { - fn process_received_fragment(&mut self, raw_fragment: Vec) -> Option { - let fragment_data = match self - .message_receiver - .recover_plaintext(self.local_encryption_keypair.private_key(), raw_fragment) - { - Err(e) => { - warn!("failed to recover fragment data: {:?}. The whole underlying message might be corrupted and unrecoverable!", e); - return None; - } - Ok(frag_data) => frag_data, - }; - - if nymsphinx::cover::is_cover(&fragment_data) { - trace!("The message was a loop cover message! Skipping it"); - return None; - } - - let fragment = match self.message_receiver.recover_fragment(&fragment_data) { - Err(e) => { - warn!("failed to recover fragment from raw data: {:?}. The whole underlying message might be corrupted and unrecoverable!", e); - return None; - } - Ok(frag) => frag, - }; - - if self.recently_reconstructed.contains(&fragment.id()) { - debug!("Received a chunk of already re-assembled message ({:?})! It probably got here because the ack got lost", fragment.id()); - return None; - } - - // if we returned an error the underlying message is malformed in some way - match self.message_receiver.insert_new_fragment(fragment) { - Err(err) => match err { - MessageRecoveryError::MalformedReconstructedMessage(message_sets) => { - // TODO: should we really insert reconstructed sets? could this be abused for some attack? - for set_id in message_sets { - if !self.recently_reconstructed.insert(set_id) { - // or perhaps we should even panic at this point? - error!("Reconstructed another message containing already used set id!") - } - } - None - } - _ => unreachable!( - "no other error kind should have been returned here! If so, it's a bug!" - ), - }, - Ok(reconstruction_result) => match reconstruction_result { - Some((reconstructed_message, used_sets)) => { - for set_id in used_sets { - if !self.recently_reconstructed.insert(set_id) { - // or perhaps we should even panic at this point? - error!("Reconstructed another message containing already used set id!") - } - } - Some(reconstructed_message) - } - None => None, - }, - } - } -} - -#[derive(Debug, Clone)] -// Note: you should NEVER create more than a single instance of this using 'new()'. -// You should always use .clone() to create additional instances -struct ReceivedMessagesBuffer { - inner: Arc>, - - /// Storage containing keys to all [`ReplySURB`]s ever sent out that we did not receive back. - // There's no need to put it behind a Mutex since it's already properly concurrent - reply_key_storage: ReplyKeyStorage, -} - -impl ReceivedMessagesBuffer { - fn new( - local_encryption_keypair: Arc, - reply_key_storage: ReplyKeyStorage, - ) -> Self { - ReceivedMessagesBuffer { - inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner { - messages: Vec::new(), - local_encryption_keypair, - message_receiver: MessageReceiver::new(), - message_sender: None, - recently_reconstructed: HashSet::new(), - })), - reply_key_storage, - } - } - - async fn disconnect_sender(&mut self) { - let mut guard = self.inner.lock().await; - if guard.message_sender.is_none() { - // in theory we could just ignore it, but that situation should have never happened - // in the first place, so this way we at least know we have an important bug to fix - panic!("trying to disconnect non-existent sender!") - } - guard.message_sender = None; - } - - async fn connect_sender(&mut self, sender: ReconstructedMessagesSender) { - let mut guard = self.inner.lock().await; - if guard.message_sender.is_some() { - // in theory we could just ignore it, but that situation should have never happened - // in the first place, so this way we at least know we have an important bug to fix - panic!("trying overwrite an existing sender!") - } - - // while we're at it, also empty the buffer if we happened to receive anything while - // no sender was connected - let stored_messages = std::mem::replace(&mut guard.messages, Vec::new()); - if !stored_messages.is_empty() { - if let Err(err) = sender.unbounded_send(stored_messages) { - error!( - "The sender channel we just received is already invalidated - {:?}", - err - ); - // put the values back to the buffer - // the returned error has two fields: err: SendError and val: T, - // where val is the value that was failed to get sent; - // it's returned by the `into_inner` call - guard.messages = err.into_inner(); - return; - } - } - guard.message_sender = Some(sender); - } - - async fn add_reconstructed_messages(&mut self, msgs: Vec) { - debug!("Adding {:?} new messages to the buffer!", msgs.len()); - trace!("Adding new messages to the buffer! {:?}", msgs); - self.inner.lock().await.messages.extend(msgs) - } - - fn process_received_reply( - reply_ciphertext: &[u8], - reply_key: SURBEncryptionKey, - ) -> Option { - let zero_iv = stream_cipher::zero_iv::(); - - let mut reply_msg = stream_cipher::decrypt::( - &reply_key.inner(), - &zero_iv, - reply_ciphertext, - ); - if let Err(err) = MessageReceiver::remove_padding(&mut reply_msg) { - warn!("Received reply had malformed padding! - {:?}", err); - None - } else { - // TODO: perhaps having to say it doesn't have a surb an indication the type should be changed? - Some(ReconstructedMessage { - message: reply_msg, - reply_SURB: None, - }) - } - } - - async fn handle_new_received(&mut self, msgs: Vec>) { - debug!( - "Processing {:?} new message that might get added to the buffer!", - msgs.len() - ); - - let mut completed_messages = Vec::new(); - let mut inner_guard = self.inner.lock().await; - - let reply_surb_digest_size = ReplySURBKeyDigestAlgorithm::output_size(); - - // first check if this is a reply or a chunked message - // TODO: verify with @AP if this way of doing it is safe or whether it could - // cause some attacks due to, I don't know, stupid edge case collisions? - // Update: this DOES introduce a possible leakage: https://github.com/nymtech/nym/issues/296 - for msg in msgs { - let possible_key_digest = - EncryptionKeyDigest::clone_from_slice(&msg[..reply_surb_digest_size]); - - // check first `HasherOutputSize` bytes if they correspond to known encryption key - // if yes - this is a reply message - - // TODO: this might be a bottleneck - since the keys are stored on disk we, presumably, - // are doing a disk operation every single received fragment - if let Some(reply_encryption_key) = self - .reply_key_storage - .get_and_remove_encryption_key(possible_key_digest) - .expect("storage operation failed!") - { - if let Some(completed_message) = Self::process_received_reply( - &msg[reply_surb_digest_size..], - reply_encryption_key, - ) { - completed_messages.push(completed_message) - } - } else { - // otherwise - it's a 'normal' message - if let Some(completed_message) = inner_guard.process_received_fragment(msg) { - completed_messages.push(completed_message) - } - } - } - - if !completed_messages.is_empty() { - if let Some(sender) = &inner_guard.message_sender { - trace!("Sending reconstructed messages to announced sender"); - if let Err(err) = sender.unbounded_send(completed_messages) { - warn!("The reconstructed message receiver went offline without explicit notification (relevant error: - {:?})", err); - // make sure to drop the lock to not deadlock - // (it is required by `add_reconstructed_messages`) - inner_guard.message_sender = None; - drop(inner_guard); - self.add_reconstructed_messages(err.into_inner()).await; - } - } else { - // make sure to drop the lock to not deadlock - // (it is required by `add_reconstructed_messages`) - drop(inner_guard); - trace!("No sender available - buffering reconstructed messages"); - self.add_reconstructed_messages(completed_messages).await; - } - } - } -} - -pub(crate) enum ReceivedBufferMessage { - // Signals a websocket connection (or a native implementation) was established and we should stop buffering messages, - // and instead send them directly to the received channel - ReceiverAnnounce(ReconstructedMessagesSender), - - // Explicit signal that Receiver connection will no longer accept messages - ReceiverDisconnect, -} - -struct RequestReceiver { - received_buffer: ReceivedMessagesBuffer, - query_receiver: ReceivedBufferRequestReceiver, -} - -impl RequestReceiver { - fn new( - received_buffer: ReceivedMessagesBuffer, - query_receiver: ReceivedBufferRequestReceiver, - ) -> Self { - RequestReceiver { - received_buffer, - query_receiver, - } - } - - fn start(mut self, handle: &Handle) -> JoinHandle<()> { - handle.spawn(async move { - while let Some(request) = self.query_receiver.next().await { - match request { - ReceivedBufferMessage::ReceiverAnnounce(sender) => { - self.received_buffer.connect_sender(sender).await; - } - ReceivedBufferMessage::ReceiverDisconnect => { - self.received_buffer.disconnect_sender().await - } - } - } - }) - } -} - -struct FragmentedMessageReceiver { - received_buffer: ReceivedMessagesBuffer, - mixnet_packet_receiver: MixnetMessageReceiver, -} - -impl FragmentedMessageReceiver { - fn new( - received_buffer: ReceivedMessagesBuffer, - mixnet_packet_receiver: MixnetMessageReceiver, - ) -> Self { - FragmentedMessageReceiver { - received_buffer, - mixnet_packet_receiver, - } - } - fn start(mut self, handle: &Handle) -> JoinHandle<()> { - handle.spawn(async move { - while let Some(new_messages) = self.mixnet_packet_receiver.next().await { - self.received_buffer.handle_new_received(new_messages).await; - } - }) - } -} - -pub(crate) struct ReceivedMessagesBufferController { - fragmented_message_receiver: FragmentedMessageReceiver, - request_receiver: RequestReceiver, -} - -impl ReceivedMessagesBufferController { - pub(crate) fn new( - local_encryption_keypair: Arc, - query_receiver: ReceivedBufferRequestReceiver, - mixnet_packet_receiver: MixnetMessageReceiver, - reply_key_storage: ReplyKeyStorage, - ) -> Self { - let received_buffer = - ReceivedMessagesBuffer::new(local_encryption_keypair, reply_key_storage); - - ReceivedMessagesBufferController { - fragmented_message_receiver: FragmentedMessageReceiver::new( - received_buffer.clone(), - mixnet_packet_receiver, - ), - request_receiver: RequestReceiver::new(received_buffer, query_receiver), - } - } - - pub(crate) fn start(self, handle: &Handle) { - // TODO: should we do anything with JoinHandle(s) returned by start methods? - self.fragmented_message_receiver.start(handle); - self.request_receiver.start(handle); - } -} diff --git a/clients/socks5/src/client/topology_control.rs b/clients/socks5/src/client/topology_control.rs deleted file mode 100644 index 366c1844f2..0000000000 --- a/clients/socks5/src/client/topology_control.rs +++ /dev/null @@ -1,204 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::built_info; -use directory_client::DirectoryClient; -use log::*; -use nymsphinx::addressing::clients::Recipient; -use nymsphinx::params::DEFAULT_NUM_MIX_HOPS; -use std::convert::TryInto; -use std::ops::Deref; -use std::sync::Arc; -use std::time; -use std::time::Duration; -use tokio::runtime::Handle; -use tokio::sync::{RwLock, RwLockReadGuard}; -use tokio::task::JoinHandle; -use topology::NymTopology; - -// I'm extremely curious why compiler NEVER complained about lack of Debug here before -#[derive(Debug)] -pub(super) struct TopologyAccessorInner(Option); - -impl AsRef> for TopologyAccessorInner { - fn as_ref(&self) -> &Option { - &self.0 - } -} - -impl TopologyAccessorInner { - fn new() -> Self { - TopologyAccessorInner(None) - } - - fn update(&mut self, new: Option) { - self.0 = new; - } -} - -pub(super) struct TopologyReadPermit<'a> { - permit: RwLockReadGuard<'a, TopologyAccessorInner>, -} - -impl<'a> Deref for TopologyReadPermit<'a> { - type Target = TopologyAccessorInner; - - fn deref(&self) -> &Self::Target { - &self.permit - } -} - -impl<'a> TopologyReadPermit<'a> { - /// Using provided topology read permit, tries to get an immutable reference to the underlying - /// topology. For obvious reasons the lifetime of the topology reference is bound to the permit. - pub(super) fn try_get_valid_topology_ref( - &'a self, - ack_recipient: &Recipient, - packet_recipient: Option<&Recipient>, - ) -> Option<&'a NymTopology> { - // Note: implicit deref with Deref for TopologyReadPermit is happening here - let topology_ref_option = self.permit.as_ref(); - match topology_ref_option { - None => None, - Some(topology_ref) => { - // see if it's possible to route the packet to both gateways - if !topology_ref.can_construct_path_through(DEFAULT_NUM_MIX_HOPS) - || !topology_ref.gateway_exists(&ack_recipient.gateway()) - || if let Some(packet_recipient) = packet_recipient { - !topology_ref.gateway_exists(&packet_recipient.gateway()) - } else { - false - } - { - None - } else { - Some(topology_ref) - } - } - } - } -} - -impl<'a> From> for TopologyReadPermit<'a> { - fn from(read_permit: RwLockReadGuard<'a, TopologyAccessorInner>) -> Self { - TopologyReadPermit { - permit: read_permit, - } - } -} - -#[derive(Clone, Debug)] -pub(crate) struct TopologyAccessor { - // `RwLock` *seems to* be the better approach for this as write access is only requested every - // few seconds, while reads are needed every single packet generated. - // However, proper benchmarks will be needed to determine if `RwLock` is indeed a better - // approach than a `Mutex` - inner: Arc>, -} - -impl TopologyAccessor { - pub(crate) fn new() -> Self { - TopologyAccessor { - inner: Arc::new(RwLock::new(TopologyAccessorInner::new())), - } - } - - pub(super) async fn get_read_permit(&self) -> TopologyReadPermit<'_> { - self.inner.read().await.into() - } - - async fn update_global_topology(&mut self, new_topology: Option) { - self.inner.write().await.update(new_topology); - } - - // only used by the client at startup to get a slightly more reasonable error message - // (currently displays as unused because health checker is disabled due to required changes) - pub(crate) async fn is_routable(&self) -> bool { - match &self.inner.read().await.0 { - None => false, - Some(ref topology) => topology.can_construct_path_through(DEFAULT_NUM_MIX_HOPS), - } - } -} - -pub(crate) struct TopologyRefresherConfig { - directory_server: String, - refresh_rate: time::Duration, -} - -impl TopologyRefresherConfig { - pub(crate) fn new(directory_server: String, refresh_rate: time::Duration) -> Self { - TopologyRefresherConfig { - directory_server, - refresh_rate, - } - } -} - -pub(crate) struct TopologyRefresher { - directory_client: directory_client::Client, - topology_accessor: TopologyAccessor, - refresh_rate: Duration, -} - -impl TopologyRefresher { - pub(crate) fn new_directory_client( - cfg: TopologyRefresherConfig, - topology_accessor: TopologyAccessor, - ) -> Self { - let directory_client_config = directory_client::Config::new(cfg.directory_server); - let directory_client = directory_client::Client::new(directory_client_config); - - TopologyRefresher { - directory_client, - topology_accessor, - refresh_rate: cfg.refresh_rate, - } - } - - async fn get_current_compatible_topology(&self) -> Option { - match self.directory_client.get_topology().await { - Err(err) => { - error!("failed to get network topology! - {:?}", err); - None - } - Ok(topology) => { - let nym_topology: NymTopology = topology.try_into().ok()?; - Some(nym_topology.filter_system_version(built_info::PKG_VERSION)) - } - } - } - - pub(crate) async fn refresh(&mut self) { - trace!("Refreshing the topology"); - let new_topology = self.get_current_compatible_topology().await; - - self.topology_accessor - .update_global_topology(new_topology) - .await; - } - - pub(crate) async fn is_topology_routable(&self) -> bool { - self.topology_accessor.is_routable().await - } - - pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { - handle.spawn(async move { - loop { - tokio::time::delay_for(self.refresh_rate).await; - self.refresh().await; - } - }) - } -} diff --git a/clients/socks5/src/commands/init.rs b/clients/socks5/src/commands/init.rs index f6f7da0c9b..4200a9b1ba 100644 --- a/clients/socks5/src/commands/init.rs +++ b/clients/socks5/src/commands/init.rs @@ -13,10 +13,10 @@ // limitations under the License. use crate::built_info; -use crate::client::key_manager::KeyManager; use crate::commands::override_config; -use crate::config::persistence::key_pathfinder::ClientKeyPathfinder; use clap::{App, Arg, ArgMatches}; +use client_core::client::key_manager::KeyManager; +use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder; use config::NymConfig; use crypto::asymmetric::identity; use directory_client::DirectoryClient; @@ -149,7 +149,7 @@ pub fn execute(matches: &ArgMatches) { println!("Initialising client..."); let id = matches.value_of("id").unwrap(); // required for now - let mut config = crate::config::Config::new(id); + let mut config = client_core::config::Config::new(id); let mut rng = OsRng; config = override_config(config, matches); diff --git a/clients/socks5/src/commands/mod.rs b/clients/socks5/src/commands/mod.rs index f9bb333966..4eba736317 100644 --- a/clients/socks5/src/commands/mod.rs +++ b/clients/socks5/src/commands/mod.rs @@ -12,8 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::config::{Config, SocketType}; use clap::ArgMatches; +use client_core::config::{Config, SocketType}; pub mod init; pub mod run; diff --git a/clients/socks5/src/commands/run.rs b/clients/socks5/src/commands/run.rs index 83af38944e..464b3d962c 100644 --- a/clients/socks5/src/commands/run.rs +++ b/clients/socks5/src/commands/run.rs @@ -14,8 +14,8 @@ use crate::client::NymClient; use crate::commands::override_config; -use crate::config::Config; use clap::{App, Arg, ArgMatches}; +use client_core::config::Config; use config::NymConfig; pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> { diff --git a/clients/socks5/src/config/mod.rs b/clients/socks5/src/config/mod.rs deleted file mode 100644 index 4ccb1606f5..0000000000 --- a/clients/socks5/src/config/mod.rs +++ /dev/null @@ -1,503 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::config::template::config_template; -use config::NymConfig; -use serde::{Deserialize, Serialize}; -use std::path::PathBuf; -use std::time; - -pub mod persistence; -mod template; - -// 'CLIENT' -const DEFAULT_LISTENING_PORT: u16 = 1977; -const DEFAULT_DIRECTORY_SERVER: &str = "https://directory.nymtech.net"; -// 'DEBUG' -// where applicable, the below are defined in milliseconds -const DEFAULT_ACK_WAIT_MULTIPLIER: f64 = 1.5; -const DEFAULT_ACK_WAIT_ADDITION: u64 = 800; -const DEFAULT_LOOP_COVER_STREAM_AVERAGE_DELAY: u64 = 1000; // 1s -const DEFAULT_MESSAGE_STREAM_AVERAGE_DELAY: u64 = 500; // 0.5s -const DEFAULT_AVERAGE_PACKET_DELAY: u64 = 200; // 0.2s -const DEFAULT_TOPOLOGY_REFRESH_RATE: u64 = 30_000; // 30s -const DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT: u64 = 5_000; // 5s - -const DEFAULT_GATEWAY_RESPONSE_TIMEOUT: u64 = 1_500; // 1.5s - -#[derive(Debug, Deserialize, PartialEq, Serialize, Clone, Copy)] -#[serde(deny_unknown_fields)] -pub enum SocketType { - WebSocket, - None, -} - -impl SocketType { - pub fn from_string>(val: S) -> Self { - let mut upper = val.into(); - upper.make_ascii_uppercase(); - match upper.as_ref() { - "WEBSOCKET" | "WS" => SocketType::WebSocket, - _ => SocketType::None, - } - } -} - -#[derive(Debug, Default, Deserialize, PartialEq, Serialize)] -#[serde(deny_unknown_fields)] -pub struct Config { - client: Client, - socket: Socket, - - #[serde(default)] - logging: Logging, - #[serde(default)] - debug: Debug, -} - -impl NymConfig for Config { - fn template() -> &'static str { - config_template() - } - - fn config_file_name() -> String { - "config.toml".to_string() - } - - fn default_root_directory() -> PathBuf { - dirs::home_dir() - .expect("Failed to evaluate $HOME value") - .join(".nym") - .join("clients") - } - - fn root_directory(&self) -> PathBuf { - self.client.nym_root_directory.clone() - } - - fn config_directory(&self) -> PathBuf { - self.client - .nym_root_directory - .join(&self.client.id) - .join("config") - } - - fn data_directory(&self) -> PathBuf { - self.client - .nym_root_directory - .join(&self.client.id) - .join("data") - } -} - -impl Config { - pub fn new>(id: S) -> Self { - Config::default().with_id(id) - } - - // builder methods - pub fn with_id>(mut self, id: S) -> Self { - let id = id.into(); - - // identity key setting - if self.client.private_identity_key_file.as_os_str().is_empty() { - self.client.private_identity_key_file = - self::Client::default_private_identity_key_file(&id); - } - if self.client.public_identity_key_file.as_os_str().is_empty() { - self.client.public_identity_key_file = - self::Client::default_public_identity_key_file(&id); - } - - // encryption key setting - if self - .client - .private_encryption_key_file - .as_os_str() - .is_empty() - { - self.client.private_encryption_key_file = - self::Client::default_private_encryption_key_file(&id); - } - if self - .client - .public_encryption_key_file - .as_os_str() - .is_empty() - { - self.client.public_encryption_key_file = - self::Client::default_public_encryption_key_file(&id); - } - - // shared gateway key setting - if self.client.gateway_shared_key_file.as_os_str().is_empty() { - self.client.gateway_shared_key_file = - self::Client::default_gateway_shared_key_file(&id); - } - - // ack key setting - if self.client.ack_key_file.as_os_str().is_empty() { - self.client.ack_key_file = self::Client::default_ack_key_file(&id); - } - - if self - .client - .reply_encryption_key_store_path - .as_os_str() - .is_empty() - { - self.client.reply_encryption_key_store_path = - self::Client::default_reply_encryption_key_store_path(&id); - } - - self.client.id = id; - self - } - - pub fn with_gateway_id>(mut self, id: S) -> Self { - self.client.gateway_id = id.into(); - self - } - - pub fn with_gateway_listener>(mut self, gateway_listener: S) -> Self { - self.client.gateway_listener = gateway_listener.into(); - self - } - - pub fn with_custom_directory>(mut self, directory_server: S) -> Self { - self.client.directory_server = directory_server.into(); - self - } - - pub fn with_socket(mut self, socket_type: SocketType) -> Self { - self.socket.socket_type = socket_type; - self - } - - pub fn with_port(mut self, port: u16) -> Self { - self.socket.listening_port = port; - self - } - - pub fn set_high_default_traffic_volume(mut self) -> Self { - self.debug.average_packet_delay = 10; - self.debug.loop_cover_traffic_average_delay = 20; // 50 cover messages / s - self.debug.message_sending_average_delay = 5; // 200 "real" messages / s - self - } - - // getters - pub fn get_config_file_save_location(&self) -> PathBuf { - self.config_directory().join(Self::config_file_name()) - } - - pub fn get_private_identity_key_file(&self) -> PathBuf { - self.client.private_identity_key_file.clone() - } - - pub fn get_public_identity_key_file(&self) -> PathBuf { - self.client.public_identity_key_file.clone() - } - - pub fn get_private_encryption_key_file(&self) -> PathBuf { - self.client.private_encryption_key_file.clone() - } - - pub fn get_public_encryption_key_file(&self) -> PathBuf { - self.client.public_encryption_key_file.clone() - } - - pub fn get_gateway_shared_key_file(&self) -> PathBuf { - self.client.gateway_shared_key_file.clone() - } - - pub fn get_reply_encryption_key_store_path(&self) -> PathBuf { - self.client.reply_encryption_key_store_path.clone() - } - - pub fn get_ack_key_file(&self) -> PathBuf { - self.client.ack_key_file.clone() - } - - pub fn get_directory_server(&self) -> String { - self.client.directory_server.clone() - } - - pub fn get_gateway_id(&self) -> String { - self.client.gateway_id.clone() - } - - pub fn get_gateway_listener(&self) -> String { - self.client.gateway_listener.clone() - } - - pub fn get_socket_type(&self) -> SocketType { - self.socket.socket_type - } - - pub fn get_listening_port(&self) -> u16 { - self.socket.listening_port - } - - // Debug getters - pub fn get_average_packet_delay(&self) -> time::Duration { - time::Duration::from_millis(self.debug.average_packet_delay) - } - - pub fn get_average_ack_delay(&self) -> time::Duration { - time::Duration::from_millis(self.debug.average_ack_delay) - } - - pub fn get_ack_wait_multiplier(&self) -> f64 { - self.debug.ack_wait_multiplier - } - - pub fn get_ack_wait_addition(&self) -> time::Duration { - time::Duration::from_millis(self.debug.ack_wait_addition) - } - - pub fn get_loop_cover_traffic_average_delay(&self) -> time::Duration { - time::Duration::from_millis(self.debug.loop_cover_traffic_average_delay) - } - - pub fn get_message_sending_average_delay(&self) -> time::Duration { - time::Duration::from_millis(self.debug.message_sending_average_delay) - } - - pub fn get_gateway_response_timeout(&self) -> time::Duration { - time::Duration::from_millis(self.debug.gateway_response_timeout) - } - - pub fn get_topology_refresh_rate(&self) -> time::Duration { - time::Duration::from_millis(self.debug.topology_refresh_rate) - } - - pub fn get_topology_resolution_timeout(&self) -> time::Duration { - time::Duration::from_millis(self.debug.topology_resolution_timeout) - } -} - -#[derive(Debug, Deserialize, PartialEq, Serialize)] -#[serde(deny_unknown_fields)] -pub struct Client { - /// ID specifies the human readable ID of this particular client. - id: String, - - /// URL to the directory server. - directory_server: String, - - /// Path to file containing private identity key. - private_identity_key_file: PathBuf, - - /// Path to file containing public identity key. - public_identity_key_file: PathBuf, - - /// Path to file containing private encryption key. - private_encryption_key_file: PathBuf, - - /// Path to file containing public encryption key. - public_encryption_key_file: PathBuf, - - /// Path to file containing shared key derived with the specified gateway that is used - /// for all communication with it. - gateway_shared_key_file: PathBuf, - - /// Path to file containing key used for encrypting and decrypting the content of an - /// acknowledgement so that nobody besides the client knows which packet it refers to. - ack_key_file: PathBuf, - - /// Full path to file containing reply encryption keys of all reply-SURBs we have ever - /// sent but not received back. - reply_encryption_key_store_path: PathBuf, - - /// gateway_id specifies ID of the gateway to which the client should send messages. - /// If initially omitted, a random gateway will be chosen from the available topology. - gateway_id: String, - - /// Address of the gateway listener to which all client requests should be sent. - gateway_listener: String, - - /// nym_home_directory specifies absolute path to the home nym Clients directory. - /// It is expected to use default value and hence .toml file should not redefine this field. - nym_root_directory: PathBuf, -} - -impl Default for Client { - fn default() -> Self { - // there must be explicit checks for whether id is not empty later - Client { - id: "".to_string(), - directory_server: DEFAULT_DIRECTORY_SERVER.to_string(), - private_identity_key_file: Default::default(), - public_identity_key_file: Default::default(), - private_encryption_key_file: Default::default(), - public_encryption_key_file: Default::default(), - gateway_shared_key_file: Default::default(), - ack_key_file: Default::default(), - reply_encryption_key_store_path: Default::default(), - gateway_id: "".to_string(), - gateway_listener: "".to_string(), - nym_root_directory: Config::default_root_directory(), - } - } -} - -impl Client { - fn default_private_identity_key_file(id: &str) -> PathBuf { - Config::default_data_directory(Some(id)).join("private_identity.pem") - } - - fn default_public_identity_key_file(id: &str) -> PathBuf { - Config::default_data_directory(Some(id)).join("public_identity.pem") - } - - fn default_private_encryption_key_file(id: &str) -> PathBuf { - Config::default_data_directory(Some(id)).join("private_encryption.pem") - } - - fn default_public_encryption_key_file(id: &str) -> PathBuf { - Config::default_data_directory(Some(id)).join("public_encryption.pem") - } - - fn default_gateway_shared_key_file(id: &str) -> PathBuf { - Config::default_data_directory(Some(id)).join("gateway_shared.pem") - } - - fn default_ack_key_file(id: &str) -> PathBuf { - Config::default_data_directory(Some(id)).join("ack_key.pem") - } - - fn default_reply_encryption_key_store_path(id: &str) -> PathBuf { - Config::default_data_directory(Some(id)).join("reply_key_store") - } -} - -#[derive(Debug, Deserialize, PartialEq, Serialize)] -#[serde(deny_unknown_fields)] -pub struct Socket { - socket_type: SocketType, - listening_port: u16, -} - -impl Default for Socket { - fn default() -> Self { - Socket { - socket_type: SocketType::WebSocket, - listening_port: DEFAULT_LISTENING_PORT, - } - } -} - -#[derive(Debug, Deserialize, PartialEq, Serialize)] -#[serde(deny_unknown_fields)] -pub struct Logging {} - -impl Default for Logging { - fn default() -> Self { - Logging {} - } -} - -#[derive(Debug, Deserialize, PartialEq, Serialize)] -#[serde(default, deny_unknown_fields)] -pub struct Debug { - /// The parameter of Poisson distribution determining how long, on average, - /// sent packet is going to be delayed at any given mix node. - /// So for a packet going through three mix nodes, on average, it will take three times this value - /// until the packet reaches its destination. - /// The provided value is interpreted as milliseconds. - average_packet_delay: u64, - - /// The parameter of Poisson distribution determining how long, on average, - /// sent acknowledgement is going to be delayed at any given mix node. - /// So for an ack going through three mix nodes, on average, it will take three times this value - /// until the packet reaches its destination. - /// The provided value is interpreted as milliseconds. - average_ack_delay: u64, - - /// Value multiplied with the expected round trip time of an acknowledgement packet before - /// it is assumed it was lost and retransmission of the data packet happens. - /// In an ideal network with 0 latency, this value would have been 1. - ack_wait_multiplier: f64, - - /// Value added to the expected round trip time of an acknowledgement packet before - /// it is assumed it was lost and retransmission of the data packet happens. - /// In an ideal network with 0 latency, this value would have been 0. - /// The provided value is interpreted as milliseconds. - ack_wait_addition: u64, - - /// The parameter of Poisson distribution determining how long, on average, - /// it is going to take for another loop cover traffic message to be sent. - /// The provided value is interpreted as milliseconds. - loop_cover_traffic_average_delay: u64, - - /// The parameter of Poisson distribution determining how long, on average, - /// it is going to take another 'real traffic stream' message to be sent. - /// If no real packets are available and cover traffic is enabled, - /// a loop cover message is sent instead in order to preserve the rate. - /// The provided value is interpreted as milliseconds. - message_sending_average_delay: u64, - - /// How long we're willing to wait for a response to a message sent to the gateway, - /// before giving up on it. - /// The provided value is interpreted as milliseconds. - gateway_response_timeout: u64, - - /// The uniform delay every which clients are querying the directory server - /// to try to obtain a compatible network topology to send sphinx packets through. - /// The provided value is interpreted as milliseconds. - topology_refresh_rate: u64, - - /// During topology refresh, test packets are sent through every single possible network - /// path. This timeout determines waiting period until it is decided that the packet - /// did not reach its destination. - /// The provided value is interpreted as milliseconds. - topology_resolution_timeout: u64, -} - -impl Default for Debug { - fn default() -> Self { - Debug { - average_packet_delay: DEFAULT_AVERAGE_PACKET_DELAY, - average_ack_delay: DEFAULT_AVERAGE_PACKET_DELAY, - ack_wait_multiplier: DEFAULT_ACK_WAIT_MULTIPLIER, - ack_wait_addition: DEFAULT_ACK_WAIT_ADDITION, - loop_cover_traffic_average_delay: DEFAULT_LOOP_COVER_STREAM_AVERAGE_DELAY, - message_sending_average_delay: DEFAULT_MESSAGE_STREAM_AVERAGE_DELAY, - gateway_response_timeout: DEFAULT_GATEWAY_RESPONSE_TIMEOUT, - topology_refresh_rate: DEFAULT_TOPOLOGY_REFRESH_RATE, - topology_resolution_timeout: DEFAULT_TOPOLOGY_RESOLUTION_TIMEOUT, - } - } -} - -#[cfg(test)] -mod client_config { - use super::*; - - #[test] - fn after_saving_default_config_the_loaded_one_is_identical() { - // need to figure out how to do something similar but without touching the disk - // or the file system at all... - let temp_location = tempfile::tempdir().unwrap().path().join("config.toml"); - let default_config = Config::default().with_id("foomp".to_string()); - default_config - .save_to_file(Some(temp_location.clone())) - .unwrap(); - - let loaded_config = Config::load_from_file(Some(temp_location), None).unwrap(); - - assert_eq!(default_config, loaded_config); - } -} diff --git a/clients/socks5/src/config/persistence/key_pathfinder.rs b/clients/socks5/src/config/persistence/key_pathfinder.rs deleted file mode 100644 index 3804c38fea..0000000000 --- a/clients/socks5/src/config/persistence/key_pathfinder.rs +++ /dev/null @@ -1,76 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::config::Config; -use std::path::{Path, PathBuf}; - -#[derive(Debug)] -pub struct ClientKeyPathfinder { - identity_private_key: PathBuf, - identity_public_key: PathBuf, - encryption_private_key: PathBuf, - encryption_public_key: PathBuf, - gateway_shared_key: PathBuf, - ack_key: PathBuf, -} - -impl ClientKeyPathfinder { - pub fn new(id: String) -> Self { - let os_config_dir = dirs::config_dir().expect("no config directory known for this OS"); // grabs the OS default config dir - let config_dir = os_config_dir.join("nym").join("clients").join(id); - ClientKeyPathfinder { - identity_private_key: config_dir.join("private_identity.pem"), - identity_public_key: config_dir.join("public_identity.pem"), - encryption_private_key: config_dir.join("public_encryption.pem"), - encryption_public_key: config_dir.join("private_encryption.pem"), - gateway_shared_key: config_dir.join("gateway_shared.pem"), - ack_key: config_dir.join("ack_key.pem"), - } - } - - pub fn new_from_config(config: &Config) -> Self { - ClientKeyPathfinder { - identity_private_key: config.get_private_identity_key_file(), - identity_public_key: config.get_public_identity_key_file(), - encryption_private_key: config.get_private_encryption_key_file(), - encryption_public_key: config.get_public_encryption_key_file(), - gateway_shared_key: config.get_gateway_shared_key_file(), - ack_key: config.get_ack_key_file(), - } - } - - pub fn private_identity_key(&self) -> &Path { - &self.identity_private_key - } - - pub fn public_identity_key(&self) -> &Path { - &self.identity_public_key - } - - pub fn private_encryption_key(&self) -> &Path { - &self.encryption_private_key - } - - pub fn public_encryption_key(&self) -> &Path { - &self.encryption_public_key - } - - pub fn gateway_shared_key(&self) -> &Path { - &self.gateway_shared_key - } - - pub fn ack_key(&self) -> &Path { - &self.ack_key - } -} diff --git a/clients/socks5/src/config/template.rs b/clients/socks5/src/config/template.rs deleted file mode 100644 index fbe0f404c2..0000000000 --- a/clients/socks5/src/config/template.rs +++ /dev/null @@ -1,102 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -pub(crate) fn config_template() -> &'static str { - // While using normal toml marshalling would have been way simpler with less overhead, - // I think it's useful to have comments attached to the saved config file to explain behaviour of - // particular fields. - // Note: any changes to the template must be reflected in the appropriate structs in mod.rs. - r#" -# This is a TOML config file. -# For more information, see https://github.com/toml-lang/toml - -##### main base client config options ##### - -[client] -# Human readable ID of this particular client. -id = '{{ client.id }}' - -# URL to the directory server. -directory_server = '{{ client.directory_server }}' - -# Path to file containing private identity key. -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 }}' - -# Path to file containing private encryption key. -private_encryption_key_file = '{{ client.private_encryption_key_file }}' - -# Path to file containing public encryption key. -public_encryption_key_file = '{{ client.public_encryption_key_file }}' - -# Full path to file containing reply encryption keys of all reply-SURBs we have ever -# sent but not received back. -reply_encryption_key_store_path = '{{ client.reply_encryption_key_store_path }}' - -##### additional client config options ##### - -# ID of the gateway from which the client should be fetching messages. -gateway_id = '{{ client.gateway_id }}' - -# Address of the gateway listener to which all client requests should be sent. -gateway_listener = '{{ client.gateway_listener }}' - -# A gateway specific, optional, base58 stringified shared key used for -# communication with particular gateway. -gateway_shared_key_file = '{{ client.gateway_shared_key_file }}' - -# Path to file containing key used for encrypting and decrypting the content of an -# acknowledgement so that nobody besides the client knows which packet it refers to. -ack_key_file = '{{ client.ack_key_file }}' - -##### advanced configuration options ##### - -# Absolute path to the home Nym Clients directory. -nym_root_directory = '{{ client.nym_root_directory }}' - - -##### socket config options ##### - -[socket] - -# allowed values are 'WebSocket' or 'None' -socket_type = '{{ socket.socket_type }}' - -# if applicable (for the case of 'WebSocket'), the port on which the client -# will be listening for incoming requests -listening_port = {{ socket.listening_port }} - - -##### logging configuration options ##### - -[logging] - -# TODO - - -##### debug configuration options ##### -# The following options should not be modified unless you know EXACTLY what you are doing -# as if set incorrectly, they may impact your anonymity. - -[debug] - -average_packet_delay = {{ debug.average_packet_delay }} -average_ack_delay = {{ debug.average_ack_delay }} -loop_cover_traffic_average_delay = {{ debug.loop_cover_traffic_average_delay }} -message_sending_average_delay = {{ debug.message_sending_average_delay }} - -"# -} diff --git a/clients/socks5/src/lib.rs b/clients/socks5/src/lib.rs index 8efb3dd47a..694607032a 100644 --- a/clients/socks5/src/lib.rs +++ b/clients/socks5/src/lib.rs @@ -13,5 +13,4 @@ // limitations under the License. pub mod built_info; pub mod client; -pub mod config; pub mod socks; diff --git a/clients/socks5/src/main.rs b/clients/socks5/src/main.rs index 3e2aac76ec..dfcbc72f15 100644 --- a/clients/socks5/src/main.rs +++ b/clients/socks5/src/main.rs @@ -17,7 +17,6 @@ use clap::{App, ArgMatches}; pub mod built_info; pub mod client; mod commands; -pub mod config; pub mod socks; fn main() { diff --git a/clients/socks5/src/socks/client.rs b/clients/socks5/src/socks/client.rs index 242bb0ac09..44acb25777 100644 --- a/clients/socks5/src/socks/client.rs +++ b/clients/socks5/src/socks/client.rs @@ -9,8 +9,8 @@ use tokio::{self, net::TcpStream}; use nymsphinx::addressing::clients::Recipient; -use crate::client::inbound_messages::InputMessage; -use crate::client::inbound_messages::InputMessageSender; +use client_core::client::inbound_messages::InputMessage; +use client_core::client::inbound_messages::InputMessageSender; use futures::{channel::oneshot, lock::Mutex}; use super::authentication::{AuthenticationMethods, Authenticator, User}; diff --git a/clients/socks5/src/socks/mixnet_responses.rs b/clients/socks5/src/socks/mixnet_responses.rs index 3b785ec8be..55e3744f73 100644 --- a/clients/socks5/src/socks/mixnet_responses.rs +++ b/clients/socks5/src/socks/mixnet_responses.rs @@ -1,6 +1,6 @@ use super::client::ActiveStreams; -use crate::client::received_buffer::ReconstructedMessagesReceiver; -use crate::client::received_buffer::{ReceivedBufferMessage, ReceivedBufferRequestSender}; +use client_core::client::received_buffer::ReconstructedMessagesReceiver; +use client_core::client::received_buffer::{ReceivedBufferMessage, ReceivedBufferRequestSender}; use futures::channel::mpsc; use futures::StreamExt; use log::*; diff --git a/clients/socks5/src/socks/server.rs b/clients/socks5/src/socks/server.rs index 1c36e429e4..1118ad3cdd 100644 --- a/clients/socks5/src/socks/server.rs +++ b/clients/socks5/src/socks/server.rs @@ -4,7 +4,7 @@ use super::{ mixnet_responses::MixnetResponseListener, types::{ResponseCode, SocksProxyError}, }; -use crate::client::{ +use client_core::client::{ inbound_messages::InputMessageSender, received_buffer::ReceivedBufferRequestSender, }; use futures::lock::Mutex;