From bbb3287029eea76b1082ea6eae64ccd78f168901 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 20 Feb 2023 13:00:34 +0100 Subject: [PATCH] Add client functionality into nym-network-requester (#2972) * network-requester: replace websocket with mixnet client * network-requester: mini tidy * network-requester: add info line about own address * network-requester: update to nym-crypto rename --- Cargo.lock | 8 + .../real_traffic_stream.rs | 2 - clients/socks5/src/socks/server.rs | 4 +- common/client-connections/src/lib.rs | 6 - .../src/connection_controller.rs | 28 +-- .../network-requester/Cargo.toml | 19 +- .../network-requester/src/cli/init.rs | 157 ++++++++++++ .../network-requester/src/cli/mod.rs | 106 ++++++++ .../network-requester/src/cli/run.rs | 157 ++++++++++++ .../network-requester/src/config/mod.rs | 128 ++++++++++ .../network-requester/src/config/template.rs | 99 ++++++++ .../network-requester/src/core.rs | 231 ++++++------------ .../network-requester/src/error.rs | 24 +- .../network-requester/src/main.rs | 104 +------- .../network-requester/src/reply.rs | 21 +- .../network-requester/src/websocket.rs | 34 --- 16 files changed, 781 insertions(+), 347 deletions(-) create mode 100644 service-providers/network-requester/src/cli/init.rs create mode 100644 service-providers/network-requester/src/cli/mod.rs create mode 100644 service-providers/network-requester/src/cli/run.rs create mode 100644 service-providers/network-requester/src/config/mod.rs create mode 100644 service-providers/network-requester/src/config/template.rs delete mode 100644 service-providers/network-requester/src/websocket.rs diff --git a/Cargo.lock b/Cargo.lock index 0dcda65a7a..c5801327f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3724,13 +3724,18 @@ dependencies = [ "build-information", "clap 4.1.4", "client-connections", + "client-core", "completions", + "config", "dirs", "futures", "ipnetwork 0.20.0", + "lazy_static", "log", "logging", "network-defaults", + "nym-crypto", + "nym-sdk", "nymsphinx", "ordered-buffer", "pretty_env_logger", @@ -3743,10 +3748,13 @@ dependencies = [ "socks5-requests", "sqlx 0.6.2", "statistics-common", + "tap", "task", "thiserror", "tokio", "tokio-tungstenite 0.17.2", + "url", + "version-checker", "websocket-requests", ] diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index d601e9c24a..769233f926 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -342,7 +342,6 @@ where if let Poll::Ready(Some(id)) = Pin::new(&mut self.client_connection_rx).poll_next(cx) { match id { ConnectionCommand::Close(id) => self.on_close_connection(id), - ConnectionCommand::ActiveConnections(_) => panic!(), } } @@ -421,7 +420,6 @@ where if let Poll::Ready(Some(id)) = Pin::new(&mut self.client_connection_rx).poll_next(cx) { match id { ConnectionCommand::Close(id) => self.on_close_connection(id), - ConnectionCommand::ActiveConnections(_) => panic!(), } } diff --git a/clients/socks5/src/socks/server.rs b/clients/socks5/src/socks/server.rs index 2d4f01f8c5..3f872a5877 100644 --- a/clients/socks5/src/socks/server.rs +++ b/clients/socks5/src/socks/server.rs @@ -10,7 +10,7 @@ use client_core::client::{ }; use log::*; use nymsphinx::addressing::clients::Recipient; -use proxy_helpers::connection_controller::{BroadcastActiveConnections, Controller}; +use proxy_helpers::connection_controller::Controller; use std::net::SocketAddr; use tap::TapFallible; use task::TaskClient; @@ -69,7 +69,7 @@ impl SphinxSocksServer { // controller for managing all active connections let (mut active_streams_controller, controller_sender) = Controller::new( client_connection_tx, - BroadcastActiveConnections::Off, + //BroadcastActiveConnections::Off, self.shutdown.clone(), ); tokio::spawn(async move { diff --git a/common/client-connections/src/lib.rs b/common/client-connections/src/lib.rs index b10963da79..af13111526 100644 --- a/common/client-connections/src/lib.rs +++ b/common/client-connections/src/lib.rs @@ -25,12 +25,6 @@ pub enum ConnectionCommand { // Announce that at a connection was closed. E.g the `OutQueueControl` uses this to discard // transmission lanes. Close(ConnectionId), - - // In the network requester for example, we usually want to broadcast active connections - // regularly, so we know what connections we need to request lane queue lengths for from the - // client. - // In the socks5-client, this is not needed since have direct access to the lane queue lengths. - ActiveConnections(Vec), } // The `OutQueueControl` publishes the backlog per lane, primarily so that upstream can slow down diff --git a/common/socks5/proxy-helpers/src/connection_controller.rs b/common/socks5/proxy-helpers/src/connection_controller.rs index 98eb12e9a9..2804d4e370 100644 --- a/common/socks5/proxy-helpers/src/connection_controller.rs +++ b/common/socks5/proxy-helpers/src/connection_controller.rs @@ -7,12 +7,8 @@ use futures::StreamExt; use log::*; use ordered_buffer::{OrderedMessage, OrderedMessageBuffer, ReadContiguousData}; use socks5_requests::{ConnectionId, NetworkData, SendRequest}; -use std::{ - collections::{HashMap, HashSet}, - time::Duration, -}; +use std::collections::{HashMap, HashSet}; use task::TaskClient; -use tokio::time; /// A generic message produced after reading from a socket/connection. It includes data that was /// actually read alongside boolean indicating whether the connection got closed so that @@ -116,10 +112,6 @@ pub struct Controller { // Broadcast closed connections client_connection_tx: ConnectionCommandSender, - // The controller can broadcast active connections. This is useful in the network-requester - // where its used to query the client for lane queue lengths - broadcast_connections: BroadcastActiveConnections, - // TODO: this can potentially be abused to ddos and kill provider. Not sure at this point // how to handle it more gracefully @@ -133,7 +125,6 @@ pub struct Controller { impl Controller { pub fn new( client_connection_tx: ConnectionCommandSender, - broadcast_connections: BroadcastActiveConnections, shutdown: TaskClient, ) -> (Self, ControllerSender) { let (sender, receiver) = mpsc::unbounded(); @@ -143,7 +134,6 @@ impl Controller { receiver, recently_closed: HashSet::new(), client_connection_tx, - broadcast_connections, pending_messages: HashMap::new(), shutdown, }, @@ -194,15 +184,6 @@ impl Controller { } } - fn broadcast_active_connections(&mut self) { - // What about the recently closed ones? Hopefully we can ignore them ... - let conn_ids = self.active_connections.keys().copied().collect(); - - self.client_connection_tx - .unbounded_send(ConnectionCommand::ActiveConnections(conn_ids)) - .unwrap(); - } - fn send_to_connection(&mut self, conn_id: ConnectionId, payload: Vec, is_closed: bool) { if let Some(active_connection) = self.active_connections.get_mut(&conn_id) { if !payload.is_empty() { @@ -259,8 +240,6 @@ impl Controller { } pub async fn run(&mut self) { - let mut interval = time::interval(Duration::from_millis(500)); - loop { tokio::select! { command = self.receiver.next() => match command { @@ -276,11 +255,6 @@ impl Controller { break; } }, - _ = interval.tick() => { - if self.broadcast_connections == BroadcastActiveConnections::On { - self.broadcast_active_connections(); - } - }, } } self.shutdown.recv_timeout().await; diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index a28693c46f..88d0b8ad41 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/service-providers/network-requester/Cargo.toml @@ -4,18 +4,19 @@ [package] name = "nym-network-requester" version = "1.1.9" -authors = ["Dave Hrycyszyn ", "Jędrzej Stuczyński "] -edition = "2021" +authors.workspace = true +edition.workspace = true rust-version = "1.65" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] async-trait = { version = "0.1.51" } -clap = {version = "4.0", features = ["derive"]} +clap = {version = "4.0", features = ["cargo", "derive"]} dirs = "4.0" futures = "0.3.24" ipnetwork = "0.20.0" +lazy_static = { workspace = true } log = { workspace = true } pretty_env_logger = "0.4.0" publicsuffix = "1.5" # Can't update this until bip updates to support newer idna version @@ -23,22 +24,28 @@ rand = "0.7.3" reqwest = { version = "0.11.11", features = ["json"] } serde = { version = "1.0", features = ["derive"] } sqlx = { version = "0.6.1", features = ["runtime-tokio-rustls", "chrono"]} +tap = { workspace = true } thiserror = "1.0" tokio = { version = "1.24.1", features = [ "net", "rt-multi-thread", "macros" ] } tokio-tungstenite = "0.17.2" - +url = { workspace = true } # internal build-information = { path = "../../common/build-information" } client-connections = { path = "../../common/client-connections" } +client-core = { path = "../../clients/client-core" } completions = { path = "../../common/completions" } -network-defaults = { path = "../../common/network-defaults" } -nymsphinx = { path = "../../common/nymsphinx" } +config = { path = "../../common/config" } +nym-crypto = { path = "../../common/crypto" } logging = { path = "../../common/logging"} +network-defaults = { path = "../../common/network-defaults" } +nym-sdk = { path = "../../sdk/rust/nym-sdk" } +nymsphinx = { path = "../../common/nymsphinx" } ordered-buffer = {path = "../../common/socks5/ordered-buffer"} proxy-helpers = { path = "../../common/socks5/proxy-helpers" } service-providers-common = { path = "../common" } socks5-requests = { path = "../../common/socks5/requests" } statistics-common = { path = "../../common/statistics" } task = { path = "../../common/task" } +version-checker = { path = "../../common/version-checker" } websocket-requests = { path = "../../clients/native/websocket-requests" } diff --git a/service-providers/network-requester/src/cli/init.rs b/service-providers/network-requester/src/cli/init.rs new file mode 100644 index 0000000000..ee21450bec --- /dev/null +++ b/service-providers/network-requester/src/cli/init.rs @@ -0,0 +1,157 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::{ + cli::{override_config, OverrideConfig}, + config::Config, + error::NetworkRequesterError, +}; +use clap::Args; +use config::NymConfig; +use nym_crypto::asymmetric::identity; +use nymsphinx::addressing::clients::Recipient; +use serde::Serialize; +use std::fmt::Display; +use tap::TapFallible; + +#[derive(Args, Clone)] +pub(crate) struct Init { + /// Id of the nym-mixnet-client we want to create config for. + #[clap(long)] + id: String, + + /// Id of the gateway we are going to connect to. + #[clap(long)] + gateway: Option, + + /// Force register gateway. WARNING: this will overwrite any existing keys for the given id, + /// potentially causing loss of access. + #[clap(long)] + force_register_gateway: bool, + + /// Comma separated list of rest endpoints of the nyxd validators + #[clap(long, alias = "nymd_validators", value_delimiter = ',')] + nyxd_urls: Option>, + + /// Comma separated list of rest endpoints of the API validators + #[clap(long, alias = "api_validators", value_delimiter = ',')] + // the alias here is included for backwards compatibility (1.1.4 and before) + nym_apis: Option>, + + /// Set this client to work in a enabled credentials mode that would attempt to use gateway + /// with bandwidth credential requirement. + #[clap(long)] + enabled_credentials_mode: Option, + + /// Save a summary of the initialization to a json file + #[clap(long)] + output_json: bool, +} + +impl From for OverrideConfig { + fn from(init_config: Init) -> Self { + OverrideConfig { + nym_apis: init_config.nym_apis, + fastmode: false, + no_cover: false, + + nyxd_urls: init_config.nyxd_urls, + enabled_credentials_mode: init_config.enabled_credentials_mode, + } + } +} + +#[derive(Debug, Serialize)] +pub struct InitResults { + #[serde(flatten)] + client_core: client_core::init::InitResults, +} + +impl InitResults { + fn new(config: &Config, address: &Recipient) -> Self { + Self { + client_core: client_core::init::InitResults::new(config.get_base(), address), + } + } +} + +impl Display for InitResults { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.client_core) + } +} + +pub(crate) async fn execute(args: &Init) -> Result<(), NetworkRequesterError> { + println!("Initialising client..."); + + let id = &args.id; + + let already_init = Config::default_config_file_path(id).exists(); + if already_init { + println!("Client \"{id}\" was already initialised before"); + } + + // Usually you only register with the gateway on the first init, however you can force + // re-registering if wanted. + let user_wants_force_register = args.force_register_gateway; + if user_wants_force_register { + println!("Instructed to force registering gateway. This might overwrite keys!"); + } + + // If the client was already initialized, don't generate new keys and don't re-register with + // the gateway (because this would create a new shared key). + // Unless the user really wants to. + let register_gateway = !already_init || user_wants_force_register; + + // Attempt to use a user-provided gateway, if possible + let user_chosen_gateway_id = args.gateway; + + // Load and potentially override config + let mut config = override_config(Config::new(id), OverrideConfig::from(args.clone())); + + // Setup gateway by either registering a new one, or creating a new config from the selected + // one but with keys kept, or reusing the gateway configuration. + let gateway = client_core::init::setup_gateway_from_config::( + register_gateway, + user_chosen_gateway_id, + config.get_base(), + ) + .await + .map_err(|source| { + eprintln!("Failed to setup gateway\nError: {source}"); + NetworkRequesterError::FailedToSetupGateway { source } + })?; + + config.get_base_mut().set_gateway_endpoint(gateway); + + config.save_to_file(None).tap_err(|_| { + log::error!("Failed to save the config file"); + })?; + + print_saved_config(&config); + + let address = client_core::init::get_client_address_from_stored_keys(config.get_base())?; + let init_results = InitResults::new(&config, &address); + println!("{init_results}"); + + // Output summary to a json file, if specified + if args.output_json { + client_core::init::output_to_json(&init_results, "client_init_results.json"); + } + + println!("\nThe address of this client is: {address}\n"); + Ok(()) +} + +fn print_saved_config(config: &Config) { + let config_save_location = config.get_config_file_save_location(); + println!("Saved configuration file to {config_save_location:?}"); + println!("Using gateway: {}", config.get_base().get_gateway_id()); + log::debug!("Gateway id: {}", config.get_base().get_gateway_id()); + log::debug!("Gateway owner: {}", config.get_base().get_gateway_owner()); + log::debug!( + "Gateway listener: {}", + config.get_base().get_gateway_listener() + ); + println!("Client configuration completed.\n"); +} diff --git a/service-providers/network-requester/src/cli/mod.rs b/service-providers/network-requester/src/cli/mod.rs new file mode 100644 index 0000000000..37471c4fea --- /dev/null +++ b/service-providers/network-requester/src/cli/mod.rs @@ -0,0 +1,106 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use build_information::BinaryBuildInformation; +use clap::{CommandFactory, Parser, Subcommand}; +use completions::{fig_generate, ArgShell}; + +use crate::{ + config::{BaseConfig, Config}, + error::NetworkRequesterError, +}; + +mod init; +mod run; + +lazy_static::lazy_static! { + pub static ref PRETTY_BUILD_INFORMATION: String = + BinaryBuildInformation::new(env!("CARGO_PKG_VERSION")).pretty_print(); +} + +// Helper for passing LONG_VERSION to clap +fn pretty_build_info_static() -> &'static str { + &PRETTY_BUILD_INFORMATION +} + +#[derive(Parser)] +#[clap(author = "Nymtech", version, about, long_version = pretty_build_info_static())] +pub(crate) struct Cli { + /// Path pointing to an env file that configures the client. + #[clap(short, long)] + pub(crate) config_env_file: Option, + + #[clap(subcommand)] + command: Commands, +} + +#[allow(clippy::large_enum_variant)] +#[derive(Subcommand)] +pub(crate) enum Commands { + /// Initialize a network-requester. Do this first! + Init(init::Init), + + /// Run the network requester with the provided configuration and optionally override + /// parameters. + Run(run::Run), + + /// Generate shell completions + Completions(ArgShell), + + /// Generate Fig specification + GenerateFigSpec, +} + +// Configuration that can be overridden. +pub(crate) struct OverrideConfig { + nym_apis: Option>, + fastmode: bool, + no_cover: bool, + nyxd_urls: Option>, + enabled_credentials_mode: Option, +} + +pub(crate) fn override_config(config: Config, args: OverrideConfig) -> Config { + config + .with_base(BaseConfig::with_high_default_traffic_volume, args.fastmode) + .with_base(BaseConfig::with_disabled_cover_traffic, args.no_cover) + .with_optional_custom_env_ext( + BaseConfig::with_custom_nym_apis, + args.nym_apis, + network_defaults::var_names::NYM_API, + config::parse_urls, + ) + .with_optional_custom_env_ext( + BaseConfig::with_custom_nyxd, + args.nyxd_urls, + network_defaults::var_names::NYXD, + config::parse_urls, + ) + .with_optional_ext( + BaseConfig::with_disabled_credentials, + args.enabled_credentials_mode.map(|b| !b), + ) +} + +pub(crate) async fn execute(args: Cli) -> Result<(), NetworkRequesterError> { + let bin_name = "nym-network-requester"; + + match &args.command { + Commands::Init(m) => init::execute(m).await?, + Commands::Run(m) => run::execute(m).await?, + Commands::Completions(s) => s.generate(&mut Cli::command(), bin_name), + Commands::GenerateFigSpec => fig_generate(&mut Cli::command(), bin_name), + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::CommandFactory; + + #[test] + fn verify_cli() { + Cli::command().debug_assert(); + } +} diff --git a/service-providers/network-requester/src/cli/run.rs b/service-providers/network-requester/src/cli/run.rs new file mode 100644 index 0000000000..98678fd910 --- /dev/null +++ b/service-providers/network-requester/src/cli/run.rs @@ -0,0 +1,157 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::{ + cli::{override_config, OverrideConfig}, + config::Config, + error::NetworkRequesterError, +}; +use clap::Args; +use config::NymConfig; +use nymsphinx::addressing::clients::Recipient; + +const ENABLE_STATISTICS: &str = "enable-statistics"; + +#[allow(clippy::struct_excessive_bools)] +#[derive(Args, Clone)] +pub(crate) struct Run { + /// Id of the nym-mixnet-client we want to run. + #[clap(long)] + id: String, + + /// Specifies whether this network requester should run in 'open-proxy' mode + #[clap(long)] + open_proxy: bool, + + /// Enable service anonymized statistics that get sent to a statistics aggregator server + #[clap(long)] + enable_statistics: bool, + + /// Mixnet client address where a statistics aggregator is running. The default value is a Nym + /// aggregator client + #[clap(long)] + statistics_recipient: Option, + + /// Set this client to work in a enabled credentials mode that would attempt to use gateway + /// with bandwidth credential requirement. + #[clap(long)] + enabled_credentials_mode: Option, + + /// Mostly debug-related option to increase default traffic rate so that you would not need to + /// modify config post init + #[clap(long, hide = true)] + fastmode: bool, + + /// Disable loop cover traffic and the Poisson rate limiter (for debugging only) + #[clap(long, hide = true)] + no_cover: bool, +} + +impl From for OverrideConfig { + fn from(run_config: Run) -> Self { + OverrideConfig { + nym_apis: None, + fastmode: run_config.fastmode, + no_cover: run_config.no_cover, + nyxd_urls: None, + enabled_credentials_mode: run_config.enabled_credentials_mode, + } + } +} + +// this only checks compatibility between config the binary. It does not take into consideration +// network version. It might do so in the future. +fn version_check(cfg: &Config) -> bool { + let binary_version = env!("CARGO_PKG_VERSION"); + let config_version = cfg.get_base().get_version(); + if binary_version == config_version { + true + } else { + log::warn!( + "The native-client binary has different version than what is specified \ + in config file! {} and {}", + binary_version, + config_version + ); + if version_checker::is_minor_version_compatible(binary_version, config_version) { + log::info!( + "but they are still semver compatible. \ + However, consider running the `upgrade` command" + ); + true + } else { + log::error!( + "and they are semver incompatible! - \ + please run the `upgrade` command before attempting `run` again" + ); + false + } + } +} + +pub(crate) async fn execute(args: &Run) -> Result<(), NetworkRequesterError> { + if args.open_proxy { + println!( + "\n\nYOU HAVE STARTED IN 'OPEN PROXY' MODE. ANYONE WITH YOUR CLIENT ADDRESS \ + CAN MAKE REQUESTS FROM YOUR MACHINE. PLEASE QUIT IF YOU DON'T UNDERSTAND WHAT \ + YOU'RE DOING.\n\n" + ); + } + + if args.enable_statistics { + println!( + "\n\nTHE NETWORK REQUESTER STATISTICS ARE ENABLED. IT WILL COLLECT AND SEND \ + ANONYMIZED STATISTICS TO A CENTRAL SERVER. PLEASE QUIT IF YOU DON'T WANT \ + THIS TO HAPPEN AND START WITHOUT THE {ENABLE_STATISTICS} FLAG .\n\n" + ); + } + + let id = &args.id; + + let mut config = match Config::load_from_file(id) { + Ok(cfg) => cfg, + Err(err) => { + log::error!( + "Failed to load config for {}. \ + Are you sure you have run `init` before? (Error was: {err})", + id + ); + return Err(NetworkRequesterError::FailedToLoadConfig(id.to_string())); + } + }; + + let override_config_fields = OverrideConfig::from(args.clone()); + config = override_config(config, override_config_fields); + + if config.get_base_mut().set_empty_fields_to_defaults() { + log::warn!( + "Some of the core config options were left unset. \ + The default values are going to get used instead." + ); + } + + if !version_check(&config) { + log::error!("Failed the local version check"); + return Err(NetworkRequesterError::FailedLocalVersionCheck); + } + + // TODO: consider incorporating statistics_recipient, open_proxuy and enable_statistics in + // `Config`. + + let stats_provider_addr = args + .statistics_recipient + .as_ref() + .map(Recipient::try_from_base58_string) + .transpose() + .unwrap_or(None); + + log::info!("Starting socks5 service provider"); + let server = crate::core::NRServiceProviderBuilder::new( + config, + args.open_proxy, + args.enable_statistics, + stats_provider_addr, + ) + .await; + server.run_service_provider().await +} diff --git a/service-providers/network-requester/src/config/mod.rs b/service-providers/network-requester/src/config/mod.rs new file mode 100644 index 0000000000..1dfa767ba9 --- /dev/null +++ b/service-providers/network-requester/src/config/mod.rs @@ -0,0 +1,128 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::config::template::config_template; +use client_core::config::ClientCoreConfigTrait; +use config::{NymConfig, OptionalSet}; +use serde::{Deserialize, Serialize}; +use std::fmt::Debug; +use std::path::PathBuf; +use std::str::FromStr; + +pub use client_core::config::Config as BaseConfig; +pub use client_core::config::MISSING_VALUE; +pub use client_core::config::{DebugConfig, GatewayEndpointConfig}; + +mod template; + +#[derive(Debug, Default, Deserialize, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Config { + #[serde(flatten)] + base: BaseConfig, +} + +impl NymConfig for Config { + fn template() -> &'static str { + config_template() + } + + // TODO: merge base dir with `HostStore`. + fn default_root_directory() -> PathBuf { + dirs::home_dir() + .expect("Failed to evaluate $HOME value") + .join(".nym") + .join("service-providers") + .join("network-requester") + } + + fn try_default_root_directory() -> Option { + dirs::home_dir().map(|path| path.join(".nym").join("clients")) + } + + fn root_directory(&self) -> PathBuf { + self.base.get_nym_root_directory() + } + + fn config_directory(&self) -> PathBuf { + self.root_directory() + .join(self.base.get_id()) + .join("config") + } + + fn data_directory(&self) -> PathBuf { + self.root_directory().join(self.base.get_id()).join("data") + } +} + +impl ClientCoreConfigTrait for Config { + fn get_gateway_endpoint(&self) -> &client_core::config::GatewayEndpointConfig { + self.base.get_gateway_endpoint() + } +} + +impl Config { + pub fn new>(id: S) -> Self { + Config { + base: BaseConfig::new(id), + } + } + + // getters + pub fn get_config_file_save_location(&self) -> PathBuf { + self.config_directory().join(Self::config_file_name()) + } + + pub fn get_base(&self) -> &BaseConfig { + &self.base + } + + pub fn get_base_mut(&mut self) -> &mut BaseConfig { + &mut self.base + } + + // poor man's 'builder' method + pub fn with_base(mut self, f: F, val: T) -> Self + where + F: Fn(BaseConfig, T) -> BaseConfig, + { + self.base = f(self.base, val); + self + } + + // helper methods to use `OptionalSet` trait. Those are defined due to very... ehm. 'specific' structure of this config + // (plz, lets refactor it) + pub fn with_optional_ext(mut self, f: F, val: Option) -> Self + where + F: Fn(BaseConfig, T) -> BaseConfig, + { + self.base = self.base.with_optional(f, val); + self + } + + #[allow(dead_code)] + pub fn with_optional_env_ext(mut self, f: F, val: Option, env_var: &str) -> Self + where + F: Fn(BaseConfig, T) -> BaseConfig, + T: FromStr, + ::Err: Debug, + { + self.base = self.base.with_optional_env(f, val, env_var); + self + } + + pub fn with_optional_custom_env_ext( + mut self, + f: F, + val: Option, + env_var: &str, + parser: G, + ) -> Self + where + F: Fn(BaseConfig, T) -> BaseConfig, + G: Fn(&str) -> T, + { + self.base = self.base.with_optional_custom_env(f, val, env_var, parser); + self + } +} diff --git a/service-providers/network-requester/src/config/template.rs b/service-providers/network-requester/src/config/template.rs new file mode 100644 index 0000000000..395b8d7049 --- /dev/null +++ b/service-providers/network-requester/src/config/template.rs @@ -0,0 +1,99 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +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. + r#" +# This is a TOML config file. +# For more information, see https://github.com/toml-lang/toml + +##### main base client config options ##### + +[client] +# Version of the client for which this configuration was created. +version = '{{ client.version }}' + +# Human readable ID of this particular client. +id = '{{ client.id }}' + +# Indicates whether this client is running in a disabled credentials mode, thus attempting +# to claim bandwidth without presenting bandwidth credentials. +disabled_credentials_mode = {{ client.disabled_credentials_mode }} + +# Addresses to nyxd validators via which the client can communicate with the chain. +nyxd_urls = [{{#each client.nyxd_urls }} + '{{this}}', +{{/each}}] + +# Addresses to APIs running on validator from which the client gets the view of the network. +nym_api_urls = [{{#each client.nym_api_urls }} + '{{this}}', +{{/each}}] + +# 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 }}' + +# Path to the database containing bandwidth credentials +database_path = '{{ client.database_path }}' + +# Path to the persistent store for received reply surbs, unused encryption keys and used sender tags. +reply_surb_database_path = '{{ client.reply_surb_database_path }}' + +##### additional client config options ##### + +# 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 }}' + +[client.gateway_endpoint] +# ID of the gateway from which the client should be fetching messages. +gateway_id = '{{ client.gateway_endpoint.gateway_id }}' + +# Address of the gateway owner to which the client should send messages. +gateway_owner = '{{ client.gateway_endpoint.gateway_owner }}' + +# Address of the gateway listener to which all client requests should be sent. +gateway_listener = '{{ client.gateway_endpoint.gateway_listener }}' + + +##### 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/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index b7b16ec7e3..5c4b7ca040 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -3,27 +3,19 @@ use crate::allowed_hosts; use crate::allowed_hosts::OutboundRequestFilter; +use crate::config::Config; use crate::error::NetworkRequesterError; use crate::reply::MixnetMessage; use crate::statistics::ServiceStatisticsCollector; -use crate::websocket; -use crate::websocket::TSWebsocketStream; use crate::{reply, socks5}; use async_trait::async_trait; use build_information::BinaryBuildInformation; -use client_connections::{ - ConnectionCommand, ConnectionCommandReceiver, LaneQueueLengths, TransmissionLane, -}; +use client_connections::LaneQueueLengths; use futures::channel::mpsc; -use futures::stream::{SplitSink, SplitStream}; -use futures::{SinkExt, StreamExt}; use log::warn; use nymsphinx::addressing::clients::Recipient; use nymsphinx::anonymous_replies::requests::AnonymousSenderTag; -use nymsphinx::receiver::ReconstructedMessage; -use proxy_helpers::connection_controller::{ - BroadcastActiveConnections, Controller, ControllerCommand, ControllerSender, -}; +use proxy_helpers::connection_controller::{Controller, ControllerCommand, ControllerSender}; use proxy_helpers::proxy_runner::{MixProxyReader, MixProxySender}; use service_providers_common::interface::{ BinaryInformation, ProviderInterfaceVersion, Request, RequestVersion, @@ -37,8 +29,6 @@ use statistics_common::collector::StatisticsSender; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; use task::{TaskClient, TaskManager}; -use tokio_tungstenite::tungstenite::protocol::Message; -use websocket_requests::{requests::ClientRequest, responses::ServerResponse}; // Since it's an atomic, it's safe to be kept static and shared across threads static ACTIVE_PROXIES: AtomicUsize = AtomicUsize::new(0); @@ -53,7 +43,7 @@ pub(crate) fn new_legacy_request_version() -> RequestVersion { // TODO: 'Builder' is not the most appropriate name here, but I needed // ... something ... pub struct NRServiceProviderBuilder { - websocket_address: String, + config: Config, outbound_request_filter: OutboundRequestFilter, open_proxy: bool, enable_statistics: bool, @@ -63,11 +53,11 @@ pub struct NRServiceProviderBuilder { struct NRServiceProvider { outbound_request_filter: OutboundRequestFilter, open_proxy: bool, + mixnet_client: nym_sdk::mixnet::MixnetClient, - websocket_stream: SplitStream, controller_sender: ControllerSender, mix_input_sender: MixProxySender, - shared_lane_queue_lengths: LaneQueueLengths, + //shared_lane_queue_lengths: LaneQueueLengths, stats_collector: Option, shutdown: TaskManager, } @@ -157,7 +147,7 @@ impl ServiceProvider for NRServiceProvider { impl NRServiceProviderBuilder { pub async fn new( - websocket_address: String, + config: Config, open_proxy: bool, enable_statistics: bool, stats_provider_addr: Option, @@ -180,7 +170,7 @@ impl NRServiceProviderBuilder { let outbound_request_filter = OutboundRequestFilter::new(allowed_hosts, unknown_hosts); NRServiceProviderBuilder { - websocket_address, + config, outbound_request_filter, open_proxy, enable_statistics, @@ -190,10 +180,8 @@ impl NRServiceProviderBuilder { /// Start all subsystems pub async fn run_service_provider(self) -> Result<(), NetworkRequesterError> { - let websocket_stream = Self::connect_websocket(&self.websocket_address).await?; - - // split the websocket so that we could read and write from separate threads - let (websocket_writer, websocket_reader) = websocket_stream.split(); + // Connect to the mixnet + let mixnet_client = create_mixnet_client(self.config.get_base()).await?; // channels responsible for managing messages that are to be sent to the mix network. The receiver is // going to be used by `mixnet_response_listener` @@ -202,21 +190,9 @@ impl NRServiceProviderBuilder { // Used to notify tasks to shutdown. Not all tasks fully supports this (yet). let shutdown = task::TaskManager::default(); - // Channel for announcing client connection state by the controller. - // The `mixnet_response_listener` will use this to either report closed connection to the - // client or request lane queue lengths. - let (client_connection_tx, client_connection_rx) = mpsc::unbounded(); - - // Shared queue length data. Published by the `OutQueueController` in the client, and used - // primarily to throttle incoming connections - let shared_lane_queue_lengths = LaneQueueLengths::new(); - // Controller for managing all active connections. - // We provide it with a ShutdownListener since it requires it, even though for the network - // requester shutdown signalling is not yet fully implemented. let (mut active_connections_controller, controller_sender) = Controller::new( - client_connection_tx, - BroadcastActiveConnections::On, + mixnet_client.connection_command_sender(), shutdown.subscribe(), ); @@ -240,13 +216,15 @@ impl NRServiceProviderBuilder { }; let stats_collector_clone = stats_collector.clone(); + let mixnet_client_sender = mixnet_client.sender(); + let self_address = *mixnet_client.nym_address(); + // start the listener for mix messages tokio::spawn(async move { NRServiceProvider::mixnet_response_listener( - websocket_writer, + mixnet_client_sender, mix_input_receiver, stats_collector_clone, - client_connection_rx, ) .await; }); @@ -254,70 +232,54 @@ impl NRServiceProviderBuilder { let service_provider = NRServiceProvider { outbound_request_filter: self.outbound_request_filter, open_proxy: self.open_proxy, - websocket_stream: websocket_reader, + mixnet_client, controller_sender, mix_input_sender, - shared_lane_queue_lengths, + //shared_lane_queue_lengths: mixnet_client.shared_lane_queue_lengths(), stats_collector, shutdown, }; + log::info!("The address of this client is: {}", self_address); log::info!("All systems go. Press CTRL-C to stop the server."); service_provider.run().await } - - // Make the websocket connection so we can receive incoming Mixnet messages. - async fn connect_websocket(uri: &str) -> Result { - match websocket::Connection::new(uri).connect().await { - Ok(ws_stream) => { - log::info!("* connected to local websocket server at {}", uri); - Ok(ws_stream) - } - Err(err) => { - log::error!( - "Error: websocket connection attempt failed, is the Nym client running?" - ); - Err(err.into()) - } - } - } } impl NRServiceProvider { async fn run(mut self) -> Result<(), NetworkRequesterError> { // TODO: incorporate graceful shutdowns - loop { - let Some(reconstructed) = self.read_websocket_message().await else { - log::error!("The websocket stream has finished!"); - return Err(NetworkRequesterError::ConnectionClosed); - }; + while let Some(reconstructed_messages) = self.mixnet_client.wait_for_messages().await { + for reconstructed in reconstructed_messages { + let sender = reconstructed.sender_tag; + let request = match Socks5ProviderRequest::try_from_bytes(&reconstructed.message) { + Ok(req) => req, + Err(err) => { + // TODO: or should it even be further lowered to debug/trace? + log::warn!("Failed to deserialize received message: {err}"); + continue; + } + }; - let sender = reconstructed.sender_tag; - let request = match Socks5ProviderRequest::try_from_bytes(&reconstructed.message) { - Ok(req) => req, - Err(err) => { - // TODO: or should it even be further lowered to debug/trace? - log::warn!("Failed to deserialize received message: {err}"); - continue; + if let Err(err) = self.on_request(sender, request).await { + // TODO: again, should it be a warning? + // we should also probably log some information regarding the origin of the request + // so that it would be easier to debug it + log::warn!("failed to resolve the received request: {err}"); } - }; - - if let Err(err) = self.on_request(sender, request).await { - // TODO: again, should it be a warning? - // we should also probably log some information regarding the origin of the request - // so that it would be easier to debug it - log::warn!("failed to resolve the received request: {err}"); } } + + log::error!("Network requester exited unexpectedly"); + Ok(()) } /// Listens for any messages from `mix_reader` that should be written back to the mix network /// via the `websocket_writer`. async fn mixnet_response_listener( - mut websocket_writer: SplitSink, + mut mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender, mut mix_input_reader: MixProxyReader, stats_collector: Option, - mut client_connection_rx: ConnectionCommandReceiver, ) { loop { tokio::select! { @@ -338,97 +300,17 @@ impl NRServiceProvider { } } - // make 'request' to native-websocket client - let response_message = msg.into_client_request(); - let message = Message::Binary(response_message.serialize()); - websocket_writer.send(message).await.unwrap(); + let response_message = msg.into_input_message(); + mixnet_client_sender.send_input_message(response_message).await; } else { log::error!("Exiting: channel closed!"); break; } }, - Some(command) = client_connection_rx.next() => { - match command { - ConnectionCommand::Close(id) => { - let msg = ClientRequest::ClosedConnection(id); - let ws_msg = Message::Binary(msg.serialize()); - websocket_writer.send(ws_msg).await.unwrap(); - } - ConnectionCommand::ActiveConnections(ids) => { - // We can optimize this by sending a single request, but this is - // usually in the low single digits, max a few tens, so we leave that - // for a rainy day. - // Also that means fiddling with the currently manual - // serialize/deserialize we do with ClientRequests ... - for id in ids { - log::trace!("Requesting lane queue length for: {}", id); - let msg = ClientRequest::GetLaneQueueLength(id); - let ws_msg = Message::Binary(msg.serialize()); - websocket_writer.send(ws_msg).await.unwrap(); - } - } - } - }, } } } - fn handle_lane_queue_length_response( - lane_queue_lengths: &LaneQueueLengths, - lane: u64, - queue_length: usize, - ) { - log::trace!("Received LaneQueueLength lane: {lane}, queue_length: {queue_length}"); - if let Ok(mut lane_queue_lengths) = lane_queue_lengths.lock() { - let lane = TransmissionLane::ConnectionId(lane); - lane_queue_lengths.map.insert(lane, queue_length); - } else { - log::warn!("Unable to lock lane queue lengths, skipping updating received lane length") - } - } - - async fn read_websocket_message(&mut self) -> Option { - while let Some(msg) = self.websocket_stream.next().await { - let data = match msg { - Ok(msg) => msg.into_data(), - Err(err) => { - log::error!("Failed to read from the websocket: {err}"); - continue; - } - }; - - // try to recover the actual message from the mix network... - let deserialized_message = match ServerResponse::deserialize(&data) { - Ok(deserialized) => deserialized, - Err(err) => { - log::error!( - "Failed to deserialize received websocket message! - {}", - err - ); - continue; - } - }; - - let received = match deserialized_message { - ServerResponse::Received(received) => received, - ServerResponse::LaneQueueLength { lane, queue_length } => { - Self::handle_lane_queue_length_response( - &self.shared_lane_queue_lengths, - lane, - queue_length, - ); - continue; - } - ServerResponse::Error(err) => { - panic!("received error from native client! - {err}") - } - _ => unimplemented!("probably should never be reached?"), - }; - return Some(received); - } - None - } - #[allow(clippy::too_many_arguments)] async fn start_proxy( remote_version: RequestVersion, @@ -545,7 +427,7 @@ impl NRServiceProvider { let controller_sender_clone = self.controller_sender.clone(); let mix_input_sender_clone = self.mix_input_sender.clone(); - let lane_queue_lengths_clone = self.shared_lane_queue_lengths.clone(); + let lane_queue_lengths_clone = self.mixnet_client.shared_lane_queue_lengths(); let shutdown = self.shutdown.subscribe(); // and start the proxy for this connection @@ -568,3 +450,34 @@ impl NRServiceProvider { self.controller_sender.unbounded_send(req.into()).unwrap() } } + +// Helper function to create the mixnet client. +// This is NOT in the SDK since we don't want to expose any of the client-core config types. +// We could however consider moving it to a crate in common in the future. +async fn create_mixnet_client( + config: &client_core::config::Config, +) -> Result { + let nym_api_endpoints = config.get_nym_api_endpoints(); + let debug_config = config.get_debug_config().clone(); + + let mixnet_config = nym_sdk::mixnet::Config { + user_chosen_gateway: None, + nym_api_endpoints, + debug_config, + }; + + let storage_paths = nym_sdk::mixnet::StoragePaths::from(config); + + let mixnet_client = nym_sdk::mixnet::MixnetClientBuilder::new() + .config(mixnet_config) + .enable_storage(storage_paths) + .gateway_config(config.get_gateway_endpoint_config().clone()) + .build::() + .await + .map_err(|err| NetworkRequesterError::FailedToSetupMixnetClient { source: err })?; + + mixnet_client + .connect_to_mixnet() + .await + .map_err(|err| NetworkRequesterError::FailedToConnectToMixnet { source: err }) +} diff --git a/service-providers/network-requester/src/error.rs b/service-providers/network-requester/src/error.rs index a651d34659..251c19d4c9 100644 --- a/service-providers/network-requester/src/error.rs +++ b/service-providers/network-requester/src/error.rs @@ -1,4 +1,4 @@ -use crate::websocket::WebsocketConnectionError; +use client_core::error::ClientCoreError; use socks5_requests::Socks5RequestError; #[derive(thiserror::Error, Debug)] @@ -6,15 +6,27 @@ pub enum NetworkRequesterError { #[error("I/O error: {0}")] IoError(#[from] std::io::Error), - #[error("Websocket error")] - WebsocketConnectionError(#[from] WebsocketConnectionError), - - #[error("Websocket connection closed")] - ConnectionClosed, + #[error("client-core error: {0}")] + ClientCoreError(#[from] ClientCoreError), #[error("encountered an error while trying to handle a provider request: {source}")] ProviderRequestError { #[from] source: Socks5RequestError, }, + + #[error("failed to setup gateway: {source}")] + FailedToSetupGateway { source: ClientCoreError }, + + #[error("failed to load configuration file: {0}")] + FailedToLoadConfig(String), + + #[error("failed local version check, client and config mismatch")] + FailedLocalVersionCheck, + + #[error("failed to setup mixnet client: {source}")] + FailedToSetupMixnetClient { source: nym_sdk::Error }, + + #[error("failed to connect to mixnet: {source}")] + FailedToConnectToMixnet { source: nym_sdk::Error }, } diff --git a/service-providers/network-requester/src/main.rs b/service-providers/network-requester/src/main.rs index ad6f1fdc59..d1987048c1 100644 --- a/service-providers/network-requester/src/main.rs +++ b/service-providers/network-requester/src/main.rs @@ -1,114 +1,28 @@ // Copyright 2020 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use clap::CommandFactory; -use clap::Subcommand; -use clap::{Args, Parser}; -use completions::{fig_generate, ArgShell}; +use clap::{crate_name, crate_version, Parser}; use logging::setup_logging; +use network_defaults::setup_env; use error::NetworkRequesterError; -use nymsphinx::addressing::clients::Recipient; mod allowed_hosts; +mod cli; +mod config; mod core; mod error; mod reply; mod socks5; mod statistics; -mod websocket; - -const ENABLE_STATISTICS: &str = "enable-statistics"; - -#[derive(Args)] -struct Run { - /// Specifies whether this network requester should run in 'open-proxy' mode - #[clap(long)] - open_proxy: bool, - - /// Websocket port to bind to. - #[clap(long)] - websocket_port: Option, - - /// Enable service anonymized statistics that get sent to a statistics aggregator server - #[clap(long)] - enable_statistics: bool, - - /// Mixnet client address where a statistics aggregator is running. The default value is a Nym aggregator client - #[clap(long)] - statistics_recipient: Option, -} - -impl Run { - async fn execute(&self) -> Result<(), NetworkRequesterError> { - if self.open_proxy { - println!("\n\nYOU HAVE STARTED IN 'OPEN PROXY' MODE. ANYONE WITH YOUR CLIENT ADDRESS CAN MAKE REQUESTS FROM YOUR MACHINE. PLEASE QUIT IF YOU DON'T UNDERSTAND WHAT YOU'RE DOING.\n\n"); - } - - if self.enable_statistics { - println!("\n\nTHE NETWORK REQUESTER STATISTICS ARE ENABLED. IT WILL COLLECT AND SEND ANONYMIZED STATISTICS TO A CENTRAL SERVER. PLEASE QUIT IF YOU DON'T WANT THIS TO HAPPEN AND START WITHOUT THE {ENABLE_STATISTICS} FLAG .\n\n"); - } - - let stats_provider_addr = self - .statistics_recipient - .as_ref() - .map(Recipient::try_from_base58_string) - .transpose() - .unwrap_or(None); - - let websocket_address = format!( - "ws://localhost:{}", - self.websocket_port - .as_ref() - .unwrap_or(&network_defaults::DEFAULT_WEBSOCKET_LISTENING_PORT.to_string()) - ); - - log::info!("Starting socks5 service provider"); - let server = core::NRServiceProviderBuilder::new( - websocket_address, - self.open_proxy, - self.enable_statistics, - stats_provider_addr, - ) - .await; - server.run_service_provider().await - } -} - -#[derive(Subcommand)] -pub(crate) enum Commands { - /// Run network requester - Run(Run), - - /// Generate shell completions - Completions(ArgShell), - - /// Generate Fig specification - GenerateFigSpec, -} - -#[derive(Parser)] -#[clap(author = "Nymtech", version, about)] -struct Cli { - #[clap(subcommand)] - command: Commands, -} - -pub(crate) async fn execute(args: Cli) -> Result<(), NetworkRequesterError> { - let bin_name = "nym-network-requester"; - - match &args.command { - Commands::Run(r) => r.execute().await?, - Commands::Completions(s) => s.generate(&mut crate::Cli::command(), bin_name), - Commands::GenerateFigSpec => fig_generate(&mut crate::Cli::command(), bin_name), - } - Ok(()) -} #[tokio::main] async fn main() -> Result<(), NetworkRequesterError> { setup_logging(); - let args = Cli::parse(); + println!("{}", logging::banner(crate_name!(), crate_version!())); - execute(args).await + let args = cli::Cli::parse(); + setup_env(args.config_env_file.as_ref()); + + cli::execute(args).await } diff --git a/service-providers/network-requester/src/reply.rs b/service-providers/network-requester/src/reply.rs index 2616f36a7f..a135c9b2e8 100644 --- a/service-providers/network-requester/src/reply.rs +++ b/service-providers/network-requester/src/reply.rs @@ -1,6 +1,8 @@ // Copyright 2022-2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use client_connections::TransmissionLane; +use nym_sdk::mixnet::InputMessage; use nymsphinx::addressing::clients::Recipient; use nymsphinx::anonymous_replies::requests::AnonymousSenderTag; use service_providers_common::interface::{ @@ -11,7 +13,6 @@ use socks5_requests::{ Socks5RequestContent, Socks5Response, Socks5ResponseContent, }; use std::fmt::{Debug, Formatter}; -use websocket_requests::requests::ClientRequest; /// Generic data this service provider will send back to the mixnet via its connected native client. /// It includes serialized socks5 proxy responses to its connected clients @@ -146,7 +147,7 @@ impl MixnetMessage { self.data.len() } - pub(crate) fn into_client_request(self) -> ClientRequest { + pub(crate) fn into_input_message(self) -> InputMessage { self.address.send_back_to(self.data, self.connection_id) } } @@ -173,17 +174,17 @@ impl MixnetAddress { None } - pub(super) fn send_back_to(self, message: Vec, connection_id: u64) -> ClientRequest { + pub(super) fn send_back_to(self, message: Vec, connection_id: u64) -> InputMessage { match self { - MixnetAddress::Known(recipient) => ClientRequest::Send { + MixnetAddress::Known(recipient) => InputMessage::Regular { recipient: *recipient, - message, - connection_id: Some(connection_id), + data: message, + lane: TransmissionLane::ConnectionId(connection_id), }, - MixnetAddress::Anonymous(sender_tag) => ClientRequest::Reply { - message, - sender_tag, - connection_id: Some(connection_id), + MixnetAddress::Anonymous(sender_tag) => InputMessage::Reply { + recipient_tag: sender_tag, + data: message, + lane: TransmissionLane::ConnectionId(connection_id), }, } } diff --git a/service-providers/network-requester/src/websocket.rs b/service-providers/network-requester/src/websocket.rs deleted file mode 100644 index 421f6553a0..0000000000 --- a/service-providers/network-requester/src/websocket.rs +++ /dev/null @@ -1,34 +0,0 @@ -// Copyright 2020 - Nym Technologies SA -// SPDX-License-Identifier: Apache-2.0 - -use tokio::net::TcpStream; -use tokio_tungstenite::WebSocketStream; -use tokio_tungstenite::{connect_async, MaybeTlsStream}; - -#[allow(clippy::upper_case_acronyms)] -pub(crate) type TSWebsocketStream = WebSocketStream>; - -pub struct Connection { - uri: String, -} - -impl Connection { - pub fn new(uri: &str) -> Connection { - Connection { - uri: String::from(uri), - } - } - - pub async fn connect(&self) -> Result { - match connect_async(&self.uri).await { - Ok((ws_stream, _)) => Ok(ws_stream), - Err(e) => Err(WebsocketConnectionError::ConnectionNotEstablished(e)), - } - } -} - -#[derive(Debug, thiserror::Error)] -pub enum WebsocketConnectionError { - #[error("Connection not established")] - ConnectionNotEstablished(tokio_tungstenite::tungstenite::Error), -}