refactor: zero-copy tunnel pipeline overhaul

.github/workflows/build.yml

@@ -21,16 +21,22 @@ jobs:
 
   test:
     needs: lint-audit
-    name: cargo test --lib
+    name: cargo test (unit + integration)
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v6
       - name: Rust Cache
         uses: swatinem/rust-cache@v2
         with:
           key: test-ubuntu
+      - name: Check formatting
+        run: cargo fmt --check
+      - name: Lint with clippy
+        run: cargo clippy --all-targets -- -D warnings
       - name: Run unit tests
         run: cargo test --lib
+      - name: Run integration tests
+        run: cargo test --test bypass --test dns --test encrypted_tunnel --test plain_tunnel --test rotation
 
   linux-gnu:
     needs: [lint-audit, test]

Cargo.toml

@@ -22,7 +22,7 @@ strip = true
 aes-gcm = "0.10.3"
 anyhow = "1.0"
 axum = {version = "0.8.9", features = ["http2", "macros"]}
-base122-fast = "0.1.3"
+base122-fast = "0.1.4"
 base64 = "0.22.1"
 bytes = "1.11"
 clap = {version = "4.6.1", features = ["derive"]}
@@ -50,7 +50,7 @@ serde = {version = "1.0", features = ["derive"]}
 serde_json = "1.0.150"
 sha2 = "0.11.0"
 singleflight-async = "0.2"
-tokio = {version = "1.52.3", features = ["rt-multi-thread"]}
+tokio = {version = "1.52.3", features = ["rt-multi-thread", "test-util"]}
 tokio-rustls = "0.26"
 tokio-socks = "0.5.3"
 tokio-stream = {version = "0.1", features = ["net"]}
@@ -71,3 +71,12 @@ wreq = "6.0.0-rc.29"
 wreq-util = "3.0.0-rc.12"
 x25519-dalek = {version = "2.0", features = ["static_secrets", "getrandom"]}
 zeroize = "1.8.2"
+
+[dev-dependencies]
+axum = {version = "0.8.9", features = ["http2", "macros"]}
+jsonwebtoken = {version = "10.4", features = ["aws_lc_rs"]}
+serde_json = "1.0.150"
+tokio = {version = "1.52.3", features = ["full"]}
+tracing-subscriber = {version = "0.3", features = ["env-filter"]}
+wreq = "6.0.0-rc.29"
+wreq-util = "3.0.0-rc.12"

src/bin/client.rs

@@ -5,7 +5,8 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 use tokio::net::TcpListener;
-use tracing::{Instrument, error_span, info, warn};
+use tokio::sync::Semaphore;
+use tracing::{Instrument, debug, error_span, info, warn};
 
 static NEXT_SPAN_ID: AtomicU64 = AtomicU64::new(1);
 
@@ -54,6 +55,8 @@ async fn main() -> anyhow::Result<()> {
 
     info!(listen = %addr, "proxy listening");
 
+    let conn_sem = Arc::new(Semaphore::new(state.max_connections));
+
     loop {
         let (socket, peer) = match listener.accept().await {
             Ok(conn) => conn,
@@ -63,13 +66,26 @@ async fn main() -> anyhow::Result<()> {
             }
         };
 
+        let permit = match conn_sem.clone().try_acquire_owned() {
+            Ok(p) => p,
+            Err(_) => {
+                debug!(
+                    client = %peer,
+                    max_connections = state.max_connections,
+                    "connection rejected: admission limit reached"
+                );
+                continue;
+            }
+        };
+
         let http_client = Arc::clone(&http_client);
         let state = Arc::clone(&state);
 
         let span_id = NEXT_SPAN_ID.fetch_add(1, Ordering::Relaxed);
 
         tokio::spawn(
             async move {
+                let _permit = permit;
                 if let Err(e) = httproxy::client::connection::handle_connection_actor(
                     socket,
                     http_client,

src/bypass/mod.rs

@@ -290,26 +290,11 @@ mod tests {
         let r = make_rules(&["example.com"], &[]);
         assert!(r.match_domain("example.com"));
         assert!(r.match_domain("sub.example.com"));
+        assert!(r.match_domain("deep.sub.example.com"));
         assert!(!r.match_domain("notexample.com"));
         assert!(!r.match_domain("com"));
     }
 
-    #[test]
-    fn domain_leading_dot() {
-        let r = make_rules(&[".example.com"], &[]);
-        assert!(r.match_domain("example.com"));
-        assert!(r.match_domain("a.b.example.com"));
-        assert!(!r.match_domain("fakeexample.com"));
-    }
-
-    #[test]
-    fn domain_nested() {
-        let r = make_rules(&["google.com"], &[]);
-        assert!(r.match_domain("mail.google.com"));
-        assert!(r.match_domain("deep.nested.google.com"));
-        assert!(!r.match_domain("notgoogle.com"));
-    }
-
     #[test]
     fn domain_tld_wildcard() {
         let r = make_rules(&["com"], &[]);