diff --git a/internal/amneziawgnet/relay.go b/internal/amneziawgnet/relay.go new file mode 100644 index 000000000..f17b42995 --- /dev/null +++ b/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) + } +} diff --git a/internal/amneziawgnet/relay_e2e_test.go b/internal/amneziawgnet/relay_e2e_test.go new file mode 100644 index 000000000..b312438a2 --- /dev/null +++ b/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@ +// 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) +}