Add the C++ TCP relay engine and run ss-tunnel on it

One engine (src/relay/tcp) for the four TCP relays: connections live in a
fixed pool owned through PoolPtr, handshake, cipher state and buffers are
inline, timeouts are intrusive deadline lists on one timer and DNS answers
come back through generation-checked handles. A full pool refuses new
connections and pauses accepting until a slot frees.

ss-tunnel is the first user; its hand-written relay callbacks are gone.
tests/tunnel_interop.py covers every method against ss-server and
shadowsocks-rust.

tests/CMakeLists.txt

@@ -354,6 +354,15 @@ if(SS_BUILD_EXECUTABLES AND NOT CMAKE_CROSSCOMPILING)
                     --self --printable-salt --bin ${CMAKE_RUNTIME_OUTPUT_DIRECTORY})
         set_tests_properties(test_printable_salt_interop PROPERTIES
             LABELS "integration" TIMEOUT 90)
+        # ss-tunnel through the C relay engine against ss-server (and
+        # shadowsocks-rust when installed), every method.
+        set(SS_TUNNEL_INTEROP_ARGS --bin ${CMAKE_RUNTIME_OUTPUT_DIRECTORY})
+        if(SS_ENABLE_LEGACY)
+            list(APPEND SS_TUNNEL_INTEROP_ARGS --legacy)
+        endif()
+        add_test(NAME test_tunnel_interop
+            COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/tunnel_interop.py ${SS_TUNNEL_INTEROP_ARGS})
+        set_tests_properties(test_tunnel_interop PROPERTIES LABELS "integration" TIMEOUT 300)
     endif()
 endif()
 

tests/test_crypto_aead2022_cpp.cpp

@@ -92,12 +92,14 @@ const std::uint8_t kAddr[] = { 0x03, 11, 'e', 'x', 'a', 'm', 'p', 'l', 'e', '.',
 std::vector<std::uint8_t> cpp_decrypt_all(sc::Aead2022TcpDecryptor &dec, const std::vector<std::uint8_t> &wire)
 {
     std::vector<std::uint8_t> got;
-    std::vector<std::uint8_t> out(sc::k2022MaxChunk);
+    std::vector<std::uint8_t> out(70000);
     for (std::size_t off = 0; off < wire.size();) {
         std::size_t n = std::min<std::size_t>(1 + randombytes_uniform(9000), wire.size() - off);
         std::span<const std::uint8_t> in(wire.data() + off, n);
         for (;;) {
-            auto step = dec.decrypt(in, out);
+            // Output windows of any size, down to one byte.
+            std::size_t room = randombytes_uniform(4) == 0 ? 1 + randombytes_uniform(16) : 1 + randombytes_uniform(70000);
+            auto step = dec.decrypt(in, std::span<std::uint8_t>(out).first(room));
             SS_CHECK(step);
             got.insert(got.end(), out.begin(), out.begin() + step->produced);
             in = in.subspan(step->consumed);

tests/tunnel_interop.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""ss-tunnel against ss-server (and shadowsocks-rust when present): concurrent TCP
+transfers and a UDP round trip through a fixed tunnel target, for every method."""
+import argparse
+import base64
+import concurrent.futures
+import os
+from pathlib import Path
+import shutil
+import socket
+import subprocess
+import threading
+
+from interop import TCPHandler, TCPOrigin, UDPHandler, UDPOrigin, free_port, peers, receive
+
+METHODS = ["2022-blake3-aes-128-gcm", "2022-blake3-aes-256-gcm", "2022-blake3-chacha20-poly1305",
+           "aes-128-gcm", "aes-256-gcm", "chacha20-ietf-poly1305"]
+LEGACY = ["aes-256-cfb", "aes-128-ctr", "camellia-256-cfb", "chacha20-ietf", "salsa20"]
+
+
+def tcp_case(port, size):
+    payload = os.urandom(size)
+    with socket.create_connection(("127.0.0.1", port), timeout=15) as sock, \
+            concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+        upload = pool.submit(sock.sendall, payload)
+        assert receive(sock, size) == payload, "TCP payload mismatch"
+        upload.result(timeout=15)
+
+
+def udp_case(port):
+    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+        sock.settimeout(.5)
+        for size in (1, 1200):
+            payload = os.urandom(size)
+            # The UDP-only tunnel has no TCP port to wait on: retry until up.
+            for attempt in range(20):
+                sock.sendto(payload, ("127.0.0.1", port))
+                try:
+                    reply, _ = sock.recvfrom(65535)
+                    break
+                except TimeoutError:
+                    if attempt == 19:
+                        raise
+            assert reply == payload, "UDP payload mismatch"
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--bin", required=True, type=Path)
+    parser.add_argument("--legacy", action="store_true", help="also run the stream ciphers")
+    parser.add_argument("--fast-open", action="store_true")
+    args = parser.parse_args()
+    bins = args.bin.resolve()
+    server, tunnel = str(bins / "ss-server"), str(bins / "ss-tunnel")
+    rust_server = shutil.which("ssserver")
+    methods = METHODS + (LEGACY if args.legacy else [])
+    failures = 0
+    with TCPOrigin(("127.0.0.1", 0), TCPHandler) as tcp, UDPOrigin(("127.0.0.1", 0), UDPHandler) as udp:
+        for origin in (tcp, udp):
+            threading.Thread(target=origin.serve_forever, daemon=True).start()
+        try:
+            for method in methods:
+                stream = method in LEGACY
+                for peer in ("C server",) + (("Rust server",) if rust_server and not stream else ()):
+                    key = base64.b64encode(os.urandom(16 if "aes-128" in method else 32)).decode()
+                    secret = ["-k", "password"] if stream else ["-k", key]
+                    sport, tport, uport = free_port(), free_port(), free_port()
+                    if peer == "C server":
+                        srv = [server, "-s", "127.0.0.1", "-p", str(sport), "-m", method, "-u"] + secret
+                    else:
+                        srv = [rust_server, "-s", f"127.0.0.1:{sport}", "-m", method, "-U"] + secret
+                    extra = ["--fast-open"] if args.fast_open else []
+                    tcp_tunnel = [tunnel, "-s", "127.0.0.1", "-p", str(sport), "-l", str(tport), "-m", method,
+                                  "-L", f"127.0.0.1:{tcp.server_address[1]}"] + secret + extra
+                    udp_tunnel = [tunnel, "-s", "127.0.0.1", "-p", str(sport), "-l", str(uport), "-m", method,
+                                  "-U", "-L", f"127.0.0.1:{udp.server_address[1]}"] + secret
+                    label = f"{method}: tunnel -> {peer}"
+                    try:
+                        with peers([srv, tcp_tunnel], [sport, tport]):
+                            with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+                                cases = [pool.submit(tcp_case, tport, size) for size in (1, 4096, 65536, 1048576)]
+                                for case in cases:
+                                    case.result(timeout=60)
+                            udp_proc = subprocess.Popen(udp_tunnel, stdout=subprocess.DEVNULL,
+                                                        stderr=subprocess.DEVNULL)
+                            try:
+                                udp_case(uport)
+                            finally:
+                                udp_proc.terminate()
+                                udp_proc.wait(timeout=5)
+                        print("PASS " + label, flush=True)
+                    except (OSError, AssertionError, TimeoutError) as error:
+                        failures += 1
+                        print(f"FAIL {label}: {error!r}", flush=True)
+        finally:
+            tcp.shutdown()
+            udp.shutdown()
+    return int(failures != 0)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())