From 0fa1961a0430bb2b672dfbf2d522053926224008 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Mon, 12 Jun 2023 09:30:02 +0200 Subject: [PATCH] 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 | 16 + Cargo.toml | 1 + .../bin-common/src/build_information/mod.rs | 2 +- common/socks5/requests/src/response.rs | 7 + sdk/rust/nym-sdk/src/mixnet.rs | 5 +- .../common/examples/control_requests.rs | 3 +- .../common/src/interface/control.rs | 6 +- .../network-requester/Cargo.toml | 10 +- .../network-requester/examples/query.rs | 7 +- tools/nym-nr-query/Cargo.toml | 18 + tools/nym-nr-query/src/main.rs | 331 ++++++++++++++++++ 11 files changed, 391 insertions(+), 15 deletions(-) create mode 100644 tools/nym-nr-query/Cargo.toml create mode 100644 tools/nym-nr-query/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 29150b66be..ae39bd97a0 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index 15bdce9f4f..5480634392 100644 --- a/Cargo.toml +++ b/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" ] diff --git a/common/bin-common/src/build_information/mod.rs b/common/bin-common/src/build_information/mod.rs index 0e4efeff7f..d4fa7dd5c6 100644 --- a/common/bin-common/src/build_information/mod.rs +++ b/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`. diff --git a/common/socks5/requests/src/response.rs b/common/socks5/requests/src/response.rs index d0a34e074b..4a9c422d35 100644 --- a/common/socks5/requests/src/response.rs +++ b/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 diff --git a/sdk/rust/nym-sdk/src/mixnet.rs b/sdk/rust/nym-sdk/src/mixnet.rs index ecaf9523a3..c230094dbb 100644 --- a/sdk/rust/nym-sdk/src/mixnet.rs +++ b/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}; diff --git a/service-providers/common/examples/control_requests.rs b/service-providers/common/examples/control_requests.rs index eaff26b375..17af10448c 100644 --- a/service-providers/common/examples/control_requests.rs +++ b/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(()) } diff --git a/service-providers/common/src/interface/control.rs b/service-providers/common/src/interface/control.rs index 3c172fce80..8e808d6fdd 100644 --- a/service-providers/common/src/interface/control.rs +++ b/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), diff --git a/service-providers/network-requester/Cargo.toml b/service-providers/network-requester/Cargo.toml index 0333d90523..1210adb574 100644 --- a/service-providers/network-requester/Cargo.toml +++ b/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" diff --git a/service-providers/network-requester/examples/query.rs b/service-providers/network-requester/examples/query.rs index 42be5de55b..5b89e6924b 100644 --- a/service-providers/network-requester/examples/query.rs +++ b/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) -> 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(()) } diff --git a/tools/nym-nr-query/Cargo.toml b/tools/nym-nr-query/Cargo.toml new file mode 100644 index 0000000000..9e3dfbb727 --- /dev/null +++ b/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" ] } diff --git a/tools/nym-nr-query/src/main.rs b/tools/nym-nr-query/src/main.rs new file mode 100644 index 0000000000..1ca4697f09 --- /dev/null +++ b/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, + + #[arg(short, long)] + debug: bool, + + #[arg(short, long)] + provider: mixnet::Recipient, + + #[arg(short, long)] + gateway: Option, + + #[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, +} + +#[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) -> 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) -> Socks5Response { + assert_eq!(received.len(), 1); + let response: Response = 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::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 { + 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) -> 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 for ClientResponse { + fn from(response: ControlResponse) -> Self { + ClientResponse::Control(response) + } +} + +impl From for ClientResponse { + fn from(response: QueryResponse) -> Self { + ClientResponse::Query(response) + } +} + +impl From 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(); + } +}