feat: atm0s-sdn (#2)

Cargo.toml

@@ -1,4 +1,5 @@
 [workspace]
+resolver = "2"
 
 members = [
     "crates/*",

README.md

@@ -1,40 +1,137 @@
-# Decentralized HomeAssistant Proxy
+# Decentralized reverse proxy for HomeAssistant, IoT and more
 
-This project aims to create an innovative decentralized relay server for HomeAssistant. The server allows contributions from anyone and provides a fast and optimized path between clients.
+This project aims to create an innovative decentralized reverse proxy server for HomeAssistant, IoT and more. The server allows contributions from anyone and provides a fast and optimized path between clients.
+
+If you find it interesting or believe it could be helpful, we welcome your contributions to the codebase or consider starring the repository to show your support and motivate our team!
 
 ## Features
 
-- **Decentralized:** The server is designed to be decentralized, allowing anyone to contribute and participate in the network.
+- **Decentralized:** The server is designed to be decentralized, allowing anyone to contribute and participate in the network. More details: [atm0s-sdn](https://github.com/8xff/atm0s-sdn)
 
 - **Fast Relay:** The relay server ensures fast and efficient communication between clients, optimizing the path for data transmission using the atm0s-sdn overlay network.
 
 - **Data Safety:** The TCP proxy used in this project is based on SNI (Server Name Indication), ensuring the safety and integrity of the transmitted data.
 
 - **Written in Rust:** The project is implemented in Rust, a modern and efficient programming language known for its performance and memory safety.
 
+- **No account required:** Each client will be identified by a public-private key, just connect and it works.
+
+## Check list
+
+- [x] Decentralized Routing, Relay between nodes
+- [x] TLS(SNI) tunneling
+- [x] HTTP tunneling
+- [x] Stats Dashboard
+- [x] Anonymous client
+- [ ] Anonymous server contribution (WIP)
+
+## Performance
+
+Bellow is benchmarking results with Mac M1, and all nodes is running localy, it it very early version so it can be improve after finish all features:
+
+- Direct http
+
+```bash
+wrk http://localhost:3000
+Running 10s test @ http://localhost:3000
+  2 threads and 10 connections
+  Thread Stats   Avg      Stdev     Max   +/- Stdev
+    Latency   113.58us   32.97us   3.68ms   77.81%
+    Req/Sec    41.60k     1.27k   43.00k    89.60%
+  836084 requests in 10.10s, 143.52MB read
+Requests/sec:  82780.91
+Transfer/sec:     14.21MB
+```
+
+- Single node
+
+```bash
+wrk http://localhost:3000
+Running 10s test @ http://localhost:3000
+  2 threads and 10 connections
+  Thread Stats   Avg      Stdev     Max   +/- Stdev
+    Latency   407.38us  489.34us  16.67ms   98.87%
+    Req/Sec    13.18k     1.67k   17.61k    77.72%
+  264801 requests in 10.10s, 45.46MB read
+Requests/sec:  26218.89
+Transfer/sec:      4.50MB
+```
+
+- Two nodes
+
+```bash
+wrk http://localhost:3000
+Running 10s test @ http://localhost:3000
+  2 threads and 10 connections
+  Thread Stats   Avg      Stdev     Max   +/- Stdev
+    Latency   678.58us  681.76us  19.83ms   98.50%
+    Req/Sec     8.06k     1.19k    9.60k    64.85%
+  162101 requests in 10.10s, 27.83MB read
+Requests/sec:  16049.83
+Transfer/sec:      2.76MB
+```
+
 ## Getting Started
 
 To get started with the Decentralized HomeAssistant Proxy, follow these steps:
 
 1. Clone the repository:
 
     ```shell
-    git clone https://github.com/your-username/decentralized-homeassistant-proxy.git
+    git clone https://github.com/8xFF/atm0s-reverse-proxy.git
     ```
 
 2. Build the project:
 
     ```shell
-    cd decentralized-homeassistant-proxy
-    cargo build
+    cd atm0s-reverse-proxy
+    cargo build --release
     ```
 
-3. Run the server:
+3. Run the server node1:
 
     ```shell
-    cargo run --release
+    ./target/release/relayer \
+        --api-port 10001 \
+        --http-port 11001 \
+        --https-port 12001 \
+        --connector-port 0.0.0.0:13001 \
+        --root-domain local.ha.8xff.io \
+        --sdn-node-id 1
     ```
 
+3. Run the server node2:
+
+    ```shell
+    ./target/release/relayer \
+        --api-port 10002 \
+        --http-port 11002 \
+        --https-port 12002 \
+        --connector-port 0.0.0.0:13002 \
+        --root-domain local.ha.8xff.io \
+        --sdn-node-id 2 \
+        --sdn-seeds '1@/ip4/192.168.1.39/udp/50001'
+    ```
+
+4. Run the client:
+
+    ```shell
+    ./target/release/agent \
+        --connector-protocol tcp \
+        --connector-addr 127.0.0.1:13001 \
+        --http-dest 127.0.0.1:8080 \
+        --https-dest 127.0.0.1:8443
+    ```
+    Client will print out assigned domain like: `a5fae3d220cb062c5aed9bd57d82f226.local.ha.8xff.io`, now we can access to local 8080 service with address:
+
+    - Proxy over single node1:
+    http://a5fae3d220cb062c5aed9bd57d82f226.local.ha.8xff.io:11001
+
+    - Proxy with relay over node2 -> node1:
+    http://a5fae3d220cb062c5aed9bd57d82f226.local.ha.8xff.io:11002
+
+    Note that above url only can access in same machine.
+
 ## Contributing
 
 Contributions are welcome! If you'd like to contribute to the project, please follow the guidelines outlined in [CONTRIBUTING.md](CONTRIBUTING.md).

crates/agent/node_local.sh

@@ -0,0 +1,5 @@
+cargo run --release -- \
+    --connector-protocol tcp \
+    --connector-addr 127.0.0.1:13001 \
+    --http-dest 127.0.0.1:8080 \
+    --https-dest 127.0.0.1:8443

crates/agent/src/main.rs

@@ -6,10 +6,10 @@ static A: System = System;
 use std::net::SocketAddr;
 
 use async_std::io::WriteExt;
-use clap::Parser;
+use clap::{Parser, ValueEnum};
 
-use connection::quic::QuicSubConnection;
-use futures::{select, FutureExt};
+use connection::tcp::TcpConnection;
+use futures::{select, AsyncRead, AsyncReadExt, AsyncWrite, FutureExt};
 use local_tunnel::tcp::LocalTcpTunnel;
 use protocol::key::LocalKey;
 use tracing_subscriber::{fmt, layer::*, util::SubscriberInitExt, EnvFilter};
@@ -22,13 +22,23 @@ use crate::{
 mod connection;
 mod local_tunnel;
 
+#[derive(ValueEnum, Debug, Clone)]
+enum Protocol {
+    Tcp,
+    Quic,
+}
+
 /// A HTTP and SNI HTTPs proxy for expose your local service to the internet.
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about = None)]
 struct Args {
     /// Address of relay server
-    #[arg(env, long, long, default_value = "127.0.0.1:33333")]
-    quic_connector_addr: SocketAddr,
+    #[arg(env, long)]
+    connector_addr: SocketAddr,
+
+    /// Protocol of relay server
+    #[arg(env, long)]
+    connector_protocol: Protocol,
 
     /// Http proxy dest
     #[arg(env, long, default_value = "127.0.0.1:8080")]
@@ -87,43 +97,72 @@ async fn main() {
     };
 
     loop {
-        log::info!("Connecting to connector... {}", args.quic_connector_addr);
-        if let Ok(mut connection) = QuicConnection::new(args.quic_connector_addr, &local_key).await
-        {
-            log::info!("Connection to connector is established");
-            loop {
-                match connection.recv().await {
-                    Ok(sub_connection) => {
-                        log::info!("recv sub_connection");
-                        async_std::task::spawn_local(run_connection(
-                            sub_connection,
-                            args.http_dest,
-                            args.https_dest,
-                        ));
-                    }
-                    Err(e) => {
-                        log::error!("recv sub_connection error: {}", e);
-                        break;
-                    }
+        log::info!(
+            "Connecting to connector... {:?} addr: {:?}",
+            args.connector_protocol,
+            args.connector_addr
+        );
+        match args.connector_protocol {
+            Protocol::Tcp => match TcpConnection::new(args.connector_addr, &local_key).await {
+                Ok(conn) => {
+                    log::info!("Connected to connector via tcp");
+                    run_loop(conn, args.http_dest, args.https_dest).await;
                 }
-            }
-            log::warn!("Connection to connector is closed, try to reconnect...");
+                Err(e) => {
+                    log::error!("Connect to connector via tcp error: {}", e);
+                }
+            },
+            Protocol::Quic => match QuicConnection::new(args.connector_addr, &local_key).await {
+                Ok(conn) => {
+                    log::info!("Connected to connector via quic");
+                    run_loop(conn, args.http_dest, args.https_dest).await;
+                }
+                Err(e) => {
+                    log::error!("Connect to connector via quic error: {}", e);
+                }
+            },
         }
         //TODO exponential backoff
         async_std::task::sleep(std::time::Duration::from_secs(1)).await;
     }
 }
 
-async fn run_connection(
-    sub_connection: QuicSubConnection,
+async fn run_loop<S, R, W>(
+    mut connection: impl Connection<S, R, W>,
     http_dest: SocketAddr,
     https_dest: SocketAddr,
-) {
+) where
+    S: SubConnection<R, W> + 'static,
+    R: AsyncRead + Send + Unpin + 'static,
+    W: AsyncWrite + Send + Unpin + 'static,
+{
+    log::info!("Connection to connector is established");
+    loop {
+        match connection.recv().await {
+            Ok(sub_connection) => {
+                log::info!("recv sub_connection");
+                async_std::task::spawn_local(run_connection(sub_connection, http_dest, https_dest));
+            }
+            Err(e) => {
+                log::error!("recv sub_connection error: {}", e);
+                break;
+            }
+        }
+    }
+    log::warn!("Connection to connector is closed, try to reconnect...");
+}
+
+async fn run_connection<S, R, W>(sub_connection: S, http_dest: SocketAddr, https_dest: SocketAddr)
+where
+    S: SubConnection<R, W> + 'static,
+    R: AsyncRead + Send + Unpin,
+    W: AsyncWrite + Send + Unpin,
+{
     log::info!("sub_connection pipe to local_tunnel start");
     let (mut reader1, mut writer1) = sub_connection.split();
     let mut first_pkt = [0u8; 4096];
     let (local_tunnel, first_pkt_len) = match reader1.read(&mut first_pkt).await {
-        Ok(Some(first_pkt_len)) => {
+        Ok(first_pkt_len) => {
             log::info!("first pkt size: {}", first_pkt_len);
             if first_pkt_len == 0 {
                 log::error!("first pkt size is 0 => close");
@@ -137,10 +176,6 @@ async fn run_connection(
                 (LocalTcpTunnel::new(http_dest).await, first_pkt_len)
             }
         }
-        Ok(None) => {
-            log::error!("read first pkt error: eof");
-            return;
-        }
         Err(e) => {
             log::error!("read first pkt error: {}", e);
             return;

crates/protocol/src/cluster.rs

@@ -0,0 +1,39 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ClusterTunnelRequest {
+    pub domain: String,
+}
+
+impl From<&ClusterTunnelRequest> for Vec<u8> {
+    fn from(resp: &ClusterTunnelRequest) -> Self {
+        bincode::serialize(resp).expect("Should ok")
+    }
+}
+
+impl TryFrom<&[u8]> for ClusterTunnelRequest {
+    type Error = bincode::Error;
+
+    fn try_from(buf: &[u8]) -> Result<Self, Self::Error> {
+        bincode::deserialize(buf)
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ClusterTunnelResponse {
+    pub success: bool,
+}
+
+impl From<&ClusterTunnelResponse> for Vec<u8> {
+    fn from(resp: &ClusterTunnelResponse) -> Self {
+        bincode::serialize(resp).expect("Should ok")
+    }
+}
+
+impl TryFrom<&[u8]> for ClusterTunnelResponse {
+    type Error = bincode::Error;
+
+    fn try_from(buf: &[u8]) -> Result<Self, Self::Error> {
+        bincode::deserialize(buf)
+    }
+}

crates/protocol/src/lib.rs

@@ -1,2 +1,3 @@
+pub mod cluster;
 pub mod key;
 pub mod rpc;

crates/relayer/Cargo.toml

@@ -20,6 +20,7 @@ metrics = { version = "0.21.1" }
 quinn = { version = "0.10.2", default-features = false, features = ["native-certs", "tls-rustls", "log", "runtime-async-std", "futures-io", "ring"] }
 rustls = { version = "0.21.0", default-features = false, features = ["quic", "dangerous_configuration"] }
 rcgen = "0.12.0"
+atm0s-sdn = { version = "0.1.7", features = ["all"]}
 
 [features]
 default = []

crates/relayer/run_local_node1.sh

@@ -0,0 +1,7 @@
+cargo run --release -- \
+    --api-port 10001 \
+    --http-port 11001 \
+    --https-port 12001 \
+    --connector-port 0.0.0.0:13001 \
+    --root-domain local.ha.8xff.io \
+    --sdn-node-id 1

crates/relayer/run_local_node2.sh

@@ -0,0 +1,8 @@
+cargo run --release -- \
+    --api-port 10002 \
+    --http-port 11002 \
+    --https-port 12002 \
+    --connector-port 0.0.0.0:13002 \
+    --root-domain local.ha.8xff.io \
+    --sdn-node-id 2 \
+    --sdn-seeds '1@/ip4/192.168.1.39/udp/50001'