first pass tcpproxy

documentation/docs/pages/developers/rust/tcpproxy.mdx

@@ -1 +1,5 @@
 # TcpProxy Module
+
+This module exposes the `TcpProxyClient` and the `TcpProxyServer` which can be used to proxy traffic through the Mixnet in a way that is more familiar to developers than the methods exposed by the [`Mixnet` module](./mxinet).
+
+Both `Client` and `Server` are intended to be initialised and then run in a background thread, exposing a configurable `localhost` socket which developers can read/write/stream to without having to worry about the [message-based](../concepts/messages) nature of sending and receiving traffic to/from the Mixnet.

documentation/docs/pages/developers/rust/tcpproxy/_meta.json

@@ -0,0 +1,4 @@
+{
+  "architecture": "Architecture",
+  "examples": "Examples"
+}

documentation/docs/pages/developers/rust/tcpproxy/architecture.mdx

@@ -0,0 +1,31 @@
+import { Callout } from 'nextra/components'
+
+# Architecture
+
+tcpproxyclient is configurable with the local listen address and the timeout for client sessions. It is an abstraction layer which might also serve as a model for people wanting to build other connection logics ontop of the Mixnet module.
+
+TODO DIAGRAM
+
+## Motivations
+The motivation behind the creation of the `TcpProxy` module is to allow developers to interact with the Mixnet in a way that is far more familiar to them: simply setting up a connection with a transport, being returned a socket, and then being able to stream data to/from it, similar to something like the Tor [`arti`](https://gitlab.torproject.org/tpo/core/arti/-/tree/main/crates/arti-client) client.
+
+<Callout type="info" emoji="ℹ️">
+  This is an initial version of the module which we are releasing to developers to experiment with. If you run into problems or any functionality that is missing, do reach out on Matrix and let us know.
+
+  Furthermore we will be working on optimisations to the module over time - most of this will occur under the hood (e.g. implementing a configurable connection pool for the `ProxyClient`), but all updates will occur according to SemVer, so don't worry about breaking changes!
+</Callout>
+
+## Clients
+Each of the sub-modules exposed by the `TcpProxy` deal with Nym clients in a different way.
+- the `NymProxyClient` creates an ephemeral client per new TCP connection, which is closed according to the configurable timeout: we map one ephemeral client per TCP connection. This is to deal with multiple simultaneous streams. In the future, this will be superceded by a connection pool in order to speed up new connections.
+- the `NymProxyServer` has a single Nym client with a persistent identity.
+
+## Sessions
+We have implemented session management and message ordering, where messages are wrapped in a session ID per connection, with individual messages being given an incrememting message ID. Once all the messages have been sent, the `NymProxyClient` then sends a `Close` message as the last outgoing message. This is to notify the `NymProxyServer` that there are no more outbound messages for this session, and that it can trigger the session timeout.
+
+The reason we need to implement session management and message IDs is due to the fact that the Mixnet guarantees message delivery but not message ordering: in the case of trying to e.g. send gRPC protobuf through the Mixnet, ordering is required so that a buffer is not split across Sphinx packet payloads, and that the 2nd half of the frame is not passed upstream to the gRPC parser before the 1st half, even if it is received first.
+
+TODO DIAGRAM
+
+## session timeouts
+todo

documentation/docs/pages/developers/rust/tcpproxy/examples/_meta.json

@@ -0,0 +1,4 @@
+{
+  "singleconn": "Single Connection",
+  "multiconn": "Multi Connection"
+}

documentation/docs/pages/developers/rust/tcpproxy/examples/multiconn.mdx

@@ -0,0 +1,152 @@
+# Multi Connection Example
+
+This example starts off several Tcp connections on a loop to a remote endpoint: in this case the `TcpListener` behind the `NymProxyServer` instance on the echo server found in
+[`nym/tools/echo-server/`](https://github.com/nymtech/nym/tree/develop/tools/echo-server). It pipes a few messages to it, logs the replies, and keeps track of the number of replies received per connection.
+
+```rust
+use nym_sdk::mixnet::Recipient;
+use nym_sdk::tcp_proxy;
+use rand::rngs::SmallRng;
+use rand::Rng;
+use rand::SeedableRng;
+use serde::{Deserialize, Serialize};
+use std::env;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpStream;
+use tokio::signal;
+use tokio_stream::StreamExt;
+use tokio_util::codec;
+
+#[derive(Serialize, Deserialize, Debug)]
+struct ExampleMessage {
+    message_id: i8,
+    message_bytes: Vec<u8>,
+    tcp_conn: i8,
+}
+
+// To run:
+// - run the echo server with `cargo run`
+// - run this example with `cargo run --example tcp_proxy_multistream -- <ECHO_SERVER_NYM_ADDRESS> <ENV_FILE_PATH> <CLIENT_PORT>` e.g.
+// cargo run --example tcp_proxy_multistream -- DMHyxo8n6sKWHHTVvjRVDxDSMX8gYXRU1AQ6UpwsrWiB.6STYCWGWyRxqn2juWdgjMkAMsT9EaAzPpLWq5zkS68MB@CJG5zTcmoLijmDrtAiLV9PZHxNz8LQu6hmgA89V2RxxL ../../../envs/canary.env 8080
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let server_address = env::args().nth(1).expect("Server address not provided");
+    let server: Recipient =
+        Recipient::try_from_base58_string(&server_address).expect("Invalid server address");
+
+    // Comment this out to just see println! statements from this example.
+    // Nym client logging is very informative but quite verbose.
+    // The Message Decay related logging gives you an ideas of the internals of the proxy message ordering: you need to switch
+    // to DEBUG to see the contents of the msg buffer, sphinx packet chunking, etc.
+    tracing_subscriber::fmt()
+        .with_max_level(tracing::Level::INFO)
+        .init();
+
+    let env_path = env::args().nth(2).expect("Env file not specified");
+    let env = env_path.to_string();
+
+    let listen_port = env::args().nth(3).expect("Port not specified");
+
+    // Within the TcpProxyClient, individual client shutdown is triggered by the timeout.
+    let proxy_client =
+        tcp_proxy::NymProxyClient::new(server, "127.0.0.1", &listen_port, 45, Some(env)).await?;
+
+    tokio::spawn(async move {
+        proxy_client.run().await?;
+        Ok::<(), anyhow::Error>(())
+    });
+
+    println!("waiting for everything to be set up..");
+    tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+    println!("done. sending bytes");
+
+    // In the info traces you will see the different session IDs being set up, one for each TcpStream.
+    for i in 0..4 {
+        let conn_id = i;
+        println!("Starting TCP connection {}", conn_id);
+        let local_tcp_addr = format!("127.0.0.1:{}", listen_port.clone());
+        tokio::spawn(async move {
+            // Now the client and server proxies are running we can create and pipe traffic to/from
+            // a socket on the same port as our ProxyClient instance as if we were just communicating
+            // between a client and host via a normal TcpStream - albeit with a decent amount of additional latency.
+            //
+            // The assumption regarding integration is that you know what you're sending, and will do proper
+            // framing before and after, know what data types you're expecting; the proxies are just piping bytes
+            // back and forth using tokio's `Bytecodec` under the hood.
+
+            let stream = TcpStream::connect(local_tcp_addr).await?;
+            let (read, mut write) = stream.into_split();
+
+            // Lets just send a bunch of messages to the server with variable delays between them, with a message and tcp connection ids to keep track of ordering on the server side (for illustrative purposes **only**; keeping track of anonymous replies is handled by the proxy under the hood with Single Use Reply Blocks (SURBs); for this illustration we want some kind of app-level message id, but irl most of the time you'll probably be parsing on e.g. the incoming response type instead)
+            tokio::spawn(async move {
+                for i in 0..4 {
+                    let mut rng = SmallRng::from_entropy();
+                    let delay: f64 = rng.gen_range(2.5..5.0);
+                    tokio::time::sleep(tokio::time::Duration::from_secs_f64(delay)).await;
+                    let random_bytes = gen_bytes_fixed(i as usize);
+                    let msg = ExampleMessage {
+                        message_id: i,
+                        message_bytes: random_bytes,
+                        tcp_conn: conn_id,
+                    };
+                    let serialised = bincode::serialize(&msg)?;
+                    write
+                        .write_all(&serialised)
+                        .await
+                        .expect("couldn't write to stream");
+                    println!(
+                        ">> client sent {}: {} bytes on conn {}",
+                        &i,
+                        msg.message_bytes.len(),
+                        &conn_id
+                    );
+                }
+                Ok::<(), anyhow::Error>(())
+            });
+
+            tokio::spawn(async move {
+                let mut reply_counter = 0;
+                let codec = codec::BytesCodec::new();
+                let mut framed_read = codec::FramedRead::new(read, codec);
+                while let Some(Ok(bytes)) = framed_read.next().await {
+                    match bincode::deserialize::<ExampleMessage>(&bytes) {
+                        Ok(msg) => {
+                            println!(
+                                "<< client received {}: {} bytes on conn {}",
+                                msg.message_id,
+                                msg.message_bytes.len(),
+                                msg.tcp_conn
+                            );
+                            reply_counter += 1;
+                            println!(
+                                "tcp connection {} replies received {}/4",
+                                msg.tcp_conn, reply_counter
+                            );
+                        }
+                        Err(e) => {
+                            println!("<< client received something that wasn't an example message of {} bytes. error: {}", bytes.len(), e);
+                        }
+                    }
+                }
+            });
+            Ok::<(), anyhow::Error>(())
+        });
+        let mut rng = SmallRng::from_entropy();
+        let delay: f64 = rng.gen_range(4.5..7.0);
+        tokio::time::sleep(tokio::time::Duration::from_secs_f64(delay)).await;
+    }
+
+    // Once timeout is passed, you can either wait for graceful shutdown or just hard stop it.
+    signal::ctrl_c().await?;
+    println!("CTRL+C received, shutting down");
+    Ok(())
+}
+
+// emulate a series of small messages followed by a closing larger one
+fn gen_bytes_fixed(i: usize) -> Vec<u8> {
+    let amounts = [10, 15, 50, 1000];
+    let len = amounts[i];
+    let mut rng = rand::thread_rng();
+    (0..len).map(|_| rng.gen::<u8>()).collect()
+}
+```

documentation/docs/pages/developers/rust/tcpproxy/examples/singleconn.mdx

@@ -0,0 +1,193 @@
+# Single Connection Example
+
+This is a basic example which opens a single TCP connection and writes a bunch of messages between a client and some 'echo server' logic, so only uses a single session under the hood and doesn't really show off the message ordering capabilities; this is mainly just a quick introductory illustration on how:
+- the mixnet does message ordering
+- the NymProxyClient and NymProxyServer can be hooked into and used to communicate between two otherwise pretty vanilla TcpStreams
+
+For a more irl example check the [multi connection example](./multiconn).
+
+```rust
+use nym_sdk::tcp_proxy;
+use rand::rngs::SmallRng;
+use rand::{Rng, SeedableRng};
+use serde::{Deserialize, Serialize};
+use std::env;
+use std::fs;
+use std::sync::atomic::{AtomicU8, Ordering};
+use tokio::io::AsyncWriteExt;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::signal;
+use tokio_stream::StreamExt;
+use tokio_util::codec;
+
+#[derive(Serialize, Deserialize, Debug)]
+struct ExampleMessage {
+    message_id: i8,
+    message_bytes: Vec<u8>,
+}
+
+
+// Run this with:
+// `cargo run --example tcp_proxy_single_connection <SERVER_LISTEN_PORT> <ENV_FILE_PATH> <CLIENT_LISTEN_PATH>` e.g.
+// `cargo run --example tcp_proxy_single_connection 8081 ../../../envs/canary.env 8080 `
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // Keep track of sent/received messages
+    let counter = AtomicU8::new(0);
+
+    // Comment this out to just see println! statements from this example, as Nym client logging is very informative but quite verbose.
+    // The Message Decay related logging gives you an ideas of the internals of the proxy message ordering. To see the contents of the msg buffer, sphinx packet chunking, etc change the tracing::Level to DEBUG.
+    tracing_subscriber::fmt()
+        .with_max_level(tracing::Level::INFO)
+        .init();
+
+    let server_port = env::args()
+        .nth(1)
+        .expect("Server listen port not specified");
+    let upstream_tcp_addr = format!("127.0.0.1:{}", server_port);
+
+    // This dir gets cleaned up at the end: NOTE if you switch env between tests without letting the file do the automatic cleanup, make sure to manually remove this directory up before running again, otherwise your client will attempt to use these keys for the new env
+    let home_dir = dirs::home_dir().expect("Unable to get home directory");
+    let conf_path = format!("{}/tmp/nym-proxy-server-config", home_dir.display());
+
+    let env_path = env::args().nth(2).expect("Env file not specified");
+    let env = env_path.to_string();
+    let client_port = env::args().nth(3).expect("Port not specified");
+
+    let mut proxy_server =
+        tcp_proxy::NymProxyServer::new(&upstream_tcp_addr, &conf_path, Some(env_path.clone()))
+            .await?;
+    let proxy_nym_addr = proxy_server.nym_address();
+
+    // We'll run the instance with a long timeout since we're sending everything down the same Tcp connection, so should be using a single session.
+    // Within the TcpProxyClient, individual client shutdown is triggered by the timeout.
+    let proxy_client =
+        tcp_proxy::NymProxyClient::new(*proxy_nym_addr, "127.0.0.1", &client_port, 60, Some(env))
+            .await?;
+
+    tokio::spawn(async move {
+        proxy_server.run_with_shutdown().await?;
+        Ok::<(), anyhow::Error>(())
+    });
+
+    tokio::spawn(async move {
+        proxy_client.run().await?;
+        Ok::<(), anyhow::Error>(())
+    });
+
+    // 'Server side' thread: echo back incoming as response to the messages sent in the 'client side' thread below
+    tokio::spawn(async move {
+        let listener = TcpListener::bind(upstream_tcp_addr).await?;
+        loop {
+            let (socket, _) = listener.accept().await.unwrap();
+            let (read, mut write) = socket.into_split();
+            let codec = codec::BytesCodec::new();
+            let mut framed_read = codec::FramedRead::new(read, codec);
+            while let Some(Ok(bytes)) = framed_read.next().await {
+                match bincode::deserialize::<ExampleMessage>(&bytes) {
+                    Ok(msg) => {
+                        println!(
+                            "<< server received {}: {} bytes",
+                            msg.message_id,
+                            msg.message_bytes.len()
+                        );
+                        let msg = ExampleMessage {
+                            message_id: msg.message_id,
+                            message_bytes: msg.message_bytes,
+                        };
+                        let serialised = bincode::serialize(&msg)?;
+                        write
+                            .write_all(&serialised)
+                            .await
+                            .expect("couldnt send reply");
+                        println!(
+                            ">> server sent {}: {} bytes",
+                            msg.message_id,
+                            msg.message_bytes.len()
+                        );
+                    }
+                    Err(e) => {
+                        println!("<< server received something that wasn't an example message of {} bytes. error: {}", bytes.len(), e);
+                    }
+                }
+            }
+        }
+        #[allow(unreachable_code)]
+        Ok::<(), anyhow::Error>(())
+    });
+
+    // Just wait for Nym clients to connect, TCP clients to bind, etc.
+    println!("waiting for everything to be set up..");
+    tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+    println!("done. sending bytes");
+
+    // Now the client and server proxies are running we can create and pipe traffic to/from
+    // a socket on the same port as our ProxyClient instance as if we were just communicating
+    // between a client and host via a normal TcpStream - albeit with a decent amount of additional latency.
+    //
+    // The assumption regarding integration is that you know what you're sending, and will do proper
+    // framing before and after, know what data types you're expecting, etc; the proxies are just piping bytes
+    // back and forth using tokio's `Bytecodec` under the hood.
+    let local_tcp_addr = format!("127.0.0.1:{}", client_port);
+    let stream = TcpStream::connect(local_tcp_addr).await?;
+    let (read, mut write) = stream.into_split();
+
+    // 'Client side' thread; lets just send a bunch of messages to the server with variable delays between them, with an id to keep track of ordering in the printlns; the mixnet only guarantees message delivery, not ordering. You might not be necessarily streaming traffic in this manner IRL, but this example is a good illustration of how messages travel through the mixnet.
+    // - On the level of individual messages broken into multiple packets, the Proxy abstraction deals with making sure that everything is sent between the sockets in the corrent order.
+    // - On the level of different messages, this is not enforced: you might see in the logs that message 1 arrives at the server and is reconstructed after message 2.
+    tokio::spawn(async move {
+        let mut rng = SmallRng::from_entropy();
+        for i in 0..10 {
+            let random_bytes = gen_bytes_fixed(i as usize);
+            let msg = ExampleMessage {
+                message_id: i,
+                message_bytes: random_bytes,
+            };
+            let serialised = bincode::serialize(&msg)?;
+            write
+                .write_all(&serialised)
+                .await
+                .expect("couldn't write to stream");
+            println!(">> client sent {}: {} bytes", &i, msg.message_bytes.len());
+            let delay = rng.gen_range(3.0..7.0);
+            tokio::time::sleep(tokio::time::Duration::from_secs_f64(delay)).await;
+        }
+        Ok::<(), anyhow::Error>(())
+    });
+
+    let codec = codec::BytesCodec::new();
+    let mut framed_read = codec::FramedRead::new(read, codec);
+    while let Some(Ok(bytes)) = framed_read.next().await {
+        match bincode::deserialize::<ExampleMessage>(&bytes) {
+            Ok(msg) => {
+                println!(
+                    "<< client received {}: {} bytes",
+                    msg.message_id,
+                    msg.message_bytes.len()
+                );
+                counter.fetch_add(1, Ordering::SeqCst);
+                println!(
+                    ":: messages received back: {:?}/10",
+                    counter.load(Ordering::SeqCst)
+                );
+            }
+            Err(e) => {
+                println!("<< client received something that wasn't an example message of {} bytes. error: {}", bytes.len(), e);
+            }
+        }
+    }
+
+    // Once timeout is passed, you can either wait for graceful shutdown or just hard stop it.
+    signal::ctrl_c().await?;
+    println!(":: CTRL+C received, shutting down + cleanup up proxy server config files");
+    fs::remove_dir_all(conf_path)?;
+    Ok(())
+}
+
+fn gen_bytes_fixed(i: usize) -> Vec<u8> {
+    let amounts = vec![1, 10, 50, 100, 150, 200, 350, 500, 750, 1000];
+    let len = amounts[i];
+    let mut rng = rand::thread_rng();
+    (0..len).map(|_| rng.gen::<u8>()).collect()
+}
+```