GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

PR feedback

Browse Source
This commit is contained in:
dynco-nym
2026-01-19 14:05:26 +01:00
parent 52a1a9044b
commit e1eee567cf
7 changed files with 445 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-gateway-probe/src/lib.rs
+83 -61
View File
@@ -7,8 +7,8 @@ use std::{
time::Duration,
};
use crate::socks5_test::{HttpsConnectivityResult, HttpsConnectivityTest};
use crate::types::Entry;
use crate::types::HttpsConnectivityTest;
use anyhow::bail;
use base64::{Engine as _, engine::general_purpose};
use bytes::BytesMut;
@@ -50,7 +50,7 @@ use url::Url;
use crate::{
icmp::{check_for_icmp_beacon_reply, icmp_identifier, send_ping_v4, send_ping_v6},
types::{Exit, HttpsConnectivityResult, Socks5ProbeResults},
types::{Exit, Socks5ProbeResults},
};
mod bandwidth_helpers;
@@ -59,6 +59,7 @@ mod icmp;
pub mod mode;
mod netstack;
pub mod nodes;
mod socks5_test;
mod types;
use crate::bandwidth_helpers::{acquire_bandwidth, import_bandwidth};
@@ -129,7 +130,10 @@ impl CredentialArgs {
#[derive(Args)]
pub struct Socks5Args {
#[arg(long, default_value_t = 45)]
#[arg(long, value_delimiter = ',')]
socks5_json_rpc_url_list: Vec<String>,
#[arg(long, default_value_t = 30)]
mixnet_client_timeout_sec: u64,
#[arg(long, default_value_t = 10)]
@@ -327,28 +331,13 @@ impl Probe {
.exit_gateway_nr(&exit_gateway)?
.to_testable_node()?;
let socks5_outcome = {
if let Some(ref nr_details) = node_info.network_requester_details {
match do_socks5_connectivity_test(
&nr_details.address,
network_details,
Some(&directory),
self.socks5_args.mixnet_client_timeout_sec,
self.socks5_args.test_count,
)
.await
{
Ok(results) => Some(results),
Err(e) => {
error!("SOCKS5 test failed: {}", e);
None
}
}
} else {
info!("No NR available, skipping SOCKS5 tests");
None
}
};
let socks5_outcome = self
.test_socks5_if_possible(
network_details,
&node_info.network_requester_details,
&directory,
)
.await;
let probe_result = ProbeResult {
node: exit_gateway.to_base58_string(),
@@ -375,6 +364,7 @@ impl Probe {
only_lp_registration: bool,
test_lp_wg: bool,
min_mixnet_performance: Option<u8>,
network_details: NymNetworkDetails,
) -> anyhow::Result<ProbeResult> {
let tickets_materials = self.credentials_args.decode_attached_ticket_materials()?;
@@ -386,7 +376,7 @@ impl Probe {
// Connect to the mixnet via the entry gateway
let disconnected_mixnet_client = MixnetClientBuilder::new_with_storage(storage.clone())
.request_gateway(mixnet_entry_gateway_id.to_string())
.network_details(NymNetworkDetails::new_from_env())
.network_details(network_details.clone())
.debug_config(mixnet_debug_config(
min_mixnet_performance,
ignore_egress_epoch_role,
@@ -417,6 +407,7 @@ impl Probe {
test_mode,
only_wireguard,
false, // Not using mock ecash in regular probe mode
network_details,
)
.await
}
@@ -434,6 +425,7 @@ impl Probe {
test_lp_wg: bool,
min_mixnet_performance: Option<u8>,
use_mock_ecash: bool,
network_details: NymNetworkDetails,
) -> anyhow::Result<ProbeResult> {
// Localnet mode - identity + LP address from CLI, no HTTP query
// This path is used when --entry-gateway-identity is specified
@@ -474,6 +466,7 @@ impl Probe {
test_mode,
only_wireguard,
use_mock_ecash,
network_details,
)
.await;
}
@@ -521,6 +514,7 @@ impl Probe {
test_mode,
only_wireguard,
use_mock_ecash,
network_details,
)
.await;
}
@@ -553,7 +547,7 @@ impl Probe {
// and keeps its bandwidth between probe runs
let disconnected_mixnet_client = MixnetClientBuilder::new_with_storage(storage.clone())
.request_gateway(mixnet_entry_gateway_id.to_string())
.network_details(NymNetworkDetails::new_from_env())
.network_details(network_details.clone())
.debug_config(mixnet_debug_config(
min_mixnet_performance,
ignore_egress_epoch_role,
@@ -612,6 +606,7 @@ impl Probe {
test_mode,
only_wireguard,
use_mock_ecash,
network_details,
)
.await
}
@@ -699,6 +694,35 @@ impl Probe {
})
}
async fn test_socks5_if_possible(
&self,
network_details: NymNetworkDetails,
network_requester_details: &Option<NetworkRequesterDetails>,
directory: &NymApiDirectory,
) -> Option<Socks5ProbeResults> {
if let Some(nr_details) = network_requester_details {
match do_socks5_connectivity_test(
&nr_details.address,
network_details,
directory,
self.socks5_args.socks5_json_rpc_url_list.clone(),
self.socks5_args.mixnet_client_timeout_sec,
self.socks5_args.test_count,
)
.await
{
Ok(results) => Some(results),
Err(e) => {
error!("SOCKS5 test failed: {}", e);
None
}
}
} else {
info!("No NR available, skipping SOCKS5 tests");
None
}
}
pub async fn lookup_gateway(
&self,
directory: &Option<NymApiDirectory>,
@@ -771,11 +795,15 @@ impl Probe {
test_mode: TestMode,
only_wireguard: bool,
use_mock_ecash: bool,
network_details: NymNetworkDetails,
) -> anyhow::Result<ProbeResult>
where
T: MixnetClientStorage + Clone + 'static,
<T::CredentialStore as CredentialStorage>::StorageError: Send + Sync,
{
let Some(directory) = directory else {
bail!("You need to provide NYM API through environment")
};
// test_mode replaces the old only_lp_registration and test_lp_wg flags.
// only_wireguard is kept separate as it controls ping behavior within Mixnet mode.
let mut rng = rand::thread_rng();
@@ -919,8 +947,6 @@ impl Probe {
// The tested node is the exit
let exit_gateway = node_info.clone();
let directory = directory
.ok_or_else(|| anyhow::anyhow!("Directory is required for LP-WG test mode"))?;
let entry_gateway_node = directory.entry_gateway(&mixnet_entry_gateway_id)?;
let entry_gateway = entry_gateway_node.to_testable_node()?;
@@ -962,9 +988,8 @@ impl Probe {
Arc::new(KeyPair::new(&mut rng)),
ip_address,
);
let config = nym_validator_client::nyxd::Config::try_from_nym_network_details(
&NymNetworkDetails::new_from_env(),
)?;
let config =
nym_validator_client::nyxd::Config::try_from_nym_network_details(&network_details)?;
let client =
nym_validator_client::nyxd::NyxdClient::connect(config, nyxd_url.as_str())?;
let bw_controller = nym_bandwidth_controller::BandwidthController::new(
@@ -1037,28 +1062,13 @@ impl Probe {
};
// test failure doesn't stop further tests
let socks5_outcome = {
if let Some(ref nr_details) = node_info.network_requester_details {
match do_socks5_connectivity_test(
&nr_details.address,
NymNetworkDetails::new_from_env(),
directory.as_deref(),
self.socks5_args.mixnet_client_timeout_sec,
self.socks5_args.test_count,
)
.await
{
Ok(results) => Some(results),
Err(e) => {
error!("SOCKS5 test failed: {}", e);
None
}
}
} else {
info!("No NR available, skipping SOCKS5 tests");
None
}
};
let socks5_outcome = self
.test_socks5_if_possible(
network_details,
&node_info.network_requester_details,
directory,
)
.await;
// Disconnect the mixnet client gracefully
outcome.map(|mut outcome| {
@@ -1657,7 +1667,8 @@ async fn do_ping_exit(
async fn do_socks5_connectivity_test(
network_requester_address: &str,
network_details: NymNetworkDetails,
directory: Option<&NymApiDirectory>,
directory: &NymApiDirectory,
json_rpc_endpoints: Vec<String>,
mixnet_client_timeout: u64,
test_run_count: u64,
) -> anyhow::Result<Socks5ProbeResults> {
@@ -1703,9 +1714,6 @@ async fn do_socks5_connectivity_test(
// construction in case GW would get filtered out on topology refresh
debug_config.topology.minimum_gateway_performance = 0;
let Some(directory) = directory else {
bail!("You need to provide Nym API directory through environment")
};
// Verify the NR gateway exists in the directory with exit_nr role
let nr_gateway_id = nr_recipient.gateway();
if let Err(e) = directory.exit_gateway_nr(&nr_gateway_id) {
@@ -1729,7 +1737,7 @@ async fn do_socks5_connectivity_test(
// connect to mixnet via SOCKS5
let socks5_client = match socks5_client_builder.connect_to_mixnet_via_socks5().await {
Ok(client) => {
info!("Successfully established SOCKS5 proxy connection");
info!("🌐 Successfully connected to mixnet via SOCKS5 proxy");
info!(
"Connected via entry gateway: {}",
client.nym_address().gateway().to_base58_string()
@@ -1746,9 +1754,23 @@ async fn do_socks5_connectivity_test(
};
info!("Waiting for network topology to be ready...");
tokio::time::sleep(Duration::from_secs(10)).await;
let topology_timeout = Duration::from_secs(60);
if let Err(e) = socks5_client.wait_for_topology(topology_timeout).await {
error!(
"Topology not available after {}s: {}",
topology_timeout.as_secs(),
e
);
results.https_connectivity =
HttpsConnectivityResult::with_error(format!("Topology timeout: {}", e));
socks5_client.disconnect().await;
return Ok(results);
} else {
info!("Network topology is ready")
}
let test = HttpsConnectivityTest::new(test_run_count, mixnet_client_timeout);
let test =
HttpsConnectivityTest::new(test_run_count, mixnet_client_timeout, json_rpc_endpoints);
results.https_connectivity = test.run_tests(socks5_client.socks5_url()).await;
// cleanup
nym-gateway-probe/src/nodes.rs
+3 -3
View File
@@ -385,7 +385,7 @@ impl NymApiDirectory {
.iter()
.filter(|(_, n)| n.described.description.ip_packet_router.is_some())
.choose(&mut rand::thread_rng())
.ok_or(anyhow!("no gateways running IPR available"))
.context("no gateways running IPR available")
.map(|(id, _)| *id)
}
@@ -395,7 +395,7 @@ impl NymApiDirectory {
.iter()
.filter(|(_, n)| n.described.description.ip_packet_router.is_some())
.choose(&mut rand::thread_rng())
.ok_or(anyhow!("no gateways running NR available"))
.context("no gateways running NR available")
.map(|(id, _)| *id)
}
@@ -405,7 +405,7 @@ impl NymApiDirectory {
.iter()
.filter(|(_, n)| n.described.description.declared_role.entry)
.choose(&mut rand::thread_rng())
.ok_or(anyhow!("no entry gateways available"))
.context("no entry gateways available")
.map(|(id, _)| *id)
}
nym-gateway-probe/src/run.rs
+4
View File
@@ -333,6 +333,7 @@ pub(crate) async fn run() -> anyhow::Result<ProbeResult> {
test_lp_wg,
args.min_gateway_mixnet_performance,
*use_mock_ecash,
network,
))
.await
}
@@ -345,6 +346,7 @@ pub(crate) async fn run() -> anyhow::Result<ProbeResult> {
only_lp_registration,
test_lp_wg,
args.min_gateway_mixnet_performance,
network,
))
.await
}
@@ -487,6 +489,7 @@ pub(crate) async fn run() -> anyhow::Result<ProbeResult> {
test_lp_wg,
args.min_gateway_mixnet_performance,
use_mock_ecash,
network,
))
.await
}
@@ -499,6 +502,7 @@ pub(crate) async fn run() -> anyhow::Result<ProbeResult> {
only_lp_registration,
test_lp_wg,
args.min_gateway_mixnet_performance,
network,
))
.await
}
nym-gateway-probe/src/socks5_test.rs
+318
View File
@@ -0,0 +1,318 @@
use rand::Rng;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use tracing::{debug, error, info, warn};
pub struct HttpsConnectivityTest {
test_count: u64,
mixnet_client_timeout: Duration,
json_rpc_test_endpoints: Vec<String>,
}
impl HttpsConnectivityTest {
pub fn new(
test_count: u64,
mixnet_client_timeout: u64,
json_rpc_test_endpoints: Vec<String>,
) -> Self {
Self {
test_count: std::cmp::max(test_count, 1),
mixnet_client_timeout: Duration::from_secs(mixnet_client_timeout),
json_rpc_test_endpoints,
}
}
pub async fn run_tests(self, socks5_url: String) -> HttpsConnectivityResult {
let mut result = HttpsConnectivityResult::default();
let proxy = match reqwest::Proxy::all(socks5_url) {
Ok(p) => p,
Err(e) => {
return HttpsConnectivityResult::with_error(
format!("Failed to create proxy: {e}",),
);
}
};
let client = match reqwest::Client::builder()
.proxy(proxy)
.timeout(self.mixnet_client_timeout)
.build()
{
Ok(c) => c,
Err(e) => {
return HttpsConnectivityResult::with_error(format!(
"Failed to build HTTP client: {e}",
));
}
};
let mut successful_runs = 0u64;
for i in 1..self.test_count + 1 {
info!("Running test {}/{}", i, self.test_count);
let interim_res = self.perform_https_request(&client).await;
if interim_res.https_success
&& let Some(latency_ms) = interim_res.https_latency_ms
{
successful_runs += 1;
result.https_latency_ms = Some(
result
.https_latency_ms
.map_or(latency_ms, |existing| existing + latency_ms),
);
result.https_success = true;
result.https_status_code = interim_res.https_status_code;
info!("{}/{} latency: {}ms", i, self.test_count, latency_ms);
} else if let Some(new_error) = interim_res.error {
result.error = Some(result.error.map_or(new_error.clone(), |existing| {
format!("{},{}", existing, new_error)
}))
}
// too many failed runs: return early
let unsuccessful_runs = i - successful_runs;
if successful_runs < 2 && unsuccessful_runs > 2 {
// if < 2 runs, we don't have to calculate average before returning
warn!("Too many failed runs: returning early...");
return result;
}
}
result.https_latency_ms = result
.https_latency_ms
.map(|latency| latency / successful_runs);
info!(
"AVG latency over {} runs (in ms): {:?}",
successful_runs, result.https_latency_ms
);
result
}
async fn perform_https_request(&self, client: &reqwest::Client) -> HttpsConnectivityResult {
use tokio::time::Instant;
// TODO dz instead of initializing a mutable default, then mutating fields, use constructors for outcome
let start = Instant::now();
let mut error_msg = String::new();
// TODO dz utilize others as fallback
// let endpoint = self.json_rpc_test_endpoints.first().unwrap();
for endpoint in self.json_rpc_test_endpoints.iter() {
info!(
"Testing against {} with timeout {}s",
endpoint,
self.mixnet_client_timeout.as_secs()
);
match client
.post(endpoint)
.timeout(self.mixnet_client_timeout)
.header(reqwest::header::CONTENT_TYPE, "application/json")
.json(&JsonRpcRequestBody::eth_chainid())
.send()
.await
.and_then(reqwest::Response::error_for_status)
{
Ok(response) => {
let elapsed = start.elapsed();
let status = response.status();
if status.is_success() {
// Deserialize body into JsonRpcResponse
match response.json::<JsonRpcResponse>().await {
Ok(JsonRpcResponse::Ok { .. }) => {
let res = HttpsConnectivityResult::success(
status.as_u16(),
elapsed.as_millis() as u64,
endpoint.to_string(),
);
debug!(
"HTTPS test completed: status={}, latency={}ms",
status.as_u16(),
elapsed.as_millis()
);
return res;
}
Ok(JsonRpcResponse::Err { error, .. }) => {
warn!("JSON-RPC error: {} (code: {})", error.message, error.code);
error_msg = format!("JSON-RPC error: {}", error.message);
}
Err(e) => {
error!("Failed to parse JSON-RPC response: {}", e);
error_msg = format!("Failed to parse JSON-RPC response: {e}");
}
}
} else {
error_msg = format!("HTTP error status: {}", status.as_u16());
}
}
Err(e) => {
error!("HTTPS request failed: {}", e);
error_msg = format!("HTTPS request failed: {}", e);
}
}
}
HttpsConnectivityResult::with_error(error_msg)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct HttpsConnectivityResult {
/// successfully completed HTTPS request
https_success: bool,
/// HTTPS status code received
https_status_code: Option<u16>,
/// average HTTPS request latency in milliseconds
https_latency_ms: Option<u64>,
/// among multiple endpoints available, list the one actually used
endpoint_used: Option<String>,
/// error message(s) (if any)
error: Option<String>,
}
impl HttpsConnectivityResult {
pub fn with_error(error: impl Into<String>) -> Self {
Self {
https_success: false,
https_status_code: None,
https_latency_ms: None,
endpoint_used: None,
error: Some(error.into()),
}
}
pub fn success(status_code: u16, latency: u64, endpoint_used: String) -> Self {
Self {
https_success: true,
https_status_code: Some(status_code),
https_latency_ms: Some(latency),
endpoint_used: Some(endpoint_used),
error: None,
}
}
}
/// https://www.jsonrpc.org/specification
#[derive(Serialize)]
struct JsonRpcRequestBody {
// A String specifying the version of the JSON-RPC protocol. MUST be exactly "2.0".
jsonrpc: String,
method: String,
// A Structured value that holds the parameter values to be used during the invocation of the method. This member MAY be omitted.
params: serde_json::Value,
// The Server MUST reply with the same value in the Response object if included.
// This member is used to correlate the context between the two objects.
id: i64,
}
impl JsonRpcRequestBody {
/// Very simple endpoint that requires no dynamic input
///
/// https://ethereum.org/developers/docs/apis/json-rpc/#eth_chainId
pub fn eth_chainid() -> Self {
Self {
jsonrpc: String::from("2.0"),
method: String::from("eth_chainId"),
params: serde_json::json!([]),
id: rand::thread_rng().r#gen(),
}
}
/// Create an eth_getBlockByNumber request with invalid params for testing error responses
#[cfg(test)]
pub fn eth_get_block_by_number_invalid() -> Self {
Self {
jsonrpc: String::from("2.0"),
method: String::from("eth_getBlockByNumber"),
// Invalid params: should be [blockNumber, boolean] but we pass garbage
params: serde_json::json!(["invalid_block_number"]),
id: rand::thread_rng().r#gen(),
}
}
}
// dead code: we need these fields for deserialization, even if we don't read them explicitly
#[allow(dead_code)]
#[derive(Deserialize)]
#[serde(untagged)]
enum JsonRpcResponse {
Ok {
jsonrpc: String,
// have to use opaque Value because spec say this might be string, number or null (we don't care either way)
id: serde_json::Value,
// we don't really care for the exact result, just whether the response is OK or error
result: serde_json::Value,
},
Err {
jsonrpc: String,
// have to use opaque Value because spec say this might be string, number or null (we don't care either way)
id: serde_json::Value,
error: JsonRpcError,
},
}
// dead code: we need these fields for deserialization, even if we don't read them explicitly
#[allow(dead_code)]
#[derive(Debug, Deserialize)]
struct JsonRpcError {
pub code: i64,
pub message: String,
#[serde(default)]
pub data: Option<serde_json::Value>,
}
#[cfg(test)]
mod test {
use super::*;
const JSON_RPC_ENDPOINT: &str = "https://cloudflare-eth.com";
#[tokio::test]
async fn test_eth_chainid_returns_ok_response() {
let client = reqwest::Client::new();
let response = client
.post(JSON_RPC_ENDPOINT)
.header(reqwest::header::CONTENT_TYPE, "application/json")
.json(&JsonRpcRequestBody::eth_chainid())
.send()
.await
.expect("Failed to send request");
assert!(response.status().is_success());
let json_response: JsonRpcResponse =
response.json().await.expect("Failed to parse response");
assert!(
matches!(json_response, JsonRpcResponse::Ok { .. }),
"Expected Ok variant for eth_chainId"
);
}
#[tokio::test]
async fn test_eth_get_block_by_number_invalid_returns_error_response() {
let client = reqwest::Client::new();
let response = client
.post(JSON_RPC_ENDPOINT)
.header(reqwest::header::CONTENT_TYPE, "application/json")
.json(&JsonRpcRequestBody::eth_get_block_by_number_invalid())
.send()
.await
.expect("Failed to send request");
assert!(response.status().is_success()); // HTTP 200 but JSON-RPC error
let json_response: JsonRpcResponse =
response.json().await.expect("Failed to parse response");
assert!(
matches!(json_response, JsonRpcResponse::Err { .. }),
"Expected Err variant for invalid params"
);
}
}
nym-gateway-probe/src/types.rs
+2 -157
View File
@@ -1,8 +1,7 @@
use std::time::Duration;
use nym_connection_monitor::ConnectionStatusEvent;
use serde::{Deserialize, Serialize};
use tracing::{debug, info, warn};
use crate::socks5_test::HttpsConnectivityResult;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProbeResult {
@@ -145,160 +144,6 @@ pub struct Socks5ProbeResults {
pub https_connectivity: HttpsConnectivityResult,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct HttpsConnectivityResult {
/// successfully completed HTTPS request
https_success: bool,
/// HTTPS status code received
https_status_code: Option<u16>,
/// average HTTPS request latency in milliseconds
https_latency_ms: Option<u64>,
/// error message(s) (if any)
error: Option<String>,
}
impl HttpsConnectivityResult {
pub fn with_error(error: impl Into<String>) -> Self {
Self {
https_success: false,
https_status_code: None,
https_latency_ms: None,
error: Some(error.into()),
}
}
}
pub struct HttpsConnectivityTest {
test_count: u64,
mixnet_client_timeout: Duration,
}
/// endpoint to test against
/// https://www.quicknode.com/docs/ethereum/web3_clientVersion
const TARGET_URL: &str = "https://docs-demo.quiknode.pro";
const POST_BODY: &str = r#"{"jsonrpc":"2.0","method":"web3_clientVersion","params":[],"id":1}"#;
impl HttpsConnectivityTest {
pub fn new(test_count: u64, mixnet_client_timeout: u64) -> Self {
Self {
test_count: std::cmp::max(test_count, 1),
mixnet_client_timeout: Duration::from_secs(mixnet_client_timeout),
}
}
pub async fn run_tests(self, socks5_url: String) -> HttpsConnectivityResult {
let mut result = HttpsConnectivityResult::default();
let proxy = match reqwest::Proxy::all(socks5_url) {
Ok(p) => p,
Err(e) => {
result.error = Some(format!("Failed to create proxy: {}", e));
return result;
}
};
let client = match reqwest::Client::builder()
.proxy(proxy)
.timeout(self.mixnet_client_timeout)
.build()
{
Ok(c) => c,
Err(e) => {
result.error = Some(format!("Failed to build HTTP client: {}", e));
return result;
}
};
let mut successful_runs = 0u64;
for i in 1..self.test_count + 1 {
info!("Running test {}/{}", i, self.test_count);
let interim_res = self.perform_https_request(&client).await;
if interim_res.https_success
&& let Some(latency_ms) = interim_res.https_latency_ms
{
successful_runs += 1;
result.https_latency_ms = Some(
result
.https_latency_ms
.map_or(latency_ms, |existing| existing + latency_ms),
);
result.https_success = true;
result.https_status_code = interim_res.https_status_code;
info!("{}/{} latency: {}ms", i, self.test_count, latency_ms);
} else if let Some(new_error) = interim_res.error {
result.error = Some(result.error.map_or(new_error.clone(), |existing| {
format!("{},{}", existing, new_error)
}))
}
// too many failed runs: return early
if successful_runs < 2 && i - successful_runs > 2 {
// if < 2 runs, we don't have to calculate average before returning
return result;
}
}
result.https_latency_ms = result
.https_latency_ms
.map(|latency| latency / successful_runs);
info!(
"AVG latency over {} runs (in ms): {:?}",
successful_runs, result.https_latency_ms
);
result
}
async fn perform_https_request(&self, client: &reqwest::Client) -> HttpsConnectivityResult {
use tokio::time::Instant;
let mut result = HttpsConnectivityResult::default();
let start = Instant::now();
match tokio::time::timeout(
self.mixnet_client_timeout,
client
.post(TARGET_URL)
.header(reqwest::header::CONTENT_TYPE, "application/json")
.body(POST_BODY)
.send(),
)
.await
{
Ok(Ok(response)) => {
let elapsed = start.elapsed();
let status = response.status();
result.https_success = status.is_success();
result.https_status_code = Some(status.as_u16());
result.https_latency_ms = Some(elapsed.as_millis() as u64);
debug!(
"HTTPS test completed: status={}, latency={}ms",
status.as_u16(),
elapsed.as_millis()
);
}
Ok(Err(e)) => {
warn!("HTTPS request failed: {}", e);
if result.error.is_none() {
result.error = Some(format!("HTTPS request failed: {}", e));
}
}
Err(_) => {
warn!(
"HTTPS request timed out after {}s",
self.mixnet_client_timeout.as_secs()
);
if result.error.is_none() {
result.error = Some("HTTPS request timed out".to_string());
}
}
}
result
}
}
#[derive(Debug, Clone, Default)]
pub struct IpPingReplies {
pub ipr_tun_ip_v4: bool,
sdk/rust/nym-sdk/src/mixnet/client.rs
+7 -3
View File
@@ -8,7 +8,7 @@ use crate::mixnet::{CredentialStorage, MixnetClient, Recipient};
use crate::GatewayTransceiver;
use crate::NymNetworkDetails;
use crate::{Error, Result};
use log::{debug, info, warn};
use log::{debug, warn};
use nym_client_core::client::base_client::storage::gateways_storage::GatewayRegistration;
use nym_client_core::client::base_client::storage::helpers::{
get_active_gateway_identity, get_all_registered_identities, has_gateway_details,
@@ -601,9 +601,13 @@ where
);
let available_gateways = self.available_gateways().await?;
info!("Listing all available gateways in topology:");
debug!("Listing all available gateways in topology:");
for node in available_gateways.iter() {
info!("{}", node.identity_key.to_base58_string());
debug!(
"node_id={}, identity_key={}",
node.node_id,
node.identity_key.to_base58_string()
);
}
Ok(GatewaySetup::New {
sdk/rust/nym-sdk/src/mixnet/socks5_client.rs
+28 -1
View File
@@ -1,10 +1,13 @@
use std::time::Duration;
use nym_client_core::client::base_client::ClientState;
use nym_socks5_client_core::config::Socks5;
use nym_sphinx::addressing::clients::Recipient;
use nym_task::connections::LaneQueueLengths;
use nym_task::ShutdownTracker;
use tokio::sync::RwLockReadGuard;
use nym_topology::NymTopology;
use nym_topology::{NymRouteProvider, NymTopology, NymTopologyError};
use crate::mixnet::client::MixnetClientBuilder;
use crate::Result;
@@ -85,4 +88,28 @@ impl Socks5MixnetClient {
pub async fn disconnect(self) {
self.task_handle.shutdown().await;
}
/// Gets the current route provider if topology is available.
/// Returns `None` if topology is empty/not yet fetched.
async fn read_current_route_provider(&self) -> Option<RwLockReadGuard<'_, NymRouteProvider>> {
self.client_state
.topology_accessor
.current_route_provider()
.await
}
/// Wait for topology to become available, with a timeout.
/// Returns `Ok(())` when topology is ready, or `Err` if timeout is reached.
pub async fn wait_for_topology(&self, timeout: Duration) -> Result<(), NymTopologyError> {
let deadline = tokio::time::Instant::now() + timeout;
loop {
if self.read_current_route_provider().await.is_some() {
return Ok(());
}
if tokio::time::Instant::now() >= deadline {
return Err(NymTopologyError::EmptyNetworkTopology);
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
}