feat(amneziawg): add SOCKS5 relay for the embedded amneziawg-go path (Phase 2)

relay.go relays a recovered tunnel connection into Xray's own stock SOCKS5
inbound, authenticating as the peer's email -- the mechanism that gives
embedded AmneziaWG traffic real Xray stats/routing/sniffing with no
Xray-core fork. TCP goes through golang.org/x/net/proxy; UDP needed a
hand-rolled SOCKS5 UDP ASSOCIATE client since neither that package nor
xray-core's own internal socks client expose one.

Verified end-to-end against a real xray-core process (gated behind
XRAY_E2E_BINARY, matching internal/xray's own e2e test convention): a real
TCP and UDP round trip through the whole chain, plus real per-peer stats
counters in Xray's own log.

Xray-config auto-injection (a real SOCKS5 inbound wired into the generated
panel config) is deliberately not part of this commit -- it would require
deciding which AmneziaWG inbounds run on the kernel-module path vs. this
one, and that's an explicit later decision, not something to back into here.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

internal/amneziawgnet/relay.go

@@ -0,0 +1,387 @@
+// Phase 2: relaying a recovered tunnel connection into Xray's own,
+// completely stock SOCKS5 inbound -- authenticating as the owning peer's
+// email -- is what gives every embedded AmneziaWG connection real, native
+// Xray stats/routing/sniffing with no Xray-core fork at all (Finding 3 of
+// the migration plan: a stock SOCKS5 inbound sets its per-connection stats
+// identity directly from the SOCKS5 auth username).
+package amneziawgnet
+
+import (
+	"encoding/binary"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/netip"
+	"sync"
+	"time"
+
+	"golang.org/x/net/proxy"
+	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/logger"
+)
+
+// SocksRelay describes the loopback SOCKS5 inbound decapsulated AmneziaWG
+// traffic gets relayed into.
+type SocksRelay struct {
+	// Addr is the SOCKS5 inbound's own address, e.g. "127.0.0.1:11500".
+	Addr string
+	// Password is shared across every account. This traffic never leaves
+	// loopback, so the password is not a real secrecy boundary -- it only
+	// needs to satisfy Xray's SOCKS5 inbound requiring *some* username/
+	// password auth before it will accept a connection and use the
+	// username as the stats identity. Document this reasoning wherever a
+	// caller generates or displays it, so it's never mistaken later for a
+	// real credential.
+	Password string
+}
+
+// SocksInboundSettings builds the JSON `settings` block for a stock Xray
+// SOCKS5 inbound with one username/password account per email, all sharing
+// password (see SocksRelay's doc comment). udp:true is required: RelayUDP
+// depends on the inbound accepting UDP ASSOCIATE, not just CONNECT.
+func SocksInboundSettings(emails []string, password string) ([]byte, error) {
+	type account struct {
+		User string `json:"user"`
+		Pass string `json:"pass"`
+	}
+	settings := struct {
+		Auth     string    `json:"auth"`
+		UDP      bool      `json:"udp"`
+		Accounts []account `json:"accounts"`
+	}{Auth: "password", UDP: true}
+	for _, email := range emails {
+		settings.Accounts = append(settings.Accounts, account{User: email, Pass: password})
+	}
+	return json.Marshal(settings)
+}
+
+// RelayTCP dials r.Addr, authenticates as email, issues a SOCKS5 CONNECT to
+// dest, and pipes bytes both ways until either side closes or errors.
+// Blocks until the relay ends; meant to be called from (or as) an
+// AttachTCPForwarder handler, which already runs each connection on its own
+// goroutine.
+func (r SocksRelay) RelayTCP(conn *gonet.TCPConn, email string, dest netip.AddrPort) {
+	defer conn.Close()
+
+	auth := &proxy.Auth{User: email, Password: r.Password}
+	dialer, err := proxy.SOCKS5("tcp", r.Addr, auth, proxy.Direct)
+	if err != nil {
+		logger.Warningf("amneziawgnet: RelayTCP: build SOCKS5 dialer: %v", err)
+		return
+	}
+	upstream, err := dialer.Dial("tcp", dest.String())
+	if err != nil {
+		logger.Warningf("amneziawgnet: RelayTCP: SOCKS5 CONNECT to %s as %q: %v", dest, email, err)
+		return
+	}
+	defer upstream.Close()
+
+	done := make(chan struct{}, 2)
+	go func() { io.Copy(upstream, conn); done <- struct{}{} }()
+	go func() { io.Copy(conn, upstream); done <- struct{}{} }()
+	<-done
+}
+
+// socks5UDPSession is one established SOCKS5 UDP ASSOCIATE session: udpConn
+// is the actual socket packets are sent to (and replies read from); ctrl is
+// the TCP control connection that must stay open for the session's
+// lifetime -- per RFC 1928, closing it tears the association down.
+type socks5UDPSession struct {
+	ctrl    net.Conn
+	udpConn *net.UDPConn
+}
+
+// newSocks5UDPSession performs the SOCKS5 greeting, username/password auth,
+// and UDP ASSOCIATE request/reply by hand: golang.org/x/net/proxy's SOCKS5
+// client (used by RelayTCP above) only implements CONNECT, and xray-core's
+// own proxy/socks/client.go is written against its internal transport
+// types, not reusable as a standalone dialer -- so this is a small, direct,
+// from-the-RFC implementation rather than an existing library call.
+func newSocks5UDPSession(addr, user, password string) (*socks5UDPSession, error) {
+	ctrl, err := net.DialTimeout("tcp", addr, 5*time.Second)
+	if err != nil {
+		return nil, fmt.Errorf("amneziawgnet: dial SOCKS5 control connection: %w", err)
+	}
+	if err := socks5Handshake(ctrl, user, password); err != nil {
+		ctrl.Close()
+		return nil, err
+	}
+
+	// UDP ASSOCIATE, dst 0.0.0.0:0 ("I don't know my own source yet, and I
+	// don't need to specify one for a loopback relay").
+	if _, err := ctrl.Write([]byte{0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0}); err != nil {
+		ctrl.Close()
+		return nil, fmt.Errorf("amneziawgnet: send UDP ASSOCIATE request: %w", err)
+	}
+	bind, err := readSocks5Reply(ctrl)
+	if err != nil {
+		ctrl.Close()
+		return nil, err
+	}
+
+	udpConn, err := net.DialUDP("udp", nil, net.UDPAddrFromAddrPort(bind))
+	if err != nil {
+		ctrl.Close()
+		return nil, fmt.Errorf("amneziawgnet: dial SOCKS5 UDP relay endpoint %s: %w", bind, err)
+	}
+	return &socks5UDPSession{ctrl: ctrl, udpConn: udpConn}, nil
+}
+
+// socks5Handshake performs the version greeting and (if the server
+// requires it) username/password auth. Xray's SOCKS5 inbound with
+// auth:"password" always requires it; the no-auth branch exists so this
+// helper isn't silently wrong against a differently-configured server.
+func socks5Handshake(conn net.Conn, user, password string) error {
+	if _, err := conn.Write([]byte{0x05, 0x02, 0x00, 0x02}); err != nil {
+		return fmt.Errorf("amneziawgnet: send SOCKS5 greeting: %w", err)
+	}
+	var resp [2]byte
+	if _, err := io.ReadFull(conn, resp[:]); err != nil {
+		return fmt.Errorf("amneziawgnet: read SOCKS5 greeting reply: %w", err)
+	}
+	if resp[0] != 0x05 {
+		return fmt.Errorf("amneziawgnet: unexpected SOCKS5 version %d", resp[0])
+	}
+	switch resp[1] {
+	case 0x00: // no auth required
+		return nil
+	case 0x02: // username/password
+		req := make([]byte, 0, 3+len(user)+len(password))
+		req = append(req, 0x01, byte(len(user)))
+		req = append(req, user...)
+		req = append(req, byte(len(password)))
+		req = append(req, password...)
+		if _, err := conn.Write(req); err != nil {
+			return fmt.Errorf("amneziawgnet: send SOCKS5 auth: %w", err)
+		}
+		var authResp [2]byte
+		if _, err := io.ReadFull(conn, authResp[:]); err != nil {
+			return fmt.Errorf("amneziawgnet: read SOCKS5 auth reply: %w", err)
+		}
+		if authResp[1] != 0x00 {
+			return fmt.Errorf("amneziawgnet: SOCKS5 auth rejected (status %d)", authResp[1])
+		}
+		return nil
+	default:
+		return fmt.Errorf("amneziawgnet: SOCKS5 server offered unsupported auth method %d", resp[1])
+	}
+}
+
+// readSocks5Reply reads a SOCKS5 reply (the common format shared by CONNECT
+// and UDP ASSOCIATE replies) and returns its bound address.
+func readSocks5Reply(r io.Reader) (netip.AddrPort, error) {
+	var hdr [4]byte
+	if _, err := io.ReadFull(r, hdr[:]); err != nil {
+		return netip.AddrPort{}, fmt.Errorf("amneziawgnet: read SOCKS5 reply header: %w", err)
+	}
+	if hdr[0] != 0x05 {
+		return netip.AddrPort{}, fmt.Errorf("amneziawgnet: unexpected SOCKS5 reply version %d", hdr[0])
+	}
+	if hdr[1] != 0x00 {
+		return netip.AddrPort{}, fmt.Errorf("amneziawgnet: SOCKS5 request failed (reply code %d)", hdr[1])
+	}
+	addr, err := readSocks5Addr(r, hdr[3])
+	if err != nil {
+		return netip.AddrPort{}, err
+	}
+	var portBytes [2]byte
+	if _, err := io.ReadFull(r, portBytes[:]); err != nil {
+		return netip.AddrPort{}, fmt.Errorf("amneziawgnet: read SOCKS5 reply port: %w", err)
+	}
+	return netip.AddrPortFrom(addr, binary.BigEndian.Uint16(portBytes[:])), nil
+}
+
+// readSocks5Addr reads the address portion of a SOCKS5 reply for the given
+// address type (IPv4, IPv6, or domain -- resolved locally since a loopback
+// Xray inbound is not expected to reply with one, but it's cheap to handle
+// correctly rather than fail oddly if it ever does).
+func readSocks5Addr(r io.Reader, atyp byte) (netip.Addr, error) {
+	switch atyp {
+	case 0x01:
+		var b [4]byte
+		if _, err := io.ReadFull(r, b[:]); err != nil {
+			return netip.Addr{}, err
+		}
+		return netip.AddrFrom4(b), nil
+	case 0x04:
+		var b [16]byte
+		if _, err := io.ReadFull(r, b[:]); err != nil {
+			return netip.Addr{}, err
+		}
+		return netip.AddrFrom16(b), nil
+	case 0x03:
+		var l [1]byte
+		if _, err := io.ReadFull(r, l[:]); err != nil {
+			return netip.Addr{}, err
+		}
+		name := make([]byte, l[0])
+		if _, err := io.ReadFull(r, name); err != nil {
+			return netip.Addr{}, err
+		}
+		resolved, err := net.ResolveIPAddr("ip", string(name))
+		if err != nil {
+			return netip.Addr{}, fmt.Errorf("amneziawgnet: resolve SOCKS5 domain reply %q: %w", name, err)
+		}
+		addr, ok := netip.AddrFromSlice(resolved.IP)
+		if !ok {
+			return netip.Addr{}, fmt.Errorf("amneziawgnet: unparseable resolved SOCKS5 domain reply address")
+		}
+		return addr, nil
+	default:
+		return netip.Addr{}, fmt.Errorf("amneziawgnet: unsupported SOCKS5 address type %d", atyp)
+	}
+}
+
+// Close ends the UDP ASSOCIATE session: closing ctrl tells the SOCKS5
+// server to tear down its relay side too (RFC 1928).
+func (s *socks5UDPSession) Close() error {
+	s.udpConn.Close()
+	return s.ctrl.Close()
+}
+
+// sendTo wraps payload in a SOCKS5 UDP request header addressed to dest and
+// sends it to the session's relay endpoint.
+func (s *socks5UDPSession) sendTo(dest netip.AddrPort, payload []byte) error {
+	hdr := make([]byte, 0, 3+1+16+2+len(payload))
+	hdr = append(hdr, 0x00, 0x00, 0x00) // RSV RSV FRAG(=0, no fragmentation)
+	if dest.Addr().Is4() {
+		b := dest.Addr().As4()
+		hdr = append(hdr, 0x01)
+		hdr = append(hdr, b[:]...)
+	} else {
+		b := dest.Addr().As16()
+		hdr = append(hdr, 0x04)
+		hdr = append(hdr, b[:]...)
+	}
+	var portBytes [2]byte
+	binary.BigEndian.PutUint16(portBytes[:], dest.Port())
+	hdr = append(hdr, portBytes[:]...)
+	hdr = append(hdr, payload...)
+	_, err := s.udpConn.Write(hdr)
+	return err
+}
+
+// receive reads one reply datagram into buf, returning the address the
+// SOCKS5 server says it came from and the actual payload (a sub-slice of
+// buf -- valid only until the next receive call).
+func (s *socks5UDPSession) receive(buf []byte) (netip.AddrPort, []byte, error) {
+	n, err := s.udpConn.Read(buf)
+	if err != nil {
+		return netip.AddrPort{}, nil, err
+	}
+	data := buf[:n]
+	if len(data) < 4 {
+		return netip.AddrPort{}, nil, fmt.Errorf("amneziawgnet: short SOCKS5 UDP reply (%d bytes)", n)
+	}
+	atyp := data[3]
+	data = data[4:]
+	addr, err := readSocks5Addr(bytesReader{data}, atyp)
+	if err != nil {
+		return netip.AddrPort{}, nil, err
+	}
+	switch atyp {
+	case 0x01:
+		data = data[4:]
+	case 0x04:
+		data = data[16:]
+	}
+	if len(data) < 2 {
+		return netip.AddrPort{}, nil, fmt.Errorf("amneziawgnet: truncated SOCKS5 UDP reply port")
+	}
+	port := binary.BigEndian.Uint16(data[:2])
+	return netip.AddrPortFrom(addr, port), data[2:], nil
+}
+
+// bytesReader is the minimal io.Reader readSocks5Addr needs, over an
+// in-memory slice that's already fully available (a received UDP
+// datagram) -- avoids pulling in bytes.Reader just for this.
+type bytesReader struct{ b []byte }
+
+func (r bytesReader) Read(p []byte) (int, error) {
+	n := copy(p, r.b)
+	if n < len(p) {
+		return n, io.ErrUnexpectedEOF
+	}
+	return n, nil
+}
+
+// UDPRelay tracks one SOCKS5 UDP ASSOCIATE session per source (tunnel-
+// internal client) flow, relaying each into r's SOCKS5 inbound and writing
+// replies back through gstack -- the UDP counterpart of RelayTCP, meant to
+// be driven by an AttachUDPHandler callback (see udp.go).
+type UDPRelay struct {
+	relay  SocksRelay
+	gstack *stack.Stack
+
+	mu       sync.Mutex
+	sessions map[string]*socks5UDPSession
+}
+
+// NewUDPRelay creates a UDPRelay for one embedded AmneziaWG Device's stack.
+func NewUDPRelay(relay SocksRelay, gstack *stack.Stack) *UDPRelay {
+	return &UDPRelay{relay: relay, gstack: gstack, sessions: map[string]*socks5UDPSession{}}
+}
+
+// Handle relays one packet from src (the peer's tunnel-internal source) to
+// dst (its real, recovered destination), opening a fresh SOCKS5 UDP
+// ASSOCIATE session for src the first time it's seen (authenticating as
+// email, so Xray attributes the whole flow's stats to the right peer) and
+// reusing it for subsequent packets from the same src.
+func (u *UDPRelay) Handle(src, dst netip.AddrPort, email string, payload []byte) {
+	u.mu.Lock()
+	sess, ok := u.sessions[src.String()]
+	u.mu.Unlock()
+
+	if !ok {
+		var err error
+		sess, err = newSocks5UDPSession(u.relay.Addr, email, u.relay.Password)
+		if err != nil {
+			logger.Warningf("amneziawgnet: UDPRelay: SOCKS5 associate for %q: %v", email, err)
+			return
+		}
+		u.mu.Lock()
+		u.sessions[src.String()] = sess
+		u.mu.Unlock()
+		go u.pump(src, sess)
+	}
+	if err := sess.sendTo(dst, payload); err != nil {
+		logger.Warningf("amneziawgnet: UDPRelay: send to %s: %v", dst, err)
+	}
+}
+
+// pump reads replies from sess and writes them back into the tunnel toward
+// src until the session errors out or goes idle for 2 minutes, then tears
+// it down -- both the map entry and the underlying SOCKS5 association.
+func (u *UDPRelay) pump(src netip.AddrPort, sess *socks5UDPSession) {
+	defer func() {
+		u.mu.Lock()
+		delete(u.sessions, src.String())
+		u.mu.Unlock()
+		sess.Close()
+	}()
+	buf := make([]byte, 65536)
+	for {
+		_ = sess.udpConn.SetReadDeadline(time.Now().Add(2 * time.Minute))
+		from, payload, err := sess.receive(buf)
+		if err != nil {
+			return
+		}
+		if err := WriteUDPReply(u.gstack, from, src, payload); err != nil {
+			logger.Warningf("amneziawgnet: UDPRelay: reply write: %v", err)
+		}
+	}
+}
+
+// Close tears down every open session. Call when the owning Device is
+// closed.
+func (u *UDPRelay) Close() {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	for k, s := range u.sessions {
+		s.Close()
+		delete(u.sessions, k)
+	}
+}

internal/amneziawgnet/relay_e2e_test.go

@@ -0,0 +1,424 @@
+package amneziawgnet
+
+import (
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/netip"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	awgconn "github.com/amnezia-vpn/amneziawg-go/v3/conn"
+	"github.com/amnezia-vpn/amneziawg-go/v3/device"
+	"github.com/amnezia-vpn/amneziawg-go/v3/tun/netstack"
+	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/amneziawg"
+	"github.com/mhsanaei/3x-ui/v3/internal/util/wireguard"
+)
+
+// TestSocksRelayAgainstRealXray is Phase 2's real end-to-end proof: a
+// genuine amneziawg-go client completes a real handshake against a Device
+// built by NewDevice, dials a real TCP echo server and sends a real UDP
+// echo datagram, and this package's own AttachTCPForwarder/AttachUDPHandler
+// handlers relay both through RelayTCP/UDPRelay into an *actual xray-core
+// process* (not a mock) running a SOCKS5 inbound built by
+// SocksInboundSettings. Verifies real data round-trips on both protocols,
+// then greps the real process's own debug log for
+// "user>>>{email}>>>traffic>>>{up,down}link" -- the same proof Finding 3 of
+// the migration plan established manually in Phase 0, now permanent,
+// repo-owned test infrastructure. The UDP half in particular is the first
+// real test of this package's hand-rolled SOCKS5 UDP ASSOCIATE client
+// (relay.go) against an independent, authoritative implementation of the
+// protocol rather than a mock this same session wrote.
+//
+// Skipped unless XRAY_E2E_BINARY points at an xray executable built from
+// the same xray-core version as go.mod, matching internal/xray's own
+// TestXrayAPI_E2E convention:
+//
+//	go install github.com/xtls/xray-core/main@<version from go.mod>
+//	XRAY_E2E_BINARY=$GOBIN/main go test ./internal/amneziawgnet -run TestSocksRelayAgainstRealXray -v
+func TestSocksRelayAgainstRealXray(t *testing.T) {
+	bin := os.Getenv("XRAY_E2E_BINARY")
+	if bin == "" {
+		t.Skip("set XRAY_E2E_BINARY to an xray binary to run this test")
+	}
+
+	localIP, ok := firstNonLoopbackIPv4()
+	if !ok {
+		t.Skip("no non-loopback IPv4 address available on this host")
+	}
+
+	const wantEmail = "e2e-peer@example.com"
+	const socksPassword = "loopback-only-not-a-real-secret"
+
+	// --- real TCP + UDP echo servers on a real, non-loopback address ---
+	// (dialing 127.0.0.1 as a tunnel-internal destination hangs -- gVisor
+	// won't route loopback out an arbitrary NIC -- so the client dials
+	// localIP instead; it must still be a *real* address since the actual
+	// relay leg is a genuine OS-level dial from the xray-core process, not
+	// anything inside the tunnel's virtual netstack.)
+	tcpEcho, tcpEchoAddr := startTCPEcho(t, localIP)
+	defer tcpEcho.Close()
+	udpEcho, udpEchoAddr := startUDPEcho(t, localIP)
+	defer udpEcho.Close()
+
+	// --- real embedded AmneziaWG server + client, same shape as Phase 1's tests ---
+	serverPriv, serverPub, err := wireguard.GenerateWireguardKeypair()
+	if err != nil {
+		t.Fatalf("generate server keypair: %v", err)
+	}
+	clientPriv, clientPub, err := wireguard.GenerateWireguardKeypair()
+	if err != nil {
+		t.Fatalf("generate client keypair: %v", err)
+	}
+
+	const listenPort = 58715
+	inst := amneziawg.Instance{
+		Id:            4,
+		InterfaceName: "awgtest4",
+		ListenPort:    listenPort,
+		PrivateKey:    serverPriv,
+		PublicKey:     serverPub,
+		Address:       []string{"10.204.0.1/24"},
+		MTU:           1420,
+		Obfuscation: amneziawg.Obfuscation20{
+			Jc: 4, Jmin: 40, Jmax: 70,
+			S1: 20, S2: 30, S3: 20, S4: 20,
+		},
+		Peers: []amneziawg.Peer{{
+			Email:      wantEmail,
+			PublicKey:  clientPub,
+			AllowedIPs: []string{"10.204.0.2/32"},
+		}},
+	}
+	dev, err := NewDevice(inst, DeviceOptions{})
+	if err != nil {
+		t.Fatalf("NewDevice: %v", err)
+	}
+	defer dev.Close()
+	idx := NewPeerIndex(inst.Peers)
+
+	// --- real xray-core process with a SOCKS5 inbound built by this package ---
+	socksPort := freePort(t)
+	settingsJSON, err := SocksInboundSettings([]string{wantEmail}, socksPassword)
+	if err != nil {
+		t.Fatalf("SocksInboundSettings: %v", err)
+	}
+	var rawSettings any
+	if err := json.Unmarshal(settingsJSON, &rawSettings); err != nil {
+		t.Fatalf("unmarshal generated SOCKS5 settings: %v", err)
+	}
+	xrayCfg := map[string]any{
+		"log": map[string]any{"loglevel": "debug"},
+		"inbounds": []any{
+			map[string]any{
+				"listen":   "127.0.0.1",
+				"port":     socksPort,
+				"protocol": "socks",
+				"settings": rawSettings,
+				"tag":      "awg-e2e-socks",
+			},
+		},
+		"outbounds": []any{
+			map[string]any{"protocol": "freedom", "settings": map[string]any{}, "tag": "direct"},
+		},
+		"policy": map[string]any{
+			"levels": map[string]any{
+				"0": map[string]any{"statsUserUplink": true, "statsUserDownlink": true},
+			},
+		},
+		"stats": map[string]any{},
+	}
+	cfgBytes, err := json.MarshalIndent(xrayCfg, "", "  ")
+	if err != nil {
+		t.Fatalf("marshal xray config: %v", err)
+	}
+	cfgPath := filepath.Join(t.TempDir(), "config.json")
+	if err := os.WriteFile(cfgPath, cfgBytes, 0o644); err != nil {
+		t.Fatalf("write xray config: %v", err)
+	}
+
+	var xrayLog syncBuffer
+	cmd := exec.Command(bin, "-c", cfgPath)
+	cmd.Stdout = &xrayLog
+	cmd.Stderr = &xrayLog
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start xray: %v", err)
+	}
+	defer func() {
+		_ = cmd.Process.Kill()
+		_, _ = cmd.Process.Wait()
+	}()
+	waitForPort(t, socksPort)
+
+	socksAddr := fmt.Sprintf("127.0.0.1:%d", socksPort)
+	relay := SocksRelay{Addr: socksAddr, Password: socksPassword}
+	udpRelay := NewUDPRelay(relay, dev.Stack)
+	defer udpRelay.Close()
+
+	AttachTCPForwarder(dev.Stack, func(conn *gonet.TCPConn, dest netip.AddrPort) {
+		srcAddrPort, err := netip.ParseAddrPort(conn.RemoteAddr().String())
+		if err != nil {
+			conn.Close()
+			return
+		}
+		peer, ok := idx.Lookup(srcAddrPort.Addr().Unmap())
+		if !ok {
+			conn.Close()
+			return
+		}
+		relay.RelayTCP(conn, peer.Email, dest)
+	})
+	AttachUDPHandler(dev.Stack, func(src, dst netip.AddrPort, payload []byte) {
+		peer, ok := idx.Lookup(src.Addr())
+		if !ok {
+			return
+		}
+		udpRelay.Handle(src, dst, peer.Email, payload)
+	})
+
+	// --- real client, real handshake, real traffic through the whole chain ---
+	clientTun, clientNet, err := netstack.CreateNetTUN(
+		[]netip.Addr{netip.MustParseAddr("10.204.0.2")},
+		[]netip.Addr{netip.MustParseAddr("1.1.1.1")}, 1420)
+	if err != nil {
+		t.Fatalf("client CreateNetTUN: %v", err)
+	}
+	clientDev := device.NewDevice(clientTun, awgconn.NewDefaultBind(), device.NewLogger(device.LogLevelSilent, ""))
+	defer clientDev.Close()
+
+	clientPrivHex, err := wireguard.KeyToHex(clientPriv)
+	if err != nil {
+		t.Fatalf("client key to hex: %v", err)
+	}
+	serverPubHex, err := wireguard.KeyToHex(serverPub)
+	if err != nil {
+		t.Fatalf("server key to hex: %v", err)
+	}
+	clientConf := fmt.Sprintf(
+		"private_key=%s\njc=4\njmin=40\njmax=70\ns1=20\ns2=30\ns3=20\ns4=20\npublic_key=%s\nendpoint=127.0.0.1:%d\nallowed_ip=0.0.0.0/0\n",
+		clientPrivHex, serverPubHex, listenPort)
+	if err := clientDev.IpcSet(clientConf); err != nil {
+		t.Fatalf("client IpcSet: %v", err)
+	}
+	if err := clientDev.Up(); err != nil {
+		t.Fatalf("client Up: %v", err)
+	}
+
+	// TCP round trip.
+	const tcpMsg = "hello over amneziawgnet+socks5+xray"
+	dialDeadline := time.Now().Add(10 * time.Second)
+	var tcpConn interface {
+		Write([]byte) (int, error)
+		Read([]byte) (int, error)
+		Close() error
+	}
+	for {
+		c, dialErr := clientNet.DialContext(t.Context(), "tcp", tcpEchoAddr.String())
+		if dialErr == nil {
+			tcpConn = c
+			break
+		}
+		if time.Now().After(dialDeadline) {
+			t.Fatalf("client TCP dial via tunnel never succeeded: %v", dialErr)
+		}
+		time.Sleep(150 * time.Millisecond)
+	}
+	defer tcpConn.Close()
+	if _, err := tcpConn.Write([]byte(tcpMsg)); err != nil {
+		t.Fatalf("client TCP write: %v", err)
+	}
+	tcpBuf := make([]byte, len(tcpMsg))
+	if _, err := readFull(tcpConn, tcpBuf, 10*time.Second); err != nil {
+		t.Fatalf("client TCP read: %v", err)
+	}
+	if string(tcpBuf) != tcpMsg {
+		t.Errorf("TCP echo = %q, want %q", tcpBuf, tcpMsg)
+	}
+
+	// UDP round trip.
+	const udpMsg = "hello-udp-over-socks5"
+	uconn, err := clientNet.DialUDPAddrPort(netip.AddrPort{}, udpEchoAddr)
+	if err != nil {
+		t.Fatalf("client DialUDPAddrPort: %v", err)
+	}
+	defer uconn.Close()
+	udpDeadline := time.Now().Add(10 * time.Second)
+	var udpBuf [256]byte
+	var gotUDP string
+	for time.Now().Before(udpDeadline) {
+		_ = uconn.SetWriteDeadline(time.Now().Add(300 * time.Millisecond))
+		if _, err := uconn.Write([]byte(udpMsg)); err != nil {
+			continue
+		}
+		_ = uconn.SetReadDeadline(time.Now().Add(300 * time.Millisecond))
+		n, err := uconn.Read(udpBuf[:])
+		if err == nil {
+			gotUDP = string(udpBuf[:n])
+			break
+		}
+	}
+	if gotUDP != udpMsg {
+		t.Fatalf("UDP echo = %q, want %q (xray log follows)\n%s", gotUDP, udpMsg, xrayLog.String())
+	}
+
+	// Real per-peer stats attribution: stop xray so its log is complete, then
+	// look for both directions' counters keyed by the peer's real email --
+	// the exact proof Finding 3 established manually in Phase 0.
+	_ = cmd.Process.Kill()
+	_, _ = cmd.Process.Wait()
+	log := xrayLog.String()
+	wantUp := fmt.Sprintf("user>>>%s>>>traffic>>>uplink", wantEmail)
+	wantDown := fmt.Sprintf("user>>>%s>>>traffic>>>downlink", wantEmail)
+	if !strings.Contains(log, wantUp) {
+		t.Errorf("xray log missing uplink stats counter %q\nfull log:\n%s", wantUp, log)
+	}
+	if !strings.Contains(log, wantDown) {
+		t.Errorf("xray log missing downlink stats counter %q\nfull log:\n%s", wantDown, log)
+	}
+}
+
+// firstNonLoopbackIPv4 finds a real, locally-bound IPv4 address suitable as
+// a relay-reachable test destination.
+func firstNonLoopbackIPv4() (netip.Addr, bool) {
+	addrs, err := net.InterfaceAddrs()
+	if err != nil {
+		return netip.Addr{}, false
+	}
+	for _, a := range addrs {
+		ipNet, ok := a.(*net.IPNet)
+		if !ok || ipNet.IP.IsLoopback() {
+			continue
+		}
+		if v4 := ipNet.IP.To4(); v4 != nil {
+			addr, ok := netip.AddrFromSlice(v4)
+			if ok {
+				return addr, true
+			}
+		}
+	}
+	return netip.Addr{}, false
+}
+
+func startTCPEcho(t *testing.T, addr netip.Addr) (io interface{ Close() error }, ap netip.AddrPort) {
+	t.Helper()
+	ln, err := net.Listen("tcp", net.JoinHostPort(addr.String(), "0"))
+	if err != nil {
+		t.Fatalf("start TCP echo listener: %v", err)
+	}
+	go func() {
+		for {
+			c, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func() {
+				defer c.Close()
+				buf := make([]byte, 4096)
+				for {
+					n, err := c.Read(buf)
+					if n > 0 {
+						if _, werr := c.Write(buf[:n]); werr != nil {
+							return
+						}
+					}
+					if err != nil {
+						return
+					}
+				}
+			}()
+		}
+	}()
+	port := ln.Addr().(*net.TCPAddr).Port
+	return ln, netip.AddrPortFrom(addr, uint16(port))
+}
+
+func startUDPEcho(t *testing.T, addr netip.Addr) (io interface{ Close() error }, ap netip.AddrPort) {
+	t.Helper()
+	pc, err := net.ListenPacket("udp", net.JoinHostPort(addr.String(), "0"))
+	if err != nil {
+		t.Fatalf("start UDP echo listener: %v", err)
+	}
+	go func() {
+		buf := make([]byte, 4096)
+		for {
+			n, raddr, err := pc.ReadFrom(buf)
+			if err != nil {
+				return
+			}
+			if _, err := pc.WriteTo(buf[:n], raddr); err != nil {
+				return
+			}
+		}
+	}()
+	port := pc.LocalAddr().(*net.UDPAddr).Port
+	return pc, netip.AddrPortFrom(addr, uint16(port))
+}
+
+// readFull reads exactly len(buf) bytes or fails after timeout, since
+// gonet.TCPConn (and net.Conn generally) may return short reads.
+func readFull(r interface{ Read([]byte) (int, error) }, buf []byte, timeout time.Duration) (int, error) {
+	deadline := time.Now().Add(timeout)
+	total := 0
+	for total < len(buf) {
+		if time.Now().After(deadline) {
+			return total, fmt.Errorf("timed out after reading %d/%d bytes", total, len(buf))
+		}
+		n, err := r.Read(buf[total:])
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+	return total, nil
+}
+
+// syncBuffer is a concurrency-safe bytes buffer for capturing a subprocess's
+// combined stdout/stderr while the test may read it from another goroutine.
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf strings.Builder
+}
+
+func (s *syncBuffer) Write(p []byte) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.buf.Write(p)
+}
+
+func (s *syncBuffer) String() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.buf.String()
+}
+
+func freePort(t *testing.T) int {
+	t.Helper()
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+	return l.Addr().(*net.TCPAddr).Port
+}
+
+func waitForPort(t *testing.T, port int) {
+	t.Helper()
+	deadline := time.Now().Add(15 * time.Second)
+	addr := fmt.Sprintf("127.0.0.1:%d", port)
+	for time.Now().Before(deadline) {
+		conn, err := net.DialTimeout("tcp", addr, time.Second)
+		if err == nil {
+			conn.Close()
+			return
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	t.Fatalf("xray port %d did not open in time", port)
+}