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

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

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!(),
             }
         }
 

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 {

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<ConnectionId>),
 }
 
 // The `OutQueueControl` publishes the backlog per lane, primarily so that upstream can slow down

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<u8>, 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;

service-providers/network-requester/Cargo.toml

@@ -4,18 +4,19 @@
 [package]
 name = "nym-network-requester"
 version = "1.1.9"
-authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
-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" }

service-providers/network-requester/src/cli/init.rs

@@ -0,0 +1,157 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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<identity::PublicKey>,
+
+    /// 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<Vec<url::Url>>,
+
+    /// 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<Vec<url::Url>>,
+
+    /// 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<bool>,
+
+    /// Save a summary of the initialization to a json file
+    #[clap(long)]
+    output_json: bool,
+}
+
+impl From<Init> 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::<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");
+}

service-providers/network-requester/src/cli/mod.rs

@@ -0,0 +1,106 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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<std::path::PathBuf>,
+
+    #[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<Vec<url::Url>>,
+    fastmode: bool,
+    no_cover: bool,
+    nyxd_urls: Option<Vec<url::Url>>,
+    enabled_credentials_mode: Option<bool>,
+}
+
+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();
+    }
+}

service-providers/network-requester/src/cli/run.rs

@@ -0,0 +1,157 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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<String>,
+
+    /// 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<bool>,
+
+    /// 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<Run> 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
+}

service-providers/network-requester/src/config/mod.rs

@@ -0,0 +1,128 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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<Config>,
+}
+
+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<PathBuf> {
+        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<S: Into<String>>(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> {
+        &self.base
+    }
+
+    pub fn get_base_mut(&mut self) -> &mut BaseConfig<Self> {
+        &mut self.base
+    }
+
+    // poor man's 'builder' method
+    pub fn with_base<F, T>(mut self, f: F, val: T) -> Self
+    where
+        F: Fn(BaseConfig<Self>, T) -> BaseConfig<Self>,
+    {
+        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<F, T>(mut self, f: F, val: Option<T>) -> Self
+    where
+        F: Fn(BaseConfig<Self>, T) -> BaseConfig<Self>,
+    {
+        self.base = self.base.with_optional(f, val);
+        self
+    }
+
+    #[allow(dead_code)]
+    pub fn with_optional_env_ext<F, T>(mut self, f: F, val: Option<T>, env_var: &str) -> Self
+    where
+        F: Fn(BaseConfig<Self>, T) -> BaseConfig<Self>,
+        T: FromStr,
+        <T as FromStr>::Err: Debug,
+    {
+        self.base = self.base.with_optional_env(f, val, env_var);
+        self
+    }
+
+    pub fn with_optional_custom_env_ext<F, T, G>(
+        mut self,
+        f: F,
+        val: Option<T>,
+        env_var: &str,
+        parser: G,
+    ) -> Self
+    where
+        F: Fn(BaseConfig<Self>, T) -> BaseConfig<Self>,
+        G: Fn(&str) -> T,
+    {
+        self.base = self.base.with_optional_custom_env(f, val, env_var, parser);
+        self
+    }
+}

service-providers/network-requester/src/config/template.rs

@@ -0,0 +1,99 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// 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 }}'
+
+"#
+}

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<Socks5Request> {
 // 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<TSWebsocketStream>,
     controller_sender: ControllerSender,
     mix_input_sender: MixProxySender<MixnetMessage>,
-    shared_lane_queue_lengths: LaneQueueLengths,
+    //shared_lane_queue_lengths: LaneQueueLengths,
     stats_collector: Option<ServiceStatisticsCollector>,
     shutdown: TaskManager,
 }
@@ -157,7 +147,7 @@ impl ServiceProvider<Socks5Request> for NRServiceProvider {
 
 impl NRServiceProviderBuilder {
     pub async fn new(
-        websocket_address: String,
+        config: Config,
         open_proxy: bool,
         enable_statistics: bool,
         stats_provider_addr: Option<Recipient>,
@@ -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<TSWebsocketStream, NetworkRequesterError> {
-        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<TSWebsocketStream, Message>,
+        mut mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
         mut mix_input_reader: MixProxyReader<MixnetMessage>,
         stats_collector: Option<ServiceStatisticsCollector>,
-        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<ReconstructedMessage> {
-        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<Socks5Request>,
@@ -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<T>(
+    config: &client_core::config::Config<T>,
+) -> Result<nym_sdk::mixnet::MixnetClient, NetworkRequesterError> {
+    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::<nym_sdk::mixnet::ReplyStorage>()
+        .await
+        .map_err(|err| NetworkRequesterError::FailedToSetupMixnetClient { source: err })?;
+
+    mixnet_client
+        .connect_to_mixnet()
+        .await
+        .map_err(|err| NetworkRequesterError::FailedToConnectToMixnet { source: err })
+}

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

service-providers/network-requester/src/main.rs

@@ -1,114 +1,28 @@
 // Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
 // 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<String>,
-
-    /// 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<String>,
-}
-
-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
 }

service-providers/network-requester/src/reply.rs

@@ -1,6 +1,8 @@
 // Copyright 2022-2023 - Nym Technologies SA <contact@nymtech.net>
 // 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<u8>, connection_id: u64) -> ClientRequest {
+    pub(super) fn send_back_to(self, message: Vec<u8>, 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),
             },
         }
     }

service-providers/network-requester/src/websocket.rs

@@ -1,34 +0,0 @@
-// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
-// 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<MaybeTlsStream<TcpStream>>;
-
-pub struct Connection {
-    uri: String,
-}
-
-impl Connection {
-    pub fn new(uri: &str) -> Connection {
-        Connection {
-            uri: String::from(uri),
-        }
-    }
-
-    pub async fn connect(&self) -> Result<TSWebsocketStream, WebsocketConnectionError> {
-        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),
-}