first pass tester

tools/internal/mixnet-check-all-gateways/Cargo.toml

@@ -27,3 +27,6 @@ tokio = { workspace = true, features = ["full"] }
 reqwest = { workspace = true, features = ["json"] }
 serde_json.workspace = true
 futures = { workspace = true }
+uuid = { version = "1", features = ["v4", "serde"] }
+bincode = "1.0"
+dirs.workspace = true

tools/internal/mixnet-check-all-gateways/src/main.rs

@@ -1,22 +1,49 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use echo_server::NymEchoServer;
 use futures::stream::StreamExt;
 use nym_bin_common::logging::setup_logging;
 use nym_crypto::asymmetric::ed25519;
 use nym_sdk::mixnet;
-use nym_sdk::mixnet::{AnonymousSenderTag, MixnetMessageSender, ReconstructedMessage};
+use nym_sdk::mixnet::{
+    AnonymousSenderTag, IncludedSurbs, MixnetMessageSender, ReconstructedMessage,
+};
 use reqwest::{self, Url};
-use serde_json::Value;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+use std::fs::OpenOptions;
+use std::io::Write;
 use std::time::Duration;
-use tokio::time;
 use tokio::time::timeout;
+use tokio::time::{self, Timeout};
+#[path = "utils.rs"]
+// TODO make these exportable from tcp_proxy module and then import from there, ditto with echo server lib
+mod utils;
+use utils::{Payload, ProxiedMessage};
+
+const TIMEOUT: u64 = 10; // message ping timeout
+
+#[derive(Serialize, Deserialize, Debug)]
+struct TestResult {
+    entry_gw: String,
+    exit_gw: String,
+    error: TestError,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+enum TestError {
+    Timeout,
+    NoMessage,
+    None,
+    CouldNotCreateEchoServer(String),
+    Other(String),
+}
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    setup_logging();
+    // setup_logging(); // TODO think about parsing and noise here, could just parse on errors from all libs and then have info from here?
     let entry_gw_keys = reqwest_and_parse(
         Url::parse("https://validator.nymtech.net/api/v1/unstable/nym-nodes/skimmed/entry-gateways/all?no_legacy=true").unwrap(),
         "EntryGateway",
@@ -44,18 +71,50 @@ async fn main() -> Result<()> {
     // );
     println!("got {} exit gws", exit_gw_keys.len(),);
 
-    for gw in exit_gw_keys.clone() {
-        println!("{}", gw);
+    let mut port_range: u64 = 9000; // port that we start iterating upwards from, will go from port_range to (port_range + exit_gws.len()) by the end of the run
 
-        let mut echo_server = NymEchoServer::new(
-            Some(ed25519::PublicKey::from_base58_string(
-                exit_gw_keys[0].clone(),
-            )?),
-            None,
+    for exit_gw in exit_gw_keys.clone() {
+        println!("creating echo server connecting to {}", exit_gw);
+
+        let filepath = format!("./src/results/{}.json", exit_gw.clone());
+        let mut results = OpenOptions::new()
+            .read(true)
+            .write(true) // .append(true)
+            .create(true)
+            .open(filepath)?;
+
+        let mut results_vec: Vec<Value> = Vec::new();
+        let home_dir = dirs::home_dir().expect("Unable to get home directory");
+        let mut echo_server = match NymEchoServer::new(
+            Some(ed25519::PublicKey::from_base58_string(&exit_gw)?),
+            Some(
+                format!(
+                    "{}/tmp/nym-proxy-server-config-{}",
+                    home_dir.display(),
+                    &exit_gw
+                )
+                .as_str(),
+            ),
             "../../../envs/mainnet.env".to_string(), // make const
-            "9000", // when you run concurrently you can iterate through a port range here
+            port_range.to_string().as_str(),
         )
-        .await?;
+        .await
+        {
+            Ok(echo_server) => echo_server,
+            Err(err) => {
+                let res = TestResult {
+                    entry_gw: "".to_string(),
+                    exit_gw: exit_gw.clone(),
+                    error: TestError::CouldNotCreateEchoServer(err.to_string()),
+                };
+                results_vec.push(json!(res));
+                let json_array = json!(results_vec);
+                println!("{json_array}");
+                results.write_all(json_array.to_string().as_bytes())?;
+                continue;
+            }
+        };
+        port_range += 1;
 
         let echo_addr = echo_server.nym_address().await;
         println!("echo addr: {echo_addr}");
@@ -68,43 +127,97 @@ async fn main() -> Result<()> {
         // dumb sleep to let it startup
         time::sleep(Duration::from_secs(5)).await;
 
-        for gw in entry_gw_keys.clone() {
+        for entry_gw in entry_gw_keys.clone() {
             let builder = mixnet::MixnetClientBuilder::new_ephemeral()
-                .request_gateway(gw)
+                .request_gateway(entry_gw.clone())
                 .build()?;
 
             let mut client = match builder.connect_to_mixnet().await {
                 Ok(client) => client,
                 Err(err) => {
-                    println!("failed to connect: {err}");
-                    return Err(err.into());
+                    let res = TestResult {
+                        entry_gw: entry_gw.clone(),
+                        exit_gw: exit_gw.clone(),
+                        error: TestError::Other(err.to_string()),
+                    };
+                    println!("{res:#?}");
+                    results_vec.push(json!(res));
+                    // println!("failed to connect: {err}");
+                    continue;
                 }
             };
-            let our_address = client.nym_address();
-            println!("{our_address}");
-            client.send_plain_message(echo_addr, "ping").await.unwrap();
 
-            match timeout(Duration::from_secs(5), client.next()).await {
+            let test_address = client.nym_address();
+            println!("currently testing entry gateway: {test_address}");
+
+            // Has to be ProxiedMessage for the moment which is slightly annoying until I
+            // modify the ProxyServer to just stupidly echo back whatever it gets in a
+            // ReconstructedMessage format if it can't deseralise it to a ProxiedMessage
+            let session_id = uuid::Uuid::new_v4();
+            let message_id = 0;
+            let outgoing = ProxiedMessage::new(
+                Payload::Data("echo test".as_bytes().to_vec()),
+                session_id,
+                message_id,
+            );
+            let coded_message = bincode::serialize(&outgoing).unwrap();
+
+            match client
+                .send_message(echo_addr, &coded_message, IncludedSurbs::Amount(10))
+                .await
+            {
+                Ok(_) => {
+                    println!("Message sent");
+                }
+                Err(err) => {
+                    let res = TestResult {
+                        entry_gw: entry_gw.clone(),
+                        exit_gw: exit_gw.clone(),
+                        error: TestError::Other(err.to_string()),
+                    };
+                    println!("{res:#?}");
+                    results_vec.push(json!(res));
+                    continue;
+                }
+            };
+
+            let res = match timeout(Duration::from_secs(TIMEOUT), client.next()).await {
                 Err(_timeout) => {
                     println!("timed out");
+                    TestResult {
+                        entry_gw,
+                        exit_gw: exit_gw.clone(),
+                        error: TestError::Timeout,
+                    }
                 }
-                Ok(Some(received)) => match String::from_utf8(received.message) {
-                    Ok(message) => {
-                        println!("received '{message}' back!");
+                Ok(Some(received)) => {
+                    let incoming: ProxiedMessage = bincode::deserialize(&received.message).unwrap();
+                    println!("\ngot echo: {incoming}\n");
+                    // TODO check incoming is same as outgoing else make a MangledReply err type
+                    TestResult {
+                        entry_gw,
+                        exit_gw: exit_gw.clone(),
+                        error: TestError::None,
                     }
-                    Err(err) => {
-                        println!("the received message got malformed on the way to us: {err}");
-                    }
-                },
+                }
                 Ok(None) => {
                     println!("failed to receive any message back...");
+                    TestResult {
+                        entry_gw,
+                        exit_gw: exit_gw.clone(),
+                        error: TestError::NoMessage,
+                    }
                 }
-            }
+            };
+            println!("{res:#?}");
+            results_vec.push(json!(res));
             println!("disconnecting the client before shutting down...");
             client.disconnect().await;
         }
+        let json_array = json!(results_vec);
+        println!("{json_array}");
+        results.write_all(json_array.to_string().as_bytes())?;
     }
-
     Ok(())
 }
 
@@ -121,7 +234,7 @@ fn filter_gateway_keys(json: &Value, key: &str) -> Result<Vec<String>> {
     if let Some(nodes) = json["nodes"]["data"].as_array() {
         for node in nodes {
             if let Some(performance) = node.get("performance").and_then(|v| v.as_str()) {
-                let performance_value: f64 = performance.parse().unwrap_or(0.0);
+                let performance_value: f64 = performance.parse().unwrap_or(0.0); // TODO could make this a const @ top?
 
                 let inactive = node.get("role").and_then(|v| v.as_str()) == Some("Inactive");
 
@@ -141,7 +254,9 @@ fn filter_gateway_keys(json: &Value, key: &str) -> Result<Vec<String>> {
             }
         }
     } else {
-        println!("No nodes found ");
+        // TODO make error and return / break
+        // println!("No nodes found ");
+        return Err(anyhow!("Could not parse any gateways"));
     }
     Ok(filtered_keys)
 }

tools/internal/mixnet-check-all-gateways/src/scratch

@@ -0,0 +1,35 @@
+
+        // let echo = mixnet::MixnetClientBuilder::new_ephemeral()
+        //     .request_gateway(gw)
+        //     .build()?;
+        // let mut client = echo.connect_to_mixnet().await?;
+        // let echo_addr = client.nym_address().clone();
+
+        // tokio::task::spawn(async move {
+        //     loop {
+        //         println!("Waiting for message\n");
+        //         let mut message: Vec<ReconstructedMessage> = Vec::new();
+        //         while let Some(new_message) = client.wait_for_messages().await {
+        //             if new_message.is_empty() {
+        //                 continue;
+        //             }
+        //             message = new_message;
+        //             break;
+        //         }
+
+        //         let mut parsed = String::new();
+        //         if let Some(r) = message.first() {
+        //             parsed = String::from_utf8(r.message.clone()).unwrap();
+        //         }
+        //         let return_recipient: AnonymousSenderTag = message[0].sender_tag.unwrap();
+        //         println!("\nReceived: {} from {}", parsed, return_recipient);
+
+        //         println!("Replying using SURBs");
+        //         client
+        //             .send_reply(return_recipient, parsed.to_string())
+        //             .await
+        //             .unwrap();
+
+        //         tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+        //     }
+        // });

tools/internal/mixnet-check-all-gateways/src/utils.rs

@@ -0,0 +1,54 @@
+use serde::{Deserialize, Serialize};
+use std::fmt;
+use uuid::Uuid;
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct ProxiedMessage {
+    pub message: Payload,
+    session_id: Uuid,
+    message_id: u16,
+}
+
+impl ProxiedMessage {
+    pub fn new(message: Payload, session_id: Uuid, message_id: u16) -> Self {
+        ProxiedMessage {
+            message,
+            session_id,
+            message_id,
+        }
+    }
+
+    pub fn message(&self) -> &Payload {
+        &self.message
+    }
+
+    pub fn session_id(&self) -> Uuid {
+        self.session_id
+    }
+
+    pub fn message_id(&self) -> u16 {
+        self.message_id
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+pub enum Payload {
+    Data(Vec<u8>),
+    Close,
+}
+
+impl fmt::Display for ProxiedMessage {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        let message = match self.message() {
+            Payload::Data(ref data) => format!("Data({})", data.len()),
+            Payload::Close => "Close".to_string(),
+        };
+        write!(
+            f,
+            "ProxiedMessage {{ message: {}, session_id: {}, message_id: {} }}",
+            message,
+            self.session_id(),
+            self.message_id()
+        )
+    }
+}