chore(phantun) release v0.3.2

connection performance by 300%

.github/workflows/release.yml

@@ -31,7 +31,7 @@ jobs:
           - mipsel-unknown-linux-musl
 
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - uses: actions-rs/toolchain@v1
         with:
           toolchain: stable

.github/workflows/rust.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
     - uses: actions-rs/toolchain@v1
       with:
         toolchain: stable

LICENSE-APACHE

@@ -186,7 +186,7 @@ APPENDIX: How to apply the Apache License to your work.
    same "printed page" as the copyright notice for easier
    identification within third-party archives.
 
-Copyright 2014-2021 The Rust Project Developers
+Copyright 2021-2022 Datong Sun (dndx@idndx.com)
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

LICENSE-MIT

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2014-2021 The Rust Project Developers
+Copyright (c) 2021-2022 Datong Sun (dndx@idndx.com)
 
 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated

README.md

@@ -2,6 +2,9 @@
 
 A lightweight and fast UDP to TCP obfuscator.
 
+![GitHub Workflow Status](https://img.shields.io/github/workflow/status/dndx/phantun/Rust?style=flat-square)
+![docs.rs](https://img.shields.io/docsrs/fake-tcp)
+
 Table of Contents
 =================
 
@@ -24,6 +27,7 @@ Table of Contents
 * [MTU overhead](#mtu-overhead)
     * [MTU calculation for WireGuard](#mtu-calculation-for-wireguard)
 * [Version compatibility](#version-compatibility)
+* [Documentations](#documentations)
 * [Performance](#performance)
 * [Future plans](#future-plans)
 * [Compariation to udp2raw](#compariation-to-udp2raw)
@@ -31,16 +35,16 @@ Table of Contents
 
 # Latest release
 
-[v0.2.2](https://github.com/dndx/phantun/releases/tag/v0.2.2)
+[v0.3.1](https://github.com/dndx/phantun/releases/tag/v0.3.1)
 
 # Overview
 
-Phanton is a project that obfuscated UDP packets into TCP connections. It aims to
+Phantun is a project that obfuscated UDP packets into TCP connections. It aims to
 achieve maximum performance with minimum processing and encapsulation overhead.
 
 It is commonly used in environments where UDP is blocked/throttled but TCP is allowed through.
 
-Phanton simply converts a stream of UDP packets into obfuscated TCP stream packets. The TCP stack
+Phantun simply converts a stream of UDP packets into obfuscated TCP stream packets. The TCP stack
 used by Phantun is designed to pass through most L3/L4 stateful/stateless firewalls/NAT
 devices. It will **not** be able to pass through L7 proxies.
 However, the advantage of this approach is that none of the common UDP over TCP performance killer
@@ -51,6 +55,10 @@ connection from the perspective of firewalls/NAT devices.
 Phantun means Phantom TUN, as it is an obfuscator for UDP traffic that does just enough work
 to make it pass through stateful firewall/NATs as TCP packets.
 
+Phantun is written in 100% safe Rust. It has been optimized extensively to scale well on multi-core
+systems and has no issue saturating all available CPU resources on a fast connection.
+See the [Performance](#performance) section for benchmarking results.
+
 ![Traffic flow diagram](images/traffic-flow.png)
 
 # Usage
@@ -71,7 +79,7 @@ NIC address and Phantun's TUN interface address.
 You may customize the name of Tun interface created by Phantun and the assigned addresses. Please
 run the executable with `-h` options to see how to change them.
 
-Another way to help understand this network topology:
+Another way to help understand this network topology (please see the diagram above for an illustration of this topology):
 
 Phantun Client is like a machine with private IP address (`192.168.200.2`) behind a router.
 In order for it to reach the Internet, you will need to SNAT the private IP address before it's traffic
@@ -134,7 +142,7 @@ iptables -t nat -A POSTROUTING -o eth0 -j MASQUERADE
 Server needs to DNAT the TCP listening port to Phantun's TUN interface address.
 
 Note: change `eth0` to whatever actual physical interface name is and `4567` to
-actual TCP port number used by Phanton server
+actual TCP port number used by Phantun server
 
 [Back to TOC](#table-of-contents)
 
@@ -176,6 +184,8 @@ sudo setcap cap_net_admin=+pe phantun_client
 
 **Note:** Run Phantun executable with `-h` option to see full detailed options.
 
+[Back to TOC](#table-of-contents)
+
 ### Server
 
 Note: `4567` is the TCP port Phantun should listen on and must corresponds to the DNAT
@@ -185,6 +195,12 @@ rule specified above. `127.0.0.1:1234` is the UDP Server to connect to for new c
 RUST_LOG=info /usr/local/bin/phantun_server --local 4567 --remote 127.0.0.1:1234
 ```
 
+Or use host name with `--remote`:
+
+```
+RUST_LOG=info /usr/local/bin/phantun_server --local 4567 --remote example.com:1234
+```
+
 [Back to TOC](#table-of-contents)
 
 ### Client
@@ -196,6 +212,12 @@ the Phantun Server to connect.
 RUST_LOG=info /usr/local/bin/phantun_client --local 127.0.0.1:1234 --remote 10.0.0.1:4567
 ```
 
+Or use host name with `--remote`:
+
+```
+RUST_LOG=info /usr/local/bin/phantun_client --local 127.0.0.1:1234 --remote example.com:4567
+```
+
 [Back to TOC](#table-of-contents)
 
 # MTU overhead
@@ -203,18 +225,20 @@ RUST_LOG=info /usr/local/bin/phantun_client --local 127.0.0.1:1234 --remote 10.0
 Phantun aims to keep tunneling overhead to the minimum. The overhead compared to a plain UDP packet
 is the following:
 
-Standard UDP packet: 20 byte IP header + 8 byte UDP header = 28 bytes
+**Standard UDP packet:** `20 byte IP header + 8 byte UDP header = 28 bytes`
 
-Phantun obfuscated UDP packet: 20 byte IP header + 20 byte TCP header = 40 bytes
+**Obfuscated packet:** `20 byte IP header + 20 byte TCP header = 40 bytes`
 
 
 Note that Phantun does not add any additional header other than IP and TCP headers in order to pass through
 stateful packet inspection!
 
-Phantun's additional overhead: 12 bytes. I other words, when using Phantun, the usable payload for
+Phantun's additional overhead: `12 bytes`. I other words, when using Phantun, the usable payload for
 UDP packet is reduced by 12 bytes. This is the minimum overhead possible when doing such kind
 of obfuscation.
 
+![Packet header diagram](images/packet-headers.png)
+
 [Back to TOC](#table-of-contents)
 
 ## MTU calculation for WireGuard
@@ -222,14 +246,20 @@ of obfuscation.
 For people who use Phantun to tunnel [WireGuard®](https://www.wireguard.com) UDP packets, here are some guidelines on figuring
 out the correct MTU to use for your WireGuard interface.
 
+```
 WireGuard MTU = Interface MTU - IP header (20 bytes) - TCP header (20 bytes) - WireGuard overhead (32 bytes)
+```
 
 For example, for a Ethernet interface with 1500 bytes MTU, the WireGuard interface MTU should be set as:
 
+```
 1500 - 20 - 20 - 32 = 1428 bytes
+```
 
 The resulted Phantun TCP data packet will be 1500 bytes which does not exceed the
-interface MTU of 1500.
+interface MTU of 1500. Please note it is strongly recommended to use the same interface
+MTU for both ends of a WireGuard tunnel, or unexected packet loss may occur and these issues are
+generally very hard to troubleshoot.
 
 [Back to TOC](#table-of-contents)
 
@@ -240,15 +270,25 @@ of Server/Client of Phantun on both ends to ensure maximum compatibility.
 
 [Back to TOC](#table-of-contents)
 
+# Documentations
+
+For users who wish to use `fake-tcp` library inside their own project, refer to the documentations for the library at:
+[https://docs.rs/fake-tcp](https://docs.rs/fake-tcp).
+
+[Back to TOC](#table-of-contents)
+
 # Performance
 
-Performance was tested on AWS t3.xlarge instance with 4 vCPUs and 5 Gb/s NIC. WireGuard was used
-for tunneling TCP/UDP traffic between two test instances and MTU has been tuned to avoid fragmentation.
+Performance was tested on 2 AWS `t4g.xlarge` instances with 4 vCPUs and 5 Gb/s NIC over LAN. `nftables` was used to redirect
+UDP stream of `iperf3` to go through the Phantun/udp2raw tunnel between two test instances and MTU has been tuned to avoid fragmentation.
 
-|                 | WireGuard   | WireGuard + Phantun | WireGuard + udp2raw (cipher-mode=none auth-mode=none disable-anti-replay) |
-|-----------------|-------------|---------------------|---------------------------------------------------------------------------|
-| iperf3 -c IP -R | 1.56 Gbit/s | 540 Mbit/s          | 369 Mbit/s                                                                |
-| iperf3 -c IP    | 1.71 Gbit/s | 519 Mbit/s          | 312 Mbit/s                                                                |
+Test command: `iperf3 -c <IP> -p <PORT> -R -u -l 1400 -b 1000m -t 30 -P 5`
+
+| Mode                                                          | Speed          | Overall CPU Usage        |
+|---------------------------------------------------------------|----------------|--------------------------|
+| Direct connection                                             | 3.35 Gbits/sec | 25% (1 core at 100%)     |
+| Phantun                                                       | 2.03 Gbits/sec | 95% (all cores utilized) |
+| udp2raw (cipher-mode=none auth-mode=none disable-anti-replay) | 876 Mbits/sec  | 50% (2 cores at 100%)    |
 
 [Back to TOC](#table-of-contents)
 
@@ -264,7 +304,7 @@ for tunneling TCP/UDP traffic between two test instances and MTU has been tuned
 # Compariation to udp2raw
 [udp2raw](https://github.com/wangyu-/udp2raw-tunnel) is another popular project by [@wangyu-](https://github.com/wangyu-)
 that is very similar to what Phantun can do. In fact I took inspirations of Phantun from udp2raw. The biggest reason for
-developing Phanton is because of lack of performance when running udp2raw (especially on multi-core systems such as Raspberry Pi).
+developing Phantun is because of lack of performance when running udp2raw (especially on multi-core systems such as Raspberry Pi).
 However, the goal is never to be as feature complete as udp2raw and only support the most common use cases. Most notably, UDP over ICMP
 and UDP over UDP mode are not supported and there is no anti-replay nor encryption support. The benefit of this is much better
 performance overall and less MTU overhead because lack of additional headers inside the TCP payload.
@@ -278,7 +318,7 @@ Here is a quick overview of comparison between those two to help you choose:
 | UDP over UDP obfuscation                         |       ❌       |         ✅         |
 | Multi-threaded                                   |       ✅       |         ❌         |
 | Throughput                                       |     Better    |        Good       |
-| Raw IP mode                                      | TUN interface | Raw sockets + BPF |
+| L4 IP mode                                       | TUN interface | Raw sockets + BPF |
 | Tunneling MTU overhead                           |    12 bytes   |      44 bytes     |
 | Seprate TCP connections for each UDP connection  | Client/Server |    Server only    |
 | Anti-replay, encryption                          |       ❌       |         ✅         |
@@ -288,7 +328,7 @@ Here is a quick overview of comparison between those two to help you choose:
 
 # License
 
-Copyright 2021 Datong Sun <dndx@idndx.com>
+Copyright 2021-2022 Datong Sun (dndx@idndx.com)
 
 Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
 [https://www.apache.org/licenses/LICENSE-2.0](https://www.apache.org/licenses/LICENSE-2.0)> or the MIT license

fake-tcp/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fake-tcp"
-version = "0.2.0"
+version = "0.3.1"
 edition = "2021"
 authors = ["Datong Sun <dndx@idndx.com>"]
 license = "MIT OR Apache-2.0"
@@ -16,9 +16,10 @@ benchmark = []
 
 [dependencies]
 bytes = "1"
-pnet = "0.28.0"
-tokio = { version = "1.12.0", features = ["full"] }
-rand = { version = "0.8.4", features = ["small_rng"] }
+pnet = "0.29"
+tokio = { version = "1.14", features = ["full"] }
+rand = { version = "0.8", features = ["small_rng"] }
 log = "0.4"
-internet-checksum = "0.2.0"
-dndx-fork-tokio-tun = "0.4.0"
+internet-checksum = "0.2"
+tokio-tun = "0.5"
+flume = "0.10"

fake-tcp/LICENSE-APACHE

@@ -186,7 +186,7 @@ APPENDIX: How to apply the Apache License to your work.
    same "printed page" as the copyright notice for easier
    identification within third-party archives.
 
-Copyright 2014-2021 The Rust Project Developers
+Copyright 2021-2022 Datong Sun (dndx@idndx.com)
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

fake-tcp/LICENSE-MIT

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2014-2021 The Rust Project Developers
+Copyright (c) 2021-2022 Datong Sun (dndx@idndx.com)
 
 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated

fake-tcp/src/lib.rs

@@ -1,7 +1,46 @@
+//! A minimum, userspace TCP based datagram stack
+//!
+//! # Overview
+//!
+//! `fake-tcp` is a reusable library that implements a minimum TCP stack in
+//! user space using the Tun interface. It allows programs to send datagrams
+//! as if they are part of a TCP connection. `fake-tcp` has been tested to
+//! be able to pass through a variety of NAT and stateful firewalls while
+//! fully preserves certain desirable behavior such as out of order delivery
+//! and no congestion/flow controls.
+//!
+//! # Core Concepts
+//!
+//! The core of the `fake-tcp` crate compose of two structures. [`Stack`] and
+//! [`Socket`].
+//!
+//! ## [`Stack`]
+//!
+//! [`Stack`] represents a virtual TCP stack that operates at
+//! Layer 3. It is responsible for:
+//!
+//! * TCP active and passive open and handshake
+//! * `RST` handling
+//! * Interact with the Tun interface at Layer 3
+//! * Distribute incoming datagrams to corresponding [`Socket`]
+//!
+//! ## [`Socket`]
+//!
+//! [`Socket`] represents a TCP connection. It registers the identifying
+//! tuple `(src_ip, src_port, dest_ip, dest_port)` inside the [`Stack`] so
+//! so that incoming packets can be distributed to the right [`Socket`] with
+//! using a channel. It is also what the client should use for
+//! sending/receiving datagrams.
+//!
+//! # Examples
+//!
+//! Please see [`client.rs`](https://github.com/dndx/phantun/blob/main/phantun/src/bin/client.rs)
+//! and [`server.rs`](https://github.com/dndx/phantun/blob/main/phantun/src/bin/server.rs) files
+//! from the `phantun` crate for how to use this library in client/server mode, respectively.
+
 #![cfg_attr(feature = "benchmark", feature(test))]
 
 pub mod packet;
-extern crate dndx_fork_tokio_tun as tokio_tun;
 
 use bytes::{Bytes, BytesMut};
 use log::{error, info, trace, warn};
@@ -14,17 +53,18 @@ use std::net::{Ipv4Addr, SocketAddrV4};
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
 use tokio::sync::broadcast;
-use tokio::sync::mpsc::{self, Receiver, Sender};
-use tokio::sync::Mutex as AsyncMutex;
+use tokio::sync::mpsc;
 use tokio::time;
 use tokio_tun::Tun;
 
 const TIMEOUT: time::Duration = time::Duration::from_secs(1);
 const RETRIES: usize = 6;
-const MPSC_BUFFER_LEN: usize = 512;
+const MPMC_BUFFER_LEN: usize = 512;
+const MPSC_BUFFER_LEN: usize = 128;
+const MAX_UNACKED_LEN: u32 = 128 * 1024 * 1024; // 128MB
 
 #[derive(Hash, Eq, PartialEq, Clone, Debug)]
-pub struct AddrTuple {
+struct AddrTuple {
     local_addr: SocketAddrV4,
     remote_addr: SocketAddrV4,
 }
@@ -39,17 +79,17 @@ impl AddrTuple {
 }
 
 struct Shared {
-    tuples: RwLock<HashMap<AddrTuple, Sender<Bytes>>>,
+    tuples: RwLock<HashMap<AddrTuple, flume::Sender<Bytes>>>,
     listening: RwLock<HashSet<u16>>,
     tun: Vec<Arc<Tun>>,
-    ready: Sender<Socket>,
+    ready: mpsc::Sender<Socket>,
     tuples_purge: broadcast::Sender<AddrTuple>,
 }
 
 pub struct Stack {
     shared: Arc<Shared>,
     local_ip: Ipv4Addr,
-    ready: Receiver<Socket>,
+    ready: mpsc::Receiver<Socket>,
 }
 
 pub enum State {
@@ -57,20 +97,27 @@ pub enum State {
     SynSent,
     SynReceived,
     Established,
-    Closed,
 }
 
 pub struct Socket {
     shared: Arc<Shared>,
     tun: Arc<Tun>,
-    incoming: AsyncMutex<Receiver<Bytes>>,
+    incoming: flume::Receiver<Bytes>,
     local_addr: SocketAddrV4,
     remote_addr: SocketAddrV4,
     seq: AtomicU32,
     ack: AtomicU32,
+    last_ack: AtomicU32,
     state: State,
 }
 
+/// A socket that represents a unique TCP connection between a server and client.
+///
+/// The `Socket` object itself satisfies `Sync` and `Send`, which means it can
+/// be safely called within an async future.
+///
+/// To close a TCP connection that is no longer needed, simply drop this object
+/// out of scope.
 impl Socket {
     fn new(
         shared: Arc<Shared>,
@@ -79,18 +126,19 @@ impl Socket {
         remote_addr: SocketAddrV4,
         ack: Option<u32>,
         state: State,
-    ) -> (Socket, Sender<Bytes>) {
-        let (incoming_tx, incoming_rx) = mpsc::channel(MPSC_BUFFER_LEN);
+    ) -> (Socket, flume::Sender<Bytes>) {
+        let (incoming_tx, incoming_rx) = flume::bounded(MPMC_BUFFER_LEN);
 
         (
             Socket {
                 shared,
                 tun,
-                incoming: AsyncMutex::new(incoming_rx),
+                incoming: incoming_rx,
                 local_addr,
                 remote_addr,
                 seq: AtomicU32::new(0),
                 ack: AtomicU32::new(ack.unwrap_or(0)),
+                last_ack: AtomicU32::new(ack.unwrap_or(0)),
                 state,
             },
             incoming_tx,
@@ -98,39 +146,48 @@ impl Socket {
     }
 
     fn build_tcp_packet(&self, flags: u16, payload: Option<&[u8]>) -> Bytes {
+        let ack = self.ack.load(Ordering::Relaxed);
+        self.last_ack.store(ack, Ordering::Relaxed);
+
         build_tcp_packet(
             self.local_addr,
             self.remote_addr,
             self.seq.load(Ordering::Relaxed),
-            self.ack.load(Ordering::Relaxed),
+            ack,
             flags,
             payload,
         )
     }
 
+    /// Sends a datagram to the other end.
+    ///
+    /// This method takes `&self`, and it can be called safely by multiple threads
+    /// at the same time.
+    ///
+    /// A return of `None` means the Tun socket returned an error
+    /// and this socket must be closed.
     pub async fn send(&self, payload: &[u8]) -> Option<()> {
         match self.state {
             State::Established => {
                 let buf = self.build_tcp_packet(tcp::TcpFlags::ACK, Some(payload));
                 self.seq.fetch_add(payload.len() as u32, Ordering::Relaxed);
-
-                tokio::select! {
-                    res = self.tun.send(&buf) => {
-                        res.unwrap();
-                        Some(())
-                    },
-                }
+                self.tun.send(&buf).await.ok().and(Some(()))
             }
-            State::Closed => None,
             _ => unreachable!(),
         }
     }
 
+    /// Attempt to receive a datagram from the other end.
+    ///
+    /// This method takes `&self`, and it can be called safely by multiple threads
+    /// at the same time.
+    ///
+    /// A return of `None` means the TCP connection is broken
+    /// and this socket must be closed.
     pub async fn recv(&self, buf: &mut [u8]) -> Option<usize> {
         match self.state {
             State::Established => {
-                let mut incoming = self.incoming.lock().await;
-                incoming.recv().await.and_then(|raw_buf| {
+                self.incoming.recv_async().await.ok().and_then(|raw_buf| {
                     let (_v4_packet, tcp_packet) = parse_ipv4_packet(&raw_buf);
 
                     if (tcp_packet.get_flags() & tcp::TcpFlags::RST) != 0 {
@@ -140,15 +197,24 @@ impl Socket {
 
                     let payload = tcp_packet.payload();
 
-                    self.ack
-                        .store(tcp_packet.get_sequence().wrapping_add(1), Ordering::Relaxed);
+                    let new_ack = tcp_packet.get_sequence().wrapping_add(payload.len() as u32);
+                    let last_ask = self.last_ack.load(Ordering::Relaxed);
+                    self.ack.store(new_ack, Ordering::Relaxed);
+
+                    if new_ack.overflowing_sub(last_ask).0 > MAX_UNACKED_LEN {
+                        let buf = self.build_tcp_packet(tcp::TcpFlags::ACK, None);
+                        if let Err(e) = self.tun.try_send(&buf) {
+                            // This should not really happen as we have not sent anything for
+                            // quite some time...
+                            info!("Connection {} unable to send idling ACK back: {}", self, e)
+                        }
+                    }
 
                     buf[..payload.len()].copy_from_slice(payload);
 
                     Some(payload.len())
                 })
             }
-            State::Closed => None,
             _ => unreachable!(),
         }
     }
@@ -164,7 +230,7 @@ impl Socket {
                     info!("Sent SYN + ACK to client");
                 }
                 State::SynReceived => {
-                    let res = time::timeout(TIMEOUT, self.incoming.lock().await.recv()).await;
+                    let res = time::timeout(TIMEOUT, self.incoming.recv_async()).await;
                     if let Ok(buf) = res {
                         let buf = buf.unwrap();
                         let (_v4_packet, tcp_packet) = parse_ipv4_packet(&buf);
@@ -208,7 +274,7 @@ impl Socket {
                     info!("Sent SYN to server");
                 }
                 State::SynSent => {
-                    match time::timeout(TIMEOUT, self.incoming.lock().await.recv()).await {
+                    match time::timeout(TIMEOUT, self.incoming.recv_async()).await {
                         Ok(buf) => {
                             let buf = buf.unwrap();
                             let (_v4_packet, tcp_packet) = parse_ipv4_packet(&buf);
@@ -251,16 +317,22 @@ impl Socket {
 }
 
 impl Drop for Socket {
+    /// Drop the socket and close the TCP connection
     fn drop(&mut self) {
-        self.state = State::Closed;
-
         let tuple = AddrTuple::new(self.local_addr, self.remote_addr);
         // dissociates ourself from the dispatch map
         assert!(self.shared.tuples.write().unwrap().remove(&tuple).is_some());
         // purge cache
         self.shared.tuples_purge.send(tuple).unwrap();
 
-        let buf = self.build_tcp_packet(tcp::TcpFlags::RST, None);
+        let buf = build_tcp_packet(
+            self.local_addr,
+            self.remote_addr,
+            self.seq.load(Ordering::Relaxed),
+            0,
+            tcp::TcpFlags::RST,
+            None,
+        );
         if let Err(e) = self.tun.try_send(&buf) {
             warn!("Unable to send RST to remote end: {}", e);
         }
@@ -270,6 +342,7 @@ impl Drop for Socket {
 }
 
 impl fmt::Display for Socket {
+    /// User-friendly string representation of the socket
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
@@ -279,7 +352,12 @@ impl fmt::Display for Socket {
     }
 }
 
+/// A userspace TCP state machine
 impl Stack {
+    /// Create a new stack, `tun` is an array of [`Tun`](tokio_tun::Tun).
+    /// When more than one [`Tun`](tokio_tun::Tun) object is passed in, same amount
+    /// of reader will be spawned later. This allows user to utilize the performance
+    /// benefit of Multiqueue Tun support on machines with SMP.
     pub fn new(tun: Vec<Tun>) -> Stack {
         let tun: Vec<Arc<Tun>> = tun.into_iter().map(Arc::new).collect();
         let (ready_tx, ready_rx) = mpsc::channel(MPSC_BUFFER_LEN);
@@ -308,14 +386,18 @@ impl Stack {
         }
     }
 
+    /// Listens for incoming connections on the given `port`.
     pub fn listen(&mut self, port: u16) {
         assert!(self.shared.listening.write().unwrap().insert(port));
     }
 
+    /// Accepts an incoming connection.
     pub async fn accept(&mut self) -> Socket {
         self.ready.recv().await.unwrap()
     }
 
+    /// Connects to the remote end. `None` returned means
+    /// the connection attempt failed.
     pub async fn connect(&mut self, addr: SocketAddrV4) -> Option<Socket> {
         let mut rng = SmallRng::from_entropy();
         let local_port: u16 = rng.gen_range(1024..65535);
@@ -343,7 +425,7 @@ impl Stack {
         shared: Arc<Shared>,
         mut tuples_purge: broadcast::Receiver<AddrTuple>,
     ) {
-        let mut tuples: HashMap<AddrTuple, Sender<Bytes>> = HashMap::new();
+        let mut tuples: HashMap<AddrTuple, flume::Sender<Bytes>> = HashMap::new();
 
         loop {
             let mut buf = BytesMut::with_capacity(MAX_PACKET_LEN);
@@ -367,7 +449,7 @@ impl Stack {
 
                     let tuple = AddrTuple::new(local_addr, remote_addr);
                     if let Some(c) = tuples.get(&tuple) {
-                        if c.send(buf).await.is_err() {
+                        if c.send_async(buf).await.is_err() {
                             trace!("Cache hit, but receiver already closed, dropping packet");
                         }
 
@@ -386,7 +468,7 @@ impl Stack {
                         if let Some(c) = sender {
                             trace!("Storing connection information into local tuples");
                             tuples.insert(tuple, c.clone());
-                            c.send(buf).await.unwrap();
+                            c.send_async(buf).await.unwrap();
                             continue;
                         }
                     }
@@ -421,8 +503,8 @@ impl Stack {
                                 local_addr,
                                 remote_addr,
                                 0,
-                                tcp_packet.get_sequence() + 1,
-                                tcp::TcpFlags::RST,
+                                tcp_packet.get_sequence() + tcp_packet.payload().len() as u32 + 1, // +1 because of SYN flag set
+                                tcp::TcpFlags::RST | tcp::TcpFlags::ACK,
                                 None,
                             );
                             shared.tun[0].try_send(&buf).unwrap();
@@ -433,8 +515,8 @@ impl Stack {
                             local_addr,
                             remote_addr,
                             tcp_packet.get_acknowledgement(),
-                            0,
-                            tcp::TcpFlags::RST,
+                            tcp_packet.get_sequence() + tcp_packet.payload().len() as u32,
+                            tcp::TcpFlags::RST | tcp::TcpFlags::ACK,
                             None,
                         );
                         shared.tun[0].try_send(&buf).unwrap();
@@ -443,7 +525,7 @@ impl Stack {
                 tuple = tuples_purge.recv() => {
                     let tuple = tuple.unwrap();
                     tuples.remove(&tuple);
-                    trace!("Removed cached tuple");
+                    trace!("Removed cached tuple: {:?}", tuple);
                 }
             }
         }

phantun/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "phantun"
-version = "0.2.3"
+version = "0.3.2"
 edition = "2021"
 authors = ["Datong Sun <dndx@idndx.com>"]
 license = "MIT OR Apache-2.0"
@@ -11,11 +11,12 @@ Transforms UDP stream into (fake) TCP streams that can go through
 Layer 3 & Layer 4 (NAPT) firewalls/NATs.
 """
 [dependencies]
-clap = "2.33.3"
-socket2 = { version = "0.4.2", features = ["all"] }
-fake-tcp = "0.2.0"
-tokio = { version = "1.12.0", features = ["full"] }
+clap = { version = "3.0", features = ["cargo"] }
+socket2 = { version = "0.4", features = ["all"] }
+fake-tcp = { path = "../fake-tcp", version = "0.3" }
+tokio = { version = "1.14", features = ["full"] }
+tokio-util = "0.7"
 log = "0.4"
-pretty_env_logger = "0.4.0"
-dndx-fork-tokio-tun = "0.4.0"
-num_cpus = "1.13.0"
+pretty_env_logger = "0.4"
+tokio-tun = "0.5"
+num_cpus = "1.13"

phantun/LICENSE-APACHE

@@ -186,7 +186,7 @@ APPENDIX: How to apply the Apache License to your work.
    same "printed page" as the copyright notice for easier
    identification within third-party archives.
 
-Copyright 2014-2021 The Rust Project Developers
+Copyright 2021-2022 Datong Sun (dndx@idndx.com)
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

phantun/LICENSE-MIT

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2014-2021 The Rust Project Developers
+Copyright (c) 2021-2022 Datong Sun (dndx@idndx.com)
 
 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated

phantun/src/bin/client.rs

@@ -1,51 +1,28 @@
-extern crate dndx_fork_tokio_tun as tokio_tun;
-
-use clap::{crate_version, App, Arg};
+use clap::{crate_version, Arg, Command};
 use fake_tcp::packet::MAX_PACKET_LEN;
 use fake_tcp::{Socket, Stack};
 use log::{debug, error, info};
+use phantun::utils::new_udp_reuseport;
 use std::collections::HashMap;
-use std::convert::TryInto;
 use std::net::{Ipv4Addr, SocketAddr};
 use std::sync::Arc;
-use std::time::Duration;
-use tokio::net::UdpSocket;
-use tokio::sync::RwLock;
+use tokio::sync::{Notify, RwLock};
 use tokio::time;
 use tokio_tun::TunBuilder;
+use tokio_util::sync::CancellationToken;
 
-const UDP_TTL: Duration = Duration::from_secs(180);
-
-fn new_udp_reuseport(addr: SocketAddr) -> UdpSocket {
-    let udp_sock = socket2::Socket::new(
-        if addr.is_ipv4() {
-            socket2::Domain::IPV4
-        } else {
-            socket2::Domain::IPV6
-        },
-        socket2::Type::DGRAM,
-        None,
-    )
-    .unwrap();
-    udp_sock.set_reuse_port(true).unwrap();
-    // from tokio-rs/mio/blob/master/src/sys/unix/net.rs
-    udp_sock.set_cloexec(true).unwrap();
-    udp_sock.set_nonblocking(true).unwrap();
-    udp_sock.bind(&socket2::SockAddr::from(addr)).unwrap();
-    let udp_sock: std::net::UdpSocket = udp_sock.into();
-    udp_sock.try_into().unwrap()
-}
+use phantun::UDP_TTL;
 
 #[tokio::main]
 async fn main() {
     pretty_env_logger::init();
 
-    let matches = App::new("Phantun Client")
+    let matches = Command::new("Phantun Client")
         .version(crate_version!())
         .author("Datong Sun (github.com/dndx)")
         .arg(
-            Arg::with_name("local")
-                .short("l")
+            Arg::new("local")
+                .short('l')
                 .long("local")
                 .required(true)
                 .value_name("IP:PORT")
@@ -53,8 +30,8 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("remote")
-                .short("r")
+            Arg::new("remote")
+                .short('r')
                 .long("remote")
                 .required(true)
                 .value_name("IP or HOST NAME:PORT")
@@ -62,7 +39,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun")
+            Arg::new("tun")
                 .long("tun")
                 .required(false)
                 .value_name("tunX")
@@ -71,7 +48,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun_local")
+            Arg::new("tun_local")
                 .long("tun-local")
                 .required(false)
                 .value_name("IP")
@@ -80,7 +57,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun_peer")
+            Arg::new("tun_peer")
                 .long("tun-peer")
                 .required(false)
                 .value_name("IP")
@@ -101,12 +78,12 @@ async fn main() {
     let remote_addr = tokio::net::lookup_host(matches.value_of("remote").unwrap())
         .await
         .expect("bad remote address or host")
-        .next()
-        .expect("unable to resolve remote host name");
+        .find(|addr| addr.is_ipv4())
+        .expect("unable to resolve remote host name or no valid A record was returned");
     let remote_addr = if let SocketAddr::V4(addr) = remote_addr {
         addr
     } else {
-        panic!("only IPv4 remote address is supported");
+        unreachable!();
     };
     info!("Remote address is: {}", remote_addr);
 
@@ -121,6 +98,9 @@ async fn main() {
         .parse()
         .expect("bad peer address for Tun interface");
 
+    let num_cpus = num_cpus::get();
+    info!("{} cores available", num_cpus);
+
     let tun = TunBuilder::new()
         .name(matches.value_of("tun").unwrap()) // if name is empty, then it is set by kernel.
         .tap(false) // false (default): TUN, true: TAP.
@@ -128,7 +108,7 @@ async fn main() {
         .up() // or set it up manually using `sudo ip link set <tun-name> up`.
         .address(tun_local)
         .destination(tun_peer)
-        .try_build_mq(num_cpus::get())
+        .try_build_mq(num_cpus)
         .unwrap();
 
     info!("Created TUN device {}", tun[0].name());
@@ -170,52 +150,85 @@ async fn main() {
                     assert!(connections.write().await.insert(addr, sock.clone()).is_none());
                     debug!("inserted fake TCP socket into connection table");
 
-                    let connections = connections.clone();
-
                     // spawn "fastpath" UDP socket and task, this will offload main task
                     // from forwarding UDP packets
-                    tokio::spawn(async move {
-                        let mut buf_udp = [0u8; MAX_PACKET_LEN];
-                        let mut buf_tcp = [0u8; MAX_PACKET_LEN];
-                        let udp_sock = new_udp_reuseport(local_addr);
-                        udp_sock.connect(addr).await.unwrap();
 
+                    let packet_received = Arc::new(Notify::new());
+                    let quit = CancellationToken::new();
+
+                    for i in 0..num_cpus {
+                        let sock = sock.clone();
+                        let quit = quit.clone();
+                        let packet_received = packet_received.clone();
+
+                        tokio::spawn(async move {
+                            let mut buf_udp = [0u8; MAX_PACKET_LEN];
+                            let mut buf_tcp = [0u8; MAX_PACKET_LEN];
+                            let udp_sock = new_udp_reuseport(local_addr);
+                            udp_sock.connect(addr).await.unwrap();
+
+                            loop {
+                                tokio::select! {
+                                    Ok(size) = udp_sock.recv(&mut buf_udp) => {
+                                        if sock.send(&buf_udp[..size]).await.is_none() {
+                                            debug!("removed fake TCP socket from connections table");
+                                            quit.cancel();
+                                            return;
+                                        }
+
+                                        packet_received.notify_one();
+                                    },
+                                    res = sock.recv(&mut buf_tcp) => {
+                                        match res {
+                                            Some(size) => {
+                                                if size > 0 {
+                                                    if let Err(e) = udp_sock.send(&buf_tcp[..size]).await {
+                                                        error!("Unable to send UDP packet to {}: {}, closing connection", e, addr);
+                                                        quit.cancel();
+                                                        return;
+                                                    }
+                                                }
+                                            },
+                                            None => {
+                                                debug!("removed fake TCP socket from connections table");
+                                                quit.cancel();
+                                                return;
+                                            },
+                                        }
+
+                                        packet_received.notify_one();
+                                    },
+                                    _ = quit.cancelled() => {
+                                        debug!("worker {} terminated", i);
+                                        return;
+                                    },
+                                };
+                            }
+                        });
+                    }
+
+                    let connections = connections.clone();
+                    tokio::spawn(async move {
                         loop {
                             let read_timeout = time::sleep(UDP_TTL);
+                            let packet_received_fut = packet_received.notified();
 
                             tokio::select! {
-                                Ok(size) = udp_sock.recv(&mut buf_udp) => {
-                                    if sock.send(&buf_udp[..size]).await.is_none() {
-                                        connections.write().await.remove(&addr);
-                                        debug!("removed fake TCP socket from connections table");
-                                        return;
-                                    }
-                                },
-                                res = sock.recv(&mut buf_tcp) => {
-                                    match res {
-                                        Some(size) => {
-                                            if size > 0 {
-                                                if let Err(e) = udp_sock.send(&buf_tcp[..size]).await {
-                                                    connections.write().await.remove(&addr);
-                                                    error!("Unable to send UDP packet to {}: {}, closing connection", e, addr);
-                                                    return;
-                                                }
-                                            }
-                                        },
-                                        None => {
-                                            connections.write().await.remove(&addr);
-                                            debug!("removed fake TCP socket from connections table");
-                                            return;
-                                        },
-                                    }
-                                },
                                 _ = read_timeout => {
                                     info!("No traffic seen in the last {:?}, closing connection", UDP_TTL);
                                     connections.write().await.remove(&addr);
                                     debug!("removed fake TCP socket from connections table");
+
+                                    quit.cancel();
                                     return;
-                                }
-                            };
+                                },
+                                _ = quit.cancelled() => {
+                                    connections.write().await.remove(&addr);
+                                    debug!("removed fake TCP socket from connections table");
+                                    return;
+                                },
+                                _ = packet_received_fut => {},
+                            }
                         }
                     });
                 },

phantun/src/bin/server.rs

@@ -1,25 +1,28 @@
-extern crate dndx_fork_tokio_tun as tokio_tun;
-
-use clap::{crate_version, App, Arg};
+use clap::{crate_version, Arg, Command};
 use fake_tcp::packet::MAX_PACKET_LEN;
 use fake_tcp::Stack;
-use log::{error, info};
+use log::{debug, error, info};
+use phantun::utils::new_udp_reuseport;
 use std::net::Ipv4Addr;
+use std::sync::Arc;
 use tokio::net::UdpSocket;
-use tokio::time::{self, Duration};
+use tokio::sync::Notify;
+use tokio::time;
 use tokio_tun::TunBuilder;
-const UDP_TTL: Duration = Duration::from_secs(180);
+use tokio_util::sync::CancellationToken;
+
+use phantun::UDP_TTL;
 
 #[tokio::main]
 async fn main() {
     pretty_env_logger::init();
 
-    let matches = App::new("Phantun Server")
+    let matches = Command::new("Phantun Server")
         .version(crate_version!())
         .author("Datong Sun (github.com/dndx)")
         .arg(
-            Arg::with_name("local")
-                .short("l")
+            Arg::new("local")
+                .short('l')
                 .long("local")
                 .required(true)
                 .value_name("PORT")
@@ -27,8 +30,8 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("remote")
-                .short("r")
+            Arg::new("remote")
+                .short('r')
                 .long("remote")
                 .required(true)
                 .value_name("IP or HOST NAME:PORT")
@@ -36,7 +39,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun")
+            Arg::new("tun")
                 .long("tun")
                 .required(false)
                 .value_name("tunX")
@@ -45,7 +48,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun_local")
+            Arg::new("tun_local")
                 .long("tun-local")
                 .required(false)
                 .value_name("IP")
@@ -54,7 +57,7 @@ async fn main() {
                 .takes_value(true),
         )
         .arg(
-            Arg::with_name("tun_peer")
+            Arg::new("tun_peer")
                 .long("tun-peer")
                 .required(false)
                 .value_name("IP")
@@ -90,6 +93,9 @@ async fn main() {
         .parse()
         .expect("bad peer address for Tun interface");
 
+    let num_cpus = num_cpus::get();
+    info!("{} cores available", num_cpus);
+
     let tun = TunBuilder::new()
         .name(matches.value_of("tun").unwrap()) // if name is empty, then it is set by kernel.
         .tap(false) // false (default): TUN, true: TAP.
@@ -97,7 +103,7 @@ async fn main() {
         .up() // or set it up manually using `sudo ip link set <tun-name> up`.
         .address(tun_local)
         .destination(tun_peer)
-        .try_build_mq(num_cpus::get())
+        .try_build_mq(num_cpus)
         .unwrap();
 
     info!("Created TUN device {}", tun[0].name());
@@ -112,46 +118,82 @@ async fn main() {
         let mut buf_tcp = [0u8; MAX_PACKET_LEN];
 
         loop {
-            let sock = stack.accept().await;
+            let sock = Arc::new(stack.accept().await);
             info!("New connection: {}", sock);
 
-            tokio::spawn(async move {
-                let udp_sock = UdpSocket::bind(if remote_addr.is_ipv4() {
-                    "0.0.0.0:0"
-                } else {
-                    "[::]:0"
-                })
-                .await
-                .unwrap();
-                udp_sock.connect(remote_addr).await.unwrap();
+            let packet_received = Arc::new(Notify::new());
+            let quit = CancellationToken::new();
+            let udp_sock = UdpSocket::bind(if remote_addr.is_ipv4() {
+                "0.0.0.0:0"
+            } else {
+                "[::]:0"
+            })
+            .await
+            .unwrap();
+            let local_addr = udp_sock.local_addr().unwrap();
+            drop(udp_sock);
 
+            for i in 0..num_cpus {
+                let sock = sock.clone();
+                let quit = quit.clone();
+                let packet_received = packet_received.clone();
+                let udp_sock = new_udp_reuseport(local_addr);
+
+                tokio::spawn(async move {
+                    udp_sock.connect(remote_addr).await.unwrap();
+
+                    loop {
+                        tokio::select! {
+                            Ok(size) = udp_sock.recv(&mut buf_udp) => {
+                                if sock.send(&buf_udp[..size]).await.is_none() {
+                                    quit.cancel();
+                                    return;
+                                }
+
+                                packet_received.notify_one();
+                            },
+                            res = sock.recv(&mut buf_tcp) => {
+                                match res {
+                                    Some(size) => {
+                                        if size > 0 {
+                                            if let Err(e) = udp_sock.send(&buf_tcp[..size]).await {
+                                                error!("Unable to send UDP packet to {}: {}, closing connection", e, remote_addr);
+                                                quit.cancel();
+                                                return;
+                                            }
+                                        }
+                                    },
+                                    None => {
+                                        quit.cancel();
+                                        return;
+                                    },
+                                }
+
+                                packet_received.notify_one();
+                            },
+                            _ = quit.cancelled() => {
+                                debug!("worker {} terminated", i);
+                                return;
+                            },
+                        };
+                    }
+                });
+            }
+
+            tokio::spawn(async move {
                 loop {
                     let read_timeout = time::sleep(UDP_TTL);
+                    let packet_received_fut = packet_received.notified();
 
                     tokio::select! {
-                        Ok(size) = udp_sock.recv(&mut buf_udp) => {
-                            if sock.send(&buf_udp[..size]).await.is_none() {
-                                return;
-                            }
-                        },
-                        res = sock.recv(&mut buf_tcp) => {
-                            match res {
-                                Some(size) => {
-                                    if size > 0 {
-                                        if let Err(e) = udp_sock.send(&buf_tcp[..size]).await {
-                                            error!("Unable to send UDP packet to {}: {}, closing connection", e, remote_addr);
-                                            return;
-                                        }
-                                    }
-                                },
-                                None => { return; },
-                            }
-                        },
                         _ = read_timeout => {
                             info!("No traffic seen in the last {:?}, closing connection", UDP_TTL);
+
+                            quit.cancel();
                             return;
-                        }
-                    };
+                        },
+                        _ = packet_received_fut => {},
+                    }
                 }
             });
         }

phantun/src/lib.rs

@@ -0,0 +1,5 @@
+use std::time::Duration;
+
+pub mod utils;
+
+pub const UDP_TTL: Duration = Duration::from_secs(180);

phantun/src/utils.rs

@@ -0,0 +1,22 @@
+use std::net::SocketAddr;
+use tokio::net::UdpSocket;
+
+pub fn new_udp_reuseport(local_addr: SocketAddr) -> UdpSocket {
+    let udp_sock = socket2::Socket::new(
+        if local_addr.is_ipv4() {
+            socket2::Domain::IPV4
+        } else {
+            socket2::Domain::IPV6
+        },
+        socket2::Type::DGRAM,
+        None,
+    )
+    .unwrap();
+    udp_sock.set_reuse_port(true).unwrap();
+    // from tokio-rs/mio/blob/master/src/sys/unix/net.rs
+    udp_sock.set_cloexec(true).unwrap();
+    udp_sock.set_nonblocking(true).unwrap();
+    udp_sock.bind(&socket2::SockAddr::from(local_addr)).unwrap();
+    let udp_sock: std::net::UdpSocket = udp_sock.into();
+    udp_sock.try_into().unwrap()
+}