nym-nr-query CLI tool for query network-requester (#3538)

* Initial version

* Add open_proxy request

* Start adding cli commands

* use commands

* Common response type

* json output

* Tidy

* Remove the All case

* Remove empty file

* Remove todo

* Add ping command

* Ping 4 times by default

* Logic for controlling number of loops

* Some print tweaks

Cargo.lock

@@ -4150,6 +4150,22 @@ dependencies = [
  "wasm-timer",
 ]
 
+[[package]]
+name = "nym-nr-query"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "clap 4.2.7",
+ "log",
+ "nym-bin-common",
+ "nym-network-defaults",
+ "nym-sdk",
+ "nym-service-providers-common",
+ "nym-socks5-requests",
+ "serde",
+ "tokio",
+]
+
 [[package]]
 name = "nym-ordered-buffer"
 version = "0.1.0"

Cargo.toml

@@ -87,6 +87,7 @@ members = [
     "nym-api/nym-api-requests",
     "nym-outfox",
     "tools/nym-cli",
+    "tools/nym-nr-query",
     "tools/ts-rs-cli"
 ]
 

common/bin-common/src/build_information/mod.rs

@@ -101,7 +101,7 @@ impl BinaryBuildInformation {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct BinaryBuildInformationOwned {
     // VERGEN_BUILD_TIMESTAMP
     /// Provides the build timestamp, for example `2021-02-23T20:14:46.558472672+00:00`.

common/socks5/requests/src/response.rs

@@ -233,6 +233,13 @@ impl Socks5ResponseContent {
             }
         }
     }
+
+    pub fn as_query(&self) -> Option<&QueryResponse> {
+        match self {
+            Socks5ResponseContent::Query(query) => Some(query),
+            _ => None,
+        }
+    }
 }
 
 /// A remote network network data response retrieved by the Socks5 service provider. This

sdk/rust/nym-sdk/src/mixnet.rs

@@ -63,7 +63,10 @@ pub use nym_credential_storage::{
 pub use nym_network_defaults::NymNetworkDetails;
 pub use nym_socks5_client_core::config::Socks5;
 pub use nym_sphinx::{
-    addressing::clients::{ClientIdentity, Recipient},
+    addressing::{
+        clients::{ClientIdentity, Recipient},
+        nodes::NodeIdentity,
+    },
     receiver::ReconstructedMessage,
 };
 pub use nym_topology::{provider_trait::TopologyProvider, NymTopology};

service-providers/common/examples/control_requests.rs

@@ -48,7 +48,7 @@ async fn main() -> anyhow::Result<()> {
     let full_request_versions: Request =
         Request::new_control(ProviderInterfaceVersion::new_current(), request_versions);
 
-    // // TODO: currently we HAVE TO use surbs unfortunately
+    // TODO: currently we HAVE TO use surbs unfortunately
     println!("Sending 'Health' request...");
     client
         .send_bytes(
@@ -82,5 +82,6 @@ async fn main() -> anyhow::Result<()> {
     let response = wait_for_control_response(&mut client).await;
     println!("response to 'SupportedRequestVersions' request: {response:#?}");
 
+    client.disconnect().await;
     Ok(())
 }

service-providers/common/src/interface/control.rs

@@ -69,13 +69,13 @@ impl ControlRequest {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct BinaryInformation {
     pub binary_name: String,
     pub build_information: BinaryBuildInformationOwned,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct SupportedVersions {
     pub interface_version: String,
     pub provider_version: String,
@@ -86,7 +86,7 @@ pub struct ErrorResponse {
     message: String,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Serialize)]
 pub enum ControlResponse {
     Health,
     BinaryInfo(Box<BinaryInformation>),

service-providers/network-requester/Cargo.toml

@@ -35,22 +35,22 @@ url = { workspace = true }
 
 # internal
 async-file-watcher = { path = "../../common/async-file-watcher" }
+nym-bin-common = { path = "../../common/bin-common", features = ["output_format"] }
 nym-client-core = { path = "../../common/client-core" }
+nym-client-websocket-requests = { path = "../../clients/native/websocket-requests" }
 nym-config = { path = "../../common/config" }
 nym-credential-storage = { path = "../../common/credential-storage" }
 nym-crypto = { path = "../../common/crypto" }
-nym-bin-common = { path = "../../common/bin-common", features = ["output_format"] }
 nym-network-defaults = { path = "../../common/network-defaults" }
-nym-sdk = { path = "../../sdk/rust/nym-sdk" }
-nym-sphinx = { path = "../../common/nymsphinx" }
 nym-ordered-buffer = {path = "../../common/socks5/ordered-buffer"}
-nym-socks5-proxy-helpers = { path = "../../common/socks5/proxy-helpers" }
+nym-sdk = { path = "../../sdk/rust/nym-sdk" }
 nym-service-providers-common = { path = "../common" }
+nym-socks5-proxy-helpers = { path = "../../common/socks5/proxy-helpers" }
 nym-socks5-requests = { path = "../../common/socks5/requests" }
+nym-sphinx = { path = "../../common/nymsphinx" }
 nym-statistics-common = { path = "../../common/statistics" }
 nym-task = { path = "../../common/task" }
 nym-types = { path = "../../common/types" }
-nym-client-websocket-requests = { path = "../../clients/native/websocket-requests" }
 
 [dev-dependencies]
 tempfile = "3.5.0"

service-providers/network-requester/examples/query.rs

@@ -5,7 +5,7 @@ use nym_sdk::mixnet::{IncludedSurbs, MixnetClient, Recipient, ReconstructedMessa
 use nym_service_providers_common::interface::{
     ProviderInterfaceVersion, Request, Response, ResponseContent,
 };
-use nym_socks5_requests::{QueryRequest, Socks5Request, Socks5Response};
+use nym_socks5_requests::{QueryRequest, Socks5ProtocolVersion, Socks5Request, Socks5Response};
 
 fn parse_response(received: Vec<ReconstructedMessage>) -> Socks5Response {
     assert_eq!(received.len(), 1);
@@ -31,8 +31,6 @@ async fn main() -> anyhow::Result<()> {
     let mut client = MixnetClient::connect_new().await.unwrap();
     let provider: Recipient = "AN8eLxYWFitCkMn92zim3PrPszxJZDYyFFKP7qnnAAew.8UAxL3LwQBis6WpM3GGXaqKGaVdnLCpGJWumHT6KNdTH@77TSuVU8d1oXKbPzjec2xh4i3Wj5WwUyy9Lr36sm8gZm".parse().unwrap();
 
-    use nym_socks5_requests::Socks5ProtocolVersion;
-
     let open_proxy_request = Request::new_provider_data(
         ProviderInterfaceVersion::new_current(),
         Socks5Request::new_query(
@@ -52,7 +50,7 @@ async fn main() -> anyhow::Result<()> {
         .send_bytes(
             provider,
             open_proxy_request.into_bytes(),
-            IncludedSurbs::new(10), //crashes??
+            IncludedSurbs::new(10),
         )
         .await;
     let response = wait_for_response(&mut client).await;
@@ -69,5 +67,6 @@ async fn main() -> anyhow::Result<()> {
     let response = wait_for_response(&mut client).await;
     println!("response to 'Description' query: {response:#?}");
 
+    client.disconnect().await;
     Ok(())
 }

tools/nym-nr-query/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "nym-nr-query"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow = "1.0.68"
+clap = {version = "4.0", features = ["cargo", "derive"]}
+log = { workspace = true }
+nym-bin-common = { path = "../../common/bin-common", features = ["output_format"] }
+nym-network-defaults = { path = "../../common/network-defaults" }
+nym-sdk = { path = "../../sdk/rust/nym-sdk" }
+nym-service-providers-common = { path = "../../service-providers/common" }
+nym-socks5-requests = { path = "../../common/socks5/requests" }
+serde = { workspace = true }
+tokio = { workspace = true, features = [ "net", "rt-multi-thread", "macros" ] }

tools/nym-nr-query/src/main.rs

@@ -0,0 +1,331 @@
+use std::fmt;
+
+use clap::{Parser, ValueEnum};
+use nym_bin_common::output_format::OutputFormat;
+use nym_sdk::mixnet::{self, IncludedSurbs};
+use nym_service_providers_common::interface::{
+    ControlRequest, ControlResponse, ProviderInterfaceVersion, Request, Response, ResponseContent,
+};
+use nym_socks5_requests::{
+    QueryRequest, QueryResponse, Socks5ProtocolVersion, Socks5Request, Socks5Response,
+};
+use serde::Serialize;
+use tokio::time::{timeout, Duration};
+
+const RESPONSE_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[derive(Parser)]
+#[command(version, about, long_about = None)]
+struct Cli {
+    #[arg(short, long)]
+    config_env_file: Option<std::path::PathBuf>,
+
+    #[arg(short, long)]
+    debug: bool,
+
+    #[arg(short, long)]
+    provider: mixnet::Recipient,
+
+    #[arg(short, long)]
+    gateway: Option<mixnet::NodeIdentity>,
+
+    #[arg(short, long, default_value_t = OutputFormat::default())]
+    output: OutputFormat,
+
+    #[arg(value_enum, default_value_t = Commands::Ping)]
+    command: Commands,
+
+    /// By default, pinging is done continuously until manually stopped. This flag
+    /// specifies how many pings should be sent before stopping.
+    #[arg(short = 'n', long)]
+    ping_count: Option<usize>,
+}
+
+#[derive(Clone, ValueEnum, PartialEq, Eq)]
+enum Commands {
+    /// Binary information
+    BinaryInfo,
+
+    /// Supported request versions
+    SupportedRequestVersions,
+
+    /// Check if the network requester is acting a an open proxy
+    OpenProxy,
+
+    /// Ping the network requester
+    Ping,
+}
+
+fn parse_control_response(received: Vec<mixnet::ReconstructedMessage>) -> ControlResponse {
+    assert_eq!(received.len(), 1);
+    let response: Response = Response::try_from_bytes(&received[0].message).unwrap();
+    match response.content {
+        ResponseContent::Control(control) => control,
+        ResponseContent::ProviderData(_) => {
+            panic!("received provider data even though we sent control request!")
+        }
+    }
+}
+
+fn parse_socks5_response(received: Vec<mixnet::ReconstructedMessage>) -> Socks5Response {
+    assert_eq!(received.len(), 1);
+    let response: Response<Socks5Request> = Response::try_from_bytes(&received[0].message).unwrap();
+    match response.content {
+        ResponseContent::Control(control) => panic!("unexpected control response: {:?}", control),
+        ResponseContent::ProviderData(data) => data,
+    }
+}
+
+async fn wait_for_control_response(client: &mut mixnet::MixnetClient) -> ControlResponse {
+    loop {
+        let Ok(next) = timeout(RESPONSE_TIMEOUT, client.wait_for_messages()).await else {
+            eprintln!("Timeout waiting for response");
+            // Currently the control messages are not tagged with ID, so if we loose one we might
+            // en up out of sync. For now, just exit.
+            std::process::exit(1);
+        };
+        let next = next.unwrap();
+        if !next.is_empty() {
+            return parse_control_response(next);
+        }
+    }
+}
+
+async fn wait_for_socks5_response(client: &mut mixnet::MixnetClient) -> Socks5Response {
+    loop {
+        let Ok(next) = timeout(RESPONSE_TIMEOUT, client.wait_for_messages()).await else {
+            eprintln!("Timeout waiting for response");
+            std::process::exit(1);
+        };
+        let next = next.unwrap();
+        if !next.is_empty() {
+            return parse_socks5_response(next);
+        }
+    }
+}
+
+async fn connect_to_mixnet(gateway: Option<mixnet::NodeIdentity>) -> mixnet::MixnetClient {
+    match gateway {
+        Some(gateway) => mixnet::MixnetClientBuilder::new_ephemeral()
+            .request_gateway(gateway.to_base58_string())
+            .build()
+            .await
+            .expect("Failed to create mixnet client")
+            .connect_to_mixnet()
+            .await
+            .expect("Failed to connect to the mixnet"),
+        None => mixnet::MixnetClient::connect_new().await.unwrap(),
+    }
+}
+
+fn new_bin_info_request() -> Request {
+    let request_binary_info = ControlRequest::BinaryInfo;
+    Request::new_control(ProviderInterfaceVersion::new_current(), request_binary_info)
+}
+
+fn new_supported_request_versions_request() -> Request {
+    let request_versions = ControlRequest::SupportedRequestVersions;
+    Request::new_control(ProviderInterfaceVersion::new_current(), request_versions)
+}
+
+fn new_open_proxy_request() -> Request<Socks5Request> {
+    let request_open_proxy = Socks5Request::new_query(
+        Socks5ProtocolVersion::new_current(),
+        QueryRequest::OpenProxy,
+    );
+    Request::new_provider_data(ProviderInterfaceVersion::new_current(), request_open_proxy)
+}
+
+fn new_ping_request() -> Request {
+    let request_ping = ControlRequest::Health;
+    Request::new_control(ProviderInterfaceVersion::new_current(), request_ping)
+}
+
+struct QueryClient {
+    pub client: mixnet::MixnetClient,
+    pub provider: mixnet::Recipient,
+}
+
+impl QueryClient {
+    async fn new(provider: mixnet::Recipient, gateway: Option<mixnet::NodeIdentity>) -> Self {
+        let client = connect_to_mixnet(gateway).await;
+        Self { client, provider }
+    }
+
+    async fn query_bin_info(&mut self) -> ControlResponse {
+        self.client
+            .send_bytes(
+                self.provider,
+                new_bin_info_request().into_bytes(),
+                IncludedSurbs::new(10),
+            )
+            .await;
+        wait_for_control_response(&mut self.client).await
+    }
+
+    async fn query_supported_versions(&mut self) -> ControlResponse {
+        self.client
+            .send_bytes(
+                self.provider,
+                new_supported_request_versions_request().into_bytes(),
+                IncludedSurbs::new(10),
+            )
+            .await;
+        wait_for_control_response(&mut self.client).await
+    }
+
+    async fn query_open_proxy(&mut self) -> QueryResponse {
+        self.client
+            .send_bytes(
+                self.provider,
+                new_open_proxy_request().into_bytes(),
+                IncludedSurbs::new(10),
+            )
+            .await;
+        let response = wait_for_socks5_response(&mut self.client).await;
+        response
+            .content
+            .as_query()
+            .expect("Unexpected response type!")
+            .clone()
+    }
+
+    async fn ping(&mut self) -> PingResponse {
+        let now = std::time::Instant::now();
+        self.client
+            .send_bytes(
+                self.provider,
+                new_ping_request().into_bytes(),
+                IncludedSurbs::new(10),
+            )
+            .await;
+        let response = wait_for_control_response(&mut self.client).await;
+        assert!(matches!(response, ControlResponse::Health));
+        let elapsed = now.elapsed();
+        PingResponse {
+            provider: self.provider.to_string(),
+            ping_ms: elapsed.as_millis(),
+        }
+    }
+}
+
+#[derive(Debug, Serialize)]
+struct PingResponse {
+    provider: String,
+    ping_ms: u128,
+}
+
+impl fmt::Display for PingResponse {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}:  (surb) time={} ms", self.provider, self.ping_ms)
+    }
+}
+
+#[derive(Debug, Serialize)]
+enum ClientResponse {
+    Control(ControlResponse),
+    Query(QueryResponse),
+    Ping(PingResponse),
+}
+
+impl fmt::Display for ClientResponse {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ClientResponse::Control(control) => write!(f, "{:#?}", control),
+            ClientResponse::Query(query) => write!(f, "{:#?}", query),
+            ClientResponse::Ping(ping) => write!(f, "{}", ping),
+        }
+    }
+}
+
+impl From<ControlResponse> for ClientResponse {
+    fn from(response: ControlResponse) -> Self {
+        ClientResponse::Control(response)
+    }
+}
+
+impl From<QueryResponse> for ClientResponse {
+    fn from(response: QueryResponse) -> Self {
+        ClientResponse::Query(response)
+    }
+}
+
+impl From<PingResponse> for ClientResponse {
+    fn from(response: PingResponse) -> Self {
+        ClientResponse::Ping(response)
+    }
+}
+
+fn text_println(input: &str, output: &OutputFormat) {
+    if output.is_text() {
+        println!("{input}");
+    }
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let args = Cli::parse();
+
+    if args.debug {
+        nym_bin_common::logging::setup_logging();
+    }
+
+    nym_network_defaults::setup_env(args.config_env_file.as_ref());
+
+    text_println("Registering with gateway...", &args.output);
+    let mut client = QueryClient::new(args.provider, args.gateway).await;
+    let our_gateway = client.client.nym_address().gateway();
+    text_println(&format!("  gateway: {our_gateway}"), &args.output);
+
+    text_println("Sending request(s)...", &args.output);
+    if args.command == Commands::Ping {
+        let mut count = 0;
+        loop {
+            tokio::select! {
+                resp = client.ping() => {
+                    println!("{}", args.output.format(&resp));
+                }
+                _ = tokio::signal::ctrl_c() => {
+                    println!("Ctrl-C received, exiting...");
+                    break;
+                }
+            }
+            // Normally we loop until the use stops, but we can also specify a ping count.
+            count += 1;
+            if let Some(ping_count) = args.ping_count {
+                if count >= ping_count {
+                    break;
+                }
+            }
+            // If we specified json output, just run once, since this is likely to be called from a
+            // script.
+            if !args.output.is_text() {
+                break;
+            }
+        }
+    } else {
+        let resp: ClientResponse = match args.command {
+            Commands::BinaryInfo => client.query_bin_info().await.into(),
+            Commands::SupportedRequestVersions => client.query_supported_versions().await.into(),
+            Commands::OpenProxy => client.query_open_proxy().await.into(),
+            Commands::Ping => unreachable!(),
+        };
+        println!("{}", args.output.format(&resp));
+    }
+
+    text_println("Disconnecting...", &args.output);
+    client.client.disconnect().await;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use clap::CommandFactory;
+
+    #[test]
+    fn verify_cli() {
+        Cli::command().debug_assert();
+    }
+}