diff --git a/go.mod b/go.mod index 041ebe2e8..680c2da2d 100644 --- a/go.mod +++ b/go.mod @@ -35,6 +35,8 @@ require ( pgregory.net/rapid v1.3.0 ) +require github.com/amnezia-vpn/amneziawg-go/v3 v3.0.3 + require ( github.com/Azure/go-ntlmssp v0.1.1 // indirect github.com/andybalholm/brotli v1.2.2 // indirect @@ -110,6 +112,6 @@ require ( golang.zx2c4.com/wireguard/windows v1.0.1 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260729162451-8efbd57d26e0 // indirect google.golang.org/protobuf v1.36.11 - gvisor.dev/gvisor v0.0.0-20260122175437-89a5d21be8f0 // indirect + gvisor.dev/gvisor v0.0.0-20260122175437-89a5d21be8f0 lukechampine.com/blake3 v1.4.1 // indirect ) diff --git a/go.sum b/go.sum index 66d000f5e..10c490ca7 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e h1:4dAU9FXIyQktpoUAgOJK3OTFc/xug0PCXYCqU0FgDKI= github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e/go.mod h1:cEWa1LVoE5KvSD9ONXsZrj0z6KqySlCCNKHlLzbqAt4= +github.com/amnezia-vpn/amneziawg-go/v3 v3.0.3 h1:XYR85mN53hj2DTzToHs3OxIHrNA59QMg1m3+oiOnBi4= +github.com/amnezia-vpn/amneziawg-go/v3 v3.0.3/go.mod h1:YoPc6qcOZqD7TXZ1xpedD8Sx3aSKsxN05ZqEFmXDNHk= github.com/andybalholm/brotli v1.2.2 h1:HzTuoo2ErYQqf5qvcJInB8uvqSVxRttzkFexPWtnceM= github.com/andybalholm/brotli v1.2.2/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/apernet/quic-go v0.59.1-0.20260425001925-6c6cc9bcb716 h1:J1O+xpLuJWkdYbw5JPGwBqIHs2J8tiEP7Py9lPqkN2I= diff --git a/internal/amneziawgnet/device.go b/internal/amneziawgnet/device.go new file mode 100644 index 000000000..a5a7df695 --- /dev/null +++ b/internal/amneziawgnet/device.go @@ -0,0 +1,185 @@ +package amneziawgnet + +import ( + "fmt" + "net/netip" + "strings" + + awgconn "github.com/amnezia-vpn/amneziawg-go/v3/conn" + "github.com/amnezia-vpn/amneziawg-go/v3/device" + "gvisor.dev/gvisor/pkg/tcpip/stack" + + "github.com/mhsanaei/3x-ui/v3/internal/amneziawg" + "github.com/mhsanaei/3x-ui/v3/internal/util/wireguard" +) + +// defaultMTU matches internal/amneziawg's own kernel-module interface +// default -- 1420, WireGuard/AmneziaWG's usual accounting for tunnel +// encapsulation overhead on a standard 1500-byte-MTU host link. +const defaultMTU = 1420 + +// DeviceOptions carries the AmneziaWG 3.0 header-protection fields, kept out +// of amneziawg.Instance/Obfuscation20 deliberately: those are the shared, +// DB-backed types the still-live kernel-module path also reads and writes, +// and 3.0 header protection is a device-wide, strictly opt-in setting (see +// the migration plan's "Reference material" section) that isn't wired into +// that shared schema yet. Zero-value DeviceOptions means classic +// (non-3.0) obfuscation only, matching the kernel-module path's own +// defaults today. +type DeviceOptions struct { + // HeaderProtectionKey is a base64 32-byte key. Empty disables AWG 3.0 + // header protection entirely. Non-empty requires every one of + // Obfuscation20.S1-S4 to be >= 12 (amneziawg-go's own HeaderCipherNonceSize + // requirement) -- IpcSet will reject the config otherwise. + HeaderProtectionKey string + // ContentPaddingAddition is a "low-high" range (or a bare integer) per + // amneziawg-tools' own u16_range_from_string grammar. Empty disables it. + ContentPaddingAddition string + // Logger is passed to device.NewDevice as-is; nil uses a silent logger + // (device.NewLogger(device.LogLevelSilent, "")). + Logger *device.Logger +} + +// Device is one running embedded AmneziaWG interface: an amneziawg-go +// Device over a gVisor netstack, plus the raw *stack.Stack a caller needs to +// attach a TCP/UDP forwarder (see forwarder.go / udp.go). Closing it tears +// down both the WireGuard device and the underlying tun/stack. +type Device struct { + *device.Device + Stack *stack.Stack +} + +// NewDevice constructs and brings up an embedded AmneziaWG interface for +// inst: a gVisor-backed tun.Device sized to inst.MTU (or defaultMTU), +// addressed with inst.Address, configured via UAPI with inst.Obfuscation, +// inst.PrivateKey, opts' AWG 3.0 fields, and one UAPI peer per inst.Peers +// entry. It does not attach a forwarder or start relaying traffic -- +// that's the caller's job (see AttachTCPForwarder / AttachUDPHandler), +// keeping this constructor usable both for a real relay and for a plain +// mechanical test. +func NewDevice(inst amneziawg.Instance, opts DeviceOptions) (*Device, error) { + addrs, err := hostAddresses(inst.Address) + if err != nil { + return nil, fmt.Errorf("amneziawgnet: %w", err) + } + + mtu := inst.MTU + if mtu <= 0 { + mtu = defaultMTU + } + + tun, gstack, err := createNetTUNWithStack(addrs, mtu) + if err != nil { + return nil, fmt.Errorf("amneziawgnet: create netstack: %w", err) + } + + logger := opts.Logger + if logger == nil { + logger = device.NewLogger(device.LogLevelSilent, "") + } + dev := device.NewDevice(tun, awgconn.NewDefaultBind(), logger) + + conf, err := buildUAPIConfig(inst, opts) + if err != nil { + dev.Close() + return nil, fmt.Errorf("amneziawgnet: %w", err) + } + if err := dev.IpcSet(conf); err != nil { + dev.Close() + return nil, fmt.Errorf("amneziawgnet: IpcSet for inbound %d: %w", inst.Id, err) + } + if err := dev.Up(); err != nil { + dev.Close() + return nil, fmt.Errorf("amneziawgnet: bring up inbound %d: %w", inst.Id, err) + } + + return &Device{Device: dev, Stack: gstack}, nil +} + +// hostAddresses parses each of inst.Address's CIDR strings (e.g. +// "10.8.1.1/24") down to the bare host address the netstack's NIC gets +// configured with -- the interface's own address, not the subnet it routes. +func hostAddresses(addresses []string) ([]netip.Addr, error) { + out := make([]netip.Addr, 0, len(addresses)) + for _, a := range addresses { + prefix, err := netip.ParsePrefix(a) + if err != nil { + return nil, fmt.Errorf("invalid interface address %q: %w", a, err) + } + out = append(out, prefix.Addr()) + } + return out, nil +} + +// buildUAPIConfig renders inst (plus opts' AWG 3.0 fields) as a WireGuard +// UAPI "set" configuration string -- private_key/listen_port/jc.../s1-s4/ +// h1-h4/i1 device lines, the AWG 3.0 device lines when opts asks for them, +// then one public_key/preshared_key/allowed_ip block per peer. Field names +// and format match amneziawg-go v3.0.3's device/uapi.go exactly (confirmed +// against its real source during Phase 0 spiking, not just its docs). +func buildUAPIConfig(inst amneziawg.Instance, opts DeviceOptions) (string, error) { + var b strings.Builder + + privHex, err := wireguard.KeyToHex(inst.PrivateKey) + if err != nil { + return "", fmt.Errorf("invalid server private key: %w", err) + } + fmt.Fprintf(&b, "private_key=%s\n", privHex) + fmt.Fprintf(&b, "listen_port=%d\n", inst.ListenPort) + // replace_peers makes every apply a full resync (matches this package's + // own Manager.Ensure semantics): peers no longer in inst.Peers are + // dropped instead of lingering from a previous IpcSet call. + b.WriteString("replace_peers=true\n") + + o := inst.Obfuscation + fmt.Fprintf(&b, "jc=%d\njmin=%d\njmax=%d\n", o.Jc, o.Jmin, o.Jmax) + fmt.Fprintf(&b, "s1=%d\ns2=%d\ns3=%d\ns4=%d\n", o.S1, o.S2, o.S3, o.S4) + writeHLine(&b, "h1", o.H1) + writeHLine(&b, "h2", o.H2) + writeHLine(&b, "h3", o.H3) + writeHLine(&b, "h4", o.H4) + if o.I1 != "" { + fmt.Fprintf(&b, "i1=%s\n", o.I1) + } + + if opts.HeaderProtectionKey != "" { + hpHex, err := wireguard.KeyToHex(opts.HeaderProtectionKey) + if err != nil { + return "", fmt.Errorf("invalid header protection key: %w", err) + } + fmt.Fprintf(&b, "header_protection_key=%s\n", hpHex) + } + if opts.ContentPaddingAddition != "" { + fmt.Fprintf(&b, "content_padding_addition=%s\n", opts.ContentPaddingAddition) + } + + for _, p := range inst.Peers { + pubHex, err := wireguard.KeyToHex(p.PublicKey) + if err != nil { + return "", fmt.Errorf("peer %q: invalid public key: %w", p.Email, err) + } + fmt.Fprintf(&b, "public_key=%s\n", pubHex) + if p.PresharedKey != "" { + pskHex, err := wireguard.KeyToHex(p.PresharedKey) + if err != nil { + return "", fmt.Errorf("peer %q: invalid preshared key: %w", p.Email, err) + } + fmt.Fprintf(&b, "preshared_key=%s\n", pskHex) + } + for _, allowedIP := range p.AllowedIPs { + fmt.Fprintf(&b, "allowed_ip=%s\n", allowedIP) + } + } + + return b.String(), nil +} + +// writeHLine writes an hN UAPI line only when v is set -- an empty H value +// means "let amneziawg-go fall back to its own default," mirroring how +// internal/amneziawg's generateServerConfig treats the same optional field. +func writeHLine(b *strings.Builder, name, v string) { + if v == "" { + return + } + fmt.Fprintf(b, "%s=%s\n", name, v) +} diff --git a/internal/amneziawgnet/device_test.go b/internal/amneziawgnet/device_test.go new file mode 100644 index 000000000..667bc48d0 --- /dev/null +++ b/internal/amneziawgnet/device_test.go @@ -0,0 +1,161 @@ +package amneziawgnet + +import ( + "context" + "fmt" + "io" + "net/netip" + "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" +) + +// TestNewDeviceHandshakeForwarderAndIdentity is Phase 1's real end-to-end +// proof, not just a compile check: a genuine amneziawg-go client (via that +// project's own tun/netstack.CreateNetTUN -- the client side doesn't need a +// forwarder or peer-identity resolution, only this package's server side +// does) completes a real 3-way handshake against a Device built by +// NewDevice, dials a destination that was never configured anywhere on the +// server, and the test verifies AttachTCPForwarder recovers that exact +// destination *and* PeerIndex.Lookup resolves the connection's source back +// to the right peer's Email -- Phase 1a/1b/1c working together, the same +// mechanism Phase 0's throwaway spike validated, now as a real, repo-owned, +// repeatable test instead of scratch code. +func TestNewDeviceHandshakeForwarderAndIdentity(t *testing.T) { + 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 = 58712 // fixed loopback test port, matches the validated Phase 0 spike approach + const wantEmail = "test-peer@example.com" + + inst := amneziawg.Instance{ + Id: 1, + InterfaceName: "awgtest1", + ListenPort: listenPort, + PrivateKey: serverPriv, + PublicKey: serverPub, + Address: []string{"10.201.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.201.0.2/32"}, + }}, + } + + dev, err := NewDevice(inst, DeviceOptions{}) + if err != nil { + t.Fatalf("NewDevice: %v", err) + } + defer dev.Close() + + idx := NewPeerIndex(inst.Peers) + + type recovered struct { + email string + ok bool + dest netip.AddrPort + } + got := make(chan recovered, 1) + + // Never configured anywhere server-side: the forwarder must recover it + // purely from the decapsulated packet, not from any routing table. + wantDest := netip.MustParseAddrPort("10.201.9.9:9999") + + AttachTCPForwarder(dev.Stack, func(conn *gonet.TCPConn, dest netip.AddrPort) { + defer conn.Close() + srcAddrPort, parseErr := netip.ParseAddrPort(conn.RemoteAddr().String()) + var peer amneziawg.Peer + var ok bool + if parseErr == nil { + peer, ok = idx.Lookup(srcAddrPort.Addr().Unmap()) + } + got <- recovered{email: peer.Email, ok: ok, dest: dest} + io.Copy(io.Discard, conn) + }) + + clientTun, clientNet, err := netstack.CreateNetTUN( + []netip.Addr{netip.MustParseAddr("10.201.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) + } + // allowed_ip=0.0.0.0/0 on the client matches a real VPN client's own + // config (route everything through the tunnel) -- it's also what makes + // dialing an arbitrary, never-configured destination like wantDest + // actually get routed to the server peer at all: a narrower AllowedIPs + // here would make the client's own Device drop the packet as + // non-matching before it ever reached the wire. + 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) + } + + // Retry the dial rather than guessing a fixed handshake delay: the + // first attempts may race the handshake, later ones should succeed + // once it completes. + dialCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + var lastErr error + for { + conn, dialErr := clientNet.DialContext(dialCtx, "tcp", wantDest.String()) + if dialErr == nil { + conn.Close() + break + } + lastErr = dialErr + select { + case <-dialCtx.Done(): + t.Fatalf("client dial never succeeded: %v", lastErr) + case <-time.After(100 * time.Millisecond): + } + } + + select { + case r := <-got: + if !r.ok { + t.Fatal("forwarder: peer identity lookup failed for the recovered connection") + } + if r.email != wantEmail { + t.Errorf("resolved peer email = %q, want %q", r.email, wantEmail) + } + if r.dest != wantDest { + t.Errorf("recovered destination = %v, want %v", r.dest, wantDest) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the forwarder to hand back the recovered connection") + } +} diff --git a/internal/amneziawgnet/forwarder.go b/internal/amneziawgnet/forwarder.go new file mode 100644 index 000000000..63c13496d --- /dev/null +++ b/internal/amneziawgnet/forwarder.go @@ -0,0 +1,43 @@ +package amneziawgnet + +import ( + "net/netip" + + "gvisor.dev/gvisor/pkg/tcpip/adapters/gonet" + "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/transport/tcp" + "gvisor.dev/gvisor/pkg/waiter" +) + +// AttachTCPForwarder attaches a TCP forwarder to gstack in promiscuous + +// spoofing mode, so it accepts connections addressed to any destination -- +// not just the stack's own configured local address -- and hands the +// handler both the accepted connection and the tunnel client's real, +// dynamically-arbitrary destination (recovered from the connection's own +// TransportEndpointID, not from any preconfigured routing table). This is +// the mechanism the whole embedded-AmneziaWG design depends on: what the +// handler does with that destination (dial it directly, relay it into +// Xray's SOCKS5 inbound, ...) is entirely up to the caller. +// +// Adapted from xtls/xray-core's proxy/wireguard/tun.go createForwarder (MIT). +func AttachTCPForwarder(gstack *stack.Stack, handler func(conn *gonet.TCPConn, dest netip.AddrPort)) { + enablePromiscuousRouting(gstack) + + fwd := tcp.NewForwarder(gstack, 0, 65535, func(r *tcp.ForwarderRequest) { + go func(r *tcp.ForwarderRequest) { + var wq waiter.Queue + id := r.ID() + + ep, err := r.CreateEndpoint(&wq) + if err != nil { + r.Complete(true) + return + } + dest := netip.AddrPortFrom(addrFromTcpip(id.LocalAddress), id.LocalPort) + handler(gonet.NewTCPConn(&wq, ep), dest) + ep.Close() + r.Complete(false) + }(r) + }) + gstack.SetTransportProtocolHandler(tcp.ProtocolNumber, fwd.HandlePacket) +} diff --git a/internal/amneziawgnet/identity.go b/internal/amneziawgnet/identity.go new file mode 100644 index 000000000..087774f00 --- /dev/null +++ b/internal/amneziawgnet/identity.go @@ -0,0 +1,62 @@ +package amneziawgnet + +import ( + "net/netip" + + "github.com/mhsanaei/3x-ui/v3/internal/amneziawg" +) + +// PeerIndex resolves a decapsulated connection's tunnel-internal source +// address back to the peer it belongs to, the same role Xray-core's own +// wireguard proxy's GetUserByAddr plays -- sourced here from an +// amneziawg.Instance's own Peers (already carries Email per peer, no new +// data needed) rather than a separate user table. +type PeerIndex struct { + entries []peerIndexEntry +} + +type peerIndexEntry struct { + prefix netip.Prefix + peer amneziawg.Peer +} + +// NewPeerIndex builds a lookup index from peers' AllowedIPs. Entries with an +// unparseable AllowedIPs value are skipped rather than failing the whole +// index -- by the time an Instance reaches this package, AllowedIPs has +// already been accepted at save time (see internal/amneziawg's own +// validation), so a bad entry here would only mean stale/manually-edited +// data, not something worth refusing to serve the rest of the peers over. +func NewPeerIndex(peers []amneziawg.Peer) *PeerIndex { + idx := &PeerIndex{} + for _, p := range peers { + for _, a := range p.AllowedIPs { + prefix, err := netip.ParsePrefix(a) + if err != nil { + continue + } + idx.entries = append(idx.entries, peerIndexEntry{prefix: prefix, peer: p}) + } + } + return idx +} + +// Lookup returns the peer whose AllowedIPs most specifically contains addr -- +// the same longest-prefix-match rule a real AmneziaWG interface's own +// AllowedIPs routing table uses for outbound packets, applied here in +// reverse to attribute an inbound (tunnel-internal-source) packet back to +// its owning peer. +func (idx *PeerIndex) Lookup(addr netip.Addr) (amneziawg.Peer, bool) { + bestBits := -1 + var bestPeer amneziawg.Peer + for _, e := range idx.entries { + if e.prefix.Bits() <= bestBits || !e.prefix.Contains(addr) { + continue + } + bestBits = e.prefix.Bits() + bestPeer = e.peer + } + if bestBits < 0 { + return amneziawg.Peer{}, false + } + return bestPeer, true +} diff --git a/internal/amneziawgnet/manager.go b/internal/amneziawgnet/manager.go new file mode 100644 index 000000000..ca8276ae8 --- /dev/null +++ b/internal/amneziawgnet/manager.go @@ -0,0 +1,180 @@ +package amneziawgnet + +import ( + "fmt" + "strings" + "sync" + + "github.com/mhsanaei/3x-ui/v3/internal/amneziawg" + "github.com/mhsanaei/3x-ui/v3/internal/logger" +) + +// Desired pairs an amneziawg.Instance (the shared, DB-backed shape +// internal/amneziawg's own kernel-module Manager also reconciles toward) +// with this package's own embedded-only DeviceOptions -- the AWG 3.0 fields +// that shared type doesn't carry, see DeviceOptions' doc comment. +type Desired struct { + Instance amneziawg.Instance + Options DeviceOptions +} + +// managed is one running embedded interface: the live Device, the peer +// lookup index built from its current peer list, and enough of its own +// configuration to decide whether a later Ensure call can reconfigure it in +// place or needs to rebuild it from scratch. +type managed struct { + dev *Device + peers *PeerIndex + inst amneziawg.Instance + structFP string +} + +// Manager owns the set of running embedded AmneziaWG interfaces, keyed by +// inbound id -- the same shape as internal/amneziawg.Manager (GetManager() +// + sync.Once, mu-guarded map, Ensure/Reconcile/StopAll/HasRunning), so a +// caller already familiar with that Manager needs to learn nothing new here. +// Unlike that Manager, this one doesn't attach any traffic handling by +// itself: Ensure/Reconcile only bring each Instance's Device up to date. +// Attaching a forwarder/UDP handler (see forwarder.go / udp.go) using the +// Device and PeerIndex returned by Lookup is left to the caller -- today a +// test harness, later the Phase 2 SOCKS5 relay wiring -- since this package +// doesn't yet know what that handler should do with a recovered connection. +type Manager struct { + mu sync.Mutex + ifaces map[int]*managed +} + +var ( + managerOnce sync.Once + manager *Manager +) + +// GetManager returns the process-wide embedded-AmneziaWG manager singleton. +func GetManager() *Manager { + managerOnce.Do(func() { + manager = &Manager{ifaces: map[int]*managed{}} + }) + return manager +} + +// Ensure brings inbound d.Instance.Id's embedded interface to the state +// d describes, creating it if it doesn't exist yet. A no-op only when +// nothing has changed since the last successful Ensure/Reconcile. +func (m *Manager) Ensure(d Desired) error { + m.mu.Lock() + defer m.mu.Unlock() + return m.ensureLocked(d) +} + +// ensureLocked decides between three actions: nothing changed since the +// last apply (skip entirely); only peers/obfuscation/keys/listen_port +// changed (reconfigure the existing Device in place via IpcSet, which +// already sends replace_peers=true -- see buildUAPIConfig -- so removed +// peers are dropped correctly without a full rebuild); or the interface's +// own address(es)/MTU changed (these are fixed at netstack-construction +// time, so the only option is closing the old Device and building a fresh +// one). This is a coarser split than internal/amneziawg's own three-tier +// noop/reload/restart fingerprinting (that one also tracks host-side +// TPROXY/NDP rules this embedded path has no equivalent of) -- correct and +// sufficient for Phase 1; revisit only if reconcile frequency at real scale +// makes the address/MTU rebuild path worth avoiding too. +func (m *Manager) ensureLocked(d Desired) error { + inst, opts := d.Instance, d.Options + structFP := addressFingerprint(inst) + + cur, exists := m.ifaces[inst.Id] + if exists && cur.structFP == structFP { + conf, err := buildUAPIConfig(inst, opts) + if err != nil { + return fmt.Errorf("amneziawgnet: %w", err) + } + if err := cur.dev.IpcSet(conf); err != nil { + return fmt.Errorf("amneziawgnet: reconfigure inbound %d: %w", inst.Id, err) + } + cur.peers = NewPeerIndex(inst.Peers) + cur.inst = inst + return nil + } + + if exists { + cur.dev.Close() + delete(m.ifaces, inst.Id) + } + dev, err := NewDevice(inst, opts) + if err != nil { + return err + } + m.ifaces[inst.Id] = &managed{ + dev: dev, + peers: NewPeerIndex(inst.Peers), + inst: inst, + structFP: structFP, + } + logger.Infof("amneziawgnet: started embedded interface %s for inbound %d", inst.InterfaceName, inst.Id) + return nil +} + +// addressFingerprint captures the two Instance fields that can't be changed +// on a running Device via IpcSet alone (they're fixed when the gVisor +// netstack is built) -- everything else (keys, listen port, obfuscation, +// AWG 3.0 options, peers) amneziawg-go's own UAPI can hot-reconfigure. +func addressFingerprint(inst amneziawg.Instance) string { + return fmt.Sprintf("%d|%s", inst.MTU, strings.Join(inst.Address, ",")) +} + +// Reconcile brings every desired instance's embedded interface up to date +// and stops any managed interface whose inbound is no longer desired -- +// mirroring internal/amneziawg.Manager.Reconcile's per-tick contract. +func (m *Manager) Reconcile(desired []Desired) { + m.mu.Lock() + defer m.mu.Unlock() + + want := make(map[int]struct{}, len(desired)) + for _, d := range desired { + want[d.Instance.Id] = struct{}{} + } + for id, cur := range m.ifaces { + if _, ok := want[id]; ok { + continue + } + cur.dev.Close() + delete(m.ifaces, id) + logger.Infof("amneziawgnet: stopped embedded interface for removed inbound %d", id) + } + for _, d := range desired { + if err := m.ensureLocked(d); err != nil { + logger.Warningf("amneziawgnet: reconcile failed for inbound %d: %v", d.Instance.Id, err) + } + } +} + +// StopAll tears down every managed interface. Called on panel shutdown. +func (m *Manager) StopAll() { + m.mu.Lock() + defer m.mu.Unlock() + for id, cur := range m.ifaces { + cur.dev.Close() + delete(m.ifaces, id) + } +} + +// HasRunning reports whether any embedded interface is currently managed. +func (m *Manager) HasRunning() bool { + m.mu.Lock() + defer m.mu.Unlock() + return len(m.ifaces) > 0 +} + +// Lookup returns the running Device and PeerIndex for inbound id, if any -- +// for a caller that wants to attach its own forwarder/handler (a test +// harness today, the Phase 2 SOCKS5 relay wiring later) once the interface +// is up. +func (m *Manager) Lookup(id int) (dev *Device, peers *PeerIndex, ok bool) { + m.mu.Lock() + defer m.mu.Unlock() + cur, exists := m.ifaces[id] + if !exists { + return nil, nil, false + } + return cur.dev, cur.peers, true +} diff --git a/internal/amneziawgnet/manager_test.go b/internal/amneziawgnet/manager_test.go new file mode 100644 index 000000000..3f2741c2b --- /dev/null +++ b/internal/amneziawgnet/manager_test.go @@ -0,0 +1,85 @@ +package amneziawgnet + +import ( + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/amneziawg" + "github.com/mhsanaei/3x-ui/v3/internal/util/wireguard" +) + +// TestManagerLifecycle exercises Ensure/Reconcile's reconfigure-in-place vs. +// rebuild split (see ensureLocked's doc comment) and Reconcile's stop path, +// using a throwaway Manager rather than the process-wide singleton so this +// test doesn't interact with any other test's state. +func TestManagerLifecycle(t *testing.T) { + priv, pub, err := wireguard.GenerateWireguardKeypair() + if err != nil { + t.Fatalf("generate keypair: %v", err) + } + + m := &Manager{ifaces: map[int]*managed{}} + inst := amneziawg.Instance{ + Id: 3, + InterfaceName: "awgtest3", + ListenPort: 58714, + PrivateKey: priv, + PublicKey: pub, + Address: []string{"10.203.0.1/24"}, + MTU: 1420, + Obfuscation: amneziawg.Obfuscation20{ + Jc: 4, Jmin: 40, Jmax: 70, + S1: 20, S2: 30, S3: 20, S4: 20, + }, + } + defer m.StopAll() + + if err := m.Ensure(Desired{Instance: inst}); err != nil { + t.Fatalf("Ensure (create): %v", err) + } + if !m.HasRunning() { + t.Fatal("HasRunning() = false after Ensure created an interface") + } + dev1, _, ok := m.Lookup(inst.Id) + if !ok { + t.Fatal("Lookup after Ensure: not found") + } + + // Same Instance again: same address fingerprint, so this should + // reconfigure the existing Device via IpcSet rather than rebuild it -- + // verify by checking the *Device pointer survived unchanged. + if err := m.Ensure(Desired{Instance: inst}); err != nil { + t.Fatalf("Ensure (unchanged): %v", err) + } + dev2, _, ok := m.Lookup(inst.Id) + if !ok { + t.Fatal("Lookup after second Ensure: not found") + } + if dev1 != dev2 { + t.Error("Ensure with an unchanged Instance rebuilt the Device; expected an in-place reconfigure") + } + + // Changing the interface address is structural (fixed at netstack + // construction time) and must force a rebuild -- verify by checking the + // *Device pointer changed. + changed := inst + changed.Address = []string{"10.203.1.1/24"} + if err := m.Ensure(Desired{Instance: changed}); err != nil { + t.Fatalf("Ensure (address changed): %v", err) + } + dev3, _, ok := m.Lookup(inst.Id) + if !ok { + t.Fatal("Lookup after address-changing Ensure: not found") + } + if dev3 == dev2 { + t.Error("Ensure with a changed address reconfigured in place; expected a rebuild") + } + + // Reconcile with nothing desired stops every managed interface. + m.Reconcile(nil) + if m.HasRunning() { + t.Error("HasRunning() = true after Reconcile([]) should have stopped everything") + } + if _, _, ok := m.Lookup(inst.Id); ok { + t.Error("Lookup succeeded after Reconcile([]) removed the interface") + } +} diff --git a/internal/amneziawgnet/netstack.go b/internal/amneziawgnet/netstack.go new file mode 100644 index 000000000..984ebceaa --- /dev/null +++ b/internal/amneziawgnet/netstack.go @@ -0,0 +1,194 @@ +// Package amneziawgnet embeds amneziawg-go (a userspace AmneziaWG +// implementation, https://github.com/amnezia-vpn/amneziawg-go) directly in +// the panel process, as an alternative to internal/amneziawg's +// kernel-module (DKMS) + awg-quick approach. A gVisor userspace network +// stack (gvisor.dev/gvisor/pkg/tcpip -- already an indirect dependency via +// xray-core's own proxy/wireguard support) terminates each tunnel, and a +// forwarder recovers each connection's real, dynamically-arbitrary +// destination for the caller to relay onward (see Phase 2 of the migration +// plan: a loopback SOCKS5 dial into Xray, giving native stats/routing/ +// sniffing for free). +package amneziawgnet + +import ( + "fmt" + "net/netip" + "os" + "syscall" + + awgtun "github.com/amnezia-vpn/amneziawg-go/v3/tun" + + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/link/channel" + "gvisor.dev/gvisor/pkg/tcpip/network/ipv4" + "gvisor.dev/gvisor/pkg/tcpip/network/ipv6" + "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/transport/icmp" + "gvisor.dev/gvisor/pkg/tcpip/transport/tcp" + "gvisor.dev/gvisor/pkg/tcpip/transport/udp" +) + +// stackTun implements amneziawg-go's tun.Device directly against a gVisor +// channel endpoint, the same approach amneziawg-go's own tun/netstack +// package and xray-core's proxy/wireguard/netstack.go both take. Neither of +// those exposes the raw *stack.Stack a forwarder needs (amneziawg-go's Net +// type keeps it unexported), so this is a local, from-source reimplementation +// rather than a wrapper -- adapted from amneziawg-go v3.0.3's +// tun/netstack/tun.go (MIT licensed), trimmed to the constructor this +// package needs. +type stackTun struct { + ep *channel.Endpoint + stack *stack.Stack + events chan awgtun.Event + notifyHandle *channel.NotificationHandle + incomingPacket chan *buffer.View + mtu int +} + +// createNetTUNWithStack builds a gVisor-backed tun.Device for the given +// local addresses (interface address(es), one per family) and returns the +// underlying *stack.Stack alongside it so a caller can attach a forwarder +// (see forwarder.go / udp.go). +func createNetTUNWithStack(localAddresses []netip.Addr, mtu int) (awgtun.Device, *stack.Stack, error) { + opts := stack.Options{ + NetworkProtocols: []stack.NetworkProtocolFactory{ipv4.NewProtocol, ipv6.NewProtocol}, + TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol, udp.NewProtocol, icmp.NewProtocol6, icmp.NewProtocol4}, + // HandleLocal must stay false: promiscuous+spoofing mode (see + // forwarder.go) is what lets a destination other than the stack's + // own configured address reach the forwarder at all. + HandleLocal: false, + } + dev := &stackTun{ + ep: channel.New(1024, uint32(mtu), ""), + stack: stack.New(opts), + events: make(chan awgtun.Event, 10), + incomingPacket: make(chan *buffer.View), + mtu: mtu, + } + sackEnabledOpt := tcpip.TCPSACKEnabled(true) + if err := dev.stack.SetTransportProtocolOption(tcp.ProtocolNumber, &sackEnabledOpt); err != nil { + return nil, nil, fmt.Errorf("amneziawgnet: enable TCP SACK: %s", err) + } + dev.notifyHandle = dev.ep.AddNotify(dev) + if err := dev.stack.CreateNIC(1, dev.ep); err != nil { + return nil, nil, fmt.Errorf("amneziawgnet: CreateNIC: %s", err) + } + + var hasV4, hasV6 bool + for _, ip := range localAddresses { + var protoNumber tcpip.NetworkProtocolNumber + switch { + case ip.Is4(): + protoNumber = ipv4.ProtocolNumber + hasV4 = true + case ip.Is6(): + protoNumber = ipv6.ProtocolNumber + hasV6 = true + default: + continue + } + protoAddr := tcpip.ProtocolAddress{ + Protocol: protoNumber, + AddressWithPrefix: tcpip.AddrFromSlice(ip.AsSlice()).WithPrefix(), + } + if err := dev.stack.AddProtocolAddress(1, protoAddr, stack.AddressProperties{}); err != nil { + return nil, nil, fmt.Errorf("amneziawgnet: AddProtocolAddress(%v): %s", ip, err) + } + } + if hasV4 { + dev.stack.AddRoute(tcpip.Route{Destination: header.IPv4EmptySubnet, NIC: 1}) + } + if hasV6 { + dev.stack.AddRoute(tcpip.Route{Destination: header.IPv6EmptySubnet, NIC: 1}) + } + dev.events <- awgtun.EventUp + return dev, dev.stack, nil +} + +func (t *stackTun) Name() (string, error) { return "amneziawgnet", nil } +func (t *stackTun) File() *os.File { return nil } +func (t *stackTun) Events() <-chan awgtun.Event { return t.events } +func (t *stackTun) MTU() (int, error) { return t.mtu, nil } +func (t *stackTun) BatchSize() int { return 1 } + +func (t *stackTun) Read(buf [][]byte, sizes []int, offset int) (int, error) { + view, ok := <-t.incomingPacket + if !ok { + return 0, os.ErrClosed + } + n, err := view.Read(buf[0][offset:]) + if err != nil { + return 0, err + } + sizes[0] = n + return 1, nil +} + +func (t *stackTun) Write(buf [][]byte, offset int) (int, error) { + for _, b := range buf { + packet := b[offset:] + if len(packet) == 0 { + continue + } + pkb := stack.NewPacketBuffer(stack.PacketBufferOptions{Payload: buffer.MakeWithData(packet)}) + switch packet[0] >> 4 { + case 4: + t.ep.InjectInbound(header.IPv4ProtocolNumber, pkb) + case 6: + t.ep.InjectInbound(header.IPv6ProtocolNumber, pkb) + default: + return 0, syscall.EAFNOSUPPORT + } + } + return len(buf), nil +} + +func (t *stackTun) WriteNotify() { + pkt := t.ep.Read() + if pkt == nil { + return + } + view := pkt.ToView() + pkt.DecRef() + t.incomingPacket <- view +} + +func (t *stackTun) Close() error { + t.stack.RemoveNIC(1) + t.stack.Close() + t.ep.RemoveNotify(t.notifyHandle) + t.ep.Close() + if t.events != nil { + close(t.events) + } + if t.incomingPacket != nil { + close(t.incomingPacket) + } + return nil +} + +// enablePromiscuousRouting puts the NIC into promiscuous + spoofing mode, +// the precondition both AttachTCPForwarder and AttachUDPHandler need to see +// packets addressed to a destination other than the stack's own configured +// local address. Safe to call from both (and more than once): gVisor's +// SetPromiscuousMode/SetSpoofing just set a bool on the NIC, not something +// that accumulates or needs undoing between calls. +func enablePromiscuousRouting(gstack *stack.Stack) { + gstack.SetPromiscuousMode(1, true) + gstack.SetSpoofing(1, true) +} + +// addrFromTcpip converts a gVisor tcpip.Address (4 or 16 raw bytes) to the +// stdlib netip.Addr type the rest of this package and its callers use. +func addrFromTcpip(a tcpip.Address) netip.Addr { + if a.Len() == 4 { + var b [4]byte + copy(b[:], a.AsSlice()) + return netip.AddrFrom4(b) + } + var b [16]byte + copy(b[:], a.AsSlice()) + return netip.AddrFrom16(b) +} diff --git a/internal/amneziawgnet/udp.go b/internal/amneziawgnet/udp.go new file mode 100644 index 000000000..52814de4e --- /dev/null +++ b/internal/amneziawgnet/udp.go @@ -0,0 +1,100 @@ +package amneziawgnet + +import ( + "fmt" + "net/netip" + + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/checksum" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/transport/udp" +) + +// UDPHandler is called for every UDP packet a tunnel client sends, with its +// source (the peer's tunnel-internal address) and its real, +// dynamically-arbitrary destination -- recovered the same way the TCP +// forwarder recovers its destination, from the packet's own transport +// endpoint ID, never from a preconfigured table. The handler owns all flow +// tracking and reply delivery (via WriteUDPReply): gVisor has no +// udp.NewForwarder the way it does for TCP, so unlike AttachTCPForwarder +// this can't just hand back a ready net.Conn. +type UDPHandler func(src, dst netip.AddrPort, payload []byte) + +// AttachUDPHandler attaches a raw UDP handler to gstack, independently +// enabling the same promiscuous+spoofing mode AttachTCPForwarder needs -- +// safe and idempotent to call regardless of whether AttachTCPForwarder was +// attached to the same stack first, or at all. Adapted from xtls/xray-core's +// proxy/wireguard/tun.go UDP path (MIT), which hand-tracks flows for the +// identical reason: gVisor doesn't provide a UDP forwarder. +func AttachUDPHandler(gstack *stack.Stack, handler UDPHandler) { + enablePromiscuousRouting(gstack) + + gstack.SetTransportProtocolHandler(udp.ProtocolNumber, func(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { + data := pkt.Clone().Data().AsRange().ToSlice() + src := netip.AddrPortFrom(addrFromTcpip(id.RemoteAddress), id.RemotePort) + dst := netip.AddrPortFrom(addrFromTcpip(id.LocalAddress), id.LocalPort) + handler(src, dst, data) + return true + }) +} + +// WriteUDPReply injects a UDP packet into gstack as if it arrived from +// `from` addressed to `to` -- i.e. a reply travelling back into the tunnel +// toward the client -- constructed by hand since gVisor exposes no +// connected-socket-style Write for an address the stack doesn't itself own. +func WriteUDPReply(gstack *stack.Stack, from, to netip.AddrPort, payload []byte) error { + udpLen := header.UDPMinimumSize + len(payload) + srcIP := tcpip.AddrFromSlice(from.Addr().AsSlice()) + dstIP := tcpip.AddrFromSlice(to.Addr().AsSlice()) + + isIPv4 := from.Addr().Is4() + ipHdrSize := header.IPv6MinimumSize + ipProtocol := header.IPv6ProtocolNumber + if isIPv4 { + ipHdrSize = header.IPv4MinimumSize + ipProtocol = header.IPv4ProtocolNumber + } + + pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ + ReserveHeaderBytes: ipHdrSize + header.UDPMinimumSize, + Payload: buffer.MakeWithData(payload), + }) + defer pkt.DecRef() + + udpHdr := header.UDP(pkt.TransportHeader().Push(header.UDPMinimumSize)) + udpHdr.Encode(&header.UDPFields{ + SrcPort: from.Port(), + DstPort: to.Port(), + Length: uint16(udpLen), + }) + xsum := header.PseudoHeaderChecksum(header.UDPProtocolNumber, srcIP, dstIP, uint16(udpLen)) + udpHdr.SetChecksum(^udpHdr.CalculateChecksum(checksum.Checksum(payload, xsum))) + + if isIPv4 { + ipHdr := header.IPv4(pkt.NetworkHeader().Push(header.IPv4MinimumSize)) + ipHdr.Encode(&header.IPv4Fields{ + TotalLength: uint16(header.IPv4MinimumSize + udpLen), + TTL: 64, + Protocol: uint8(header.UDPProtocolNumber), + SrcAddr: srcIP, + DstAddr: dstIP, + }) + ipHdr.SetChecksum(^ipHdr.CalculateChecksum()) + } else { + ipHdr := header.IPv6(pkt.NetworkHeader().Push(header.IPv6MinimumSize)) + ipHdr.Encode(&header.IPv6Fields{ + PayloadLength: uint16(udpLen), + TransportProtocol: header.UDPProtocolNumber, + HopLimit: 64, + SrcAddr: srcIP, + DstAddr: dstIP, + }) + } + + if tcpipErr := gstack.WriteRawPacket(1, ipProtocol, buffer.MakeWithView(pkt.ToView())); tcpipErr != nil { + return fmt.Errorf("amneziawgnet: WriteRawPacket: %s", tcpipErr) + } + return nil +} diff --git a/internal/amneziawgnet/udp_test.go b/internal/amneziawgnet/udp_test.go new file mode 100644 index 000000000..8ca31e42f --- /dev/null +++ b/internal/amneziawgnet/udp_test.go @@ -0,0 +1,150 @@ +package amneziawgnet + +import ( + "fmt" + "net/netip" + "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" + + "github.com/mhsanaei/3x-ui/v3/internal/amneziawg" + "github.com/mhsanaei/3x-ui/v3/internal/util/wireguard" +) + +// TestNewDeviceUDPHandlerAndReply is the UDP counterpart of +// TestNewDeviceHandshakeForwarderAndIdentity: this package's own udp.go was +// refactored from the Phase 0 spike's bake-the-dial-in version to a generic +// handler-plus-reply-injection design (see AttachUDPHandler/WriteUDPReply's +// doc comments), a real behavior change worth its own verification rather +// than assuming the port preserved correctness -- UDP was flagged as "the +// harder half" in the migration plan's own risk list, precisely because +// gVisor has no udp.NewForwarder and the reply path has to be constructed +// by hand. +func TestNewDeviceUDPHandlerAndReply(t *testing.T) { + 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 = 58713 // distinct from the TCP test's port + const wantEmail = "udp-test-peer@example.com" + const echoPayload = "hello-from-client" + + inst := amneziawg.Instance{ + Id: 2, + InterfaceName: "awgtest2", + ListenPort: listenPort, + PrivateKey: serverPriv, + PublicKey: serverPub, + Address: []string{"10.202.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.202.0.2/32"}, + }}, + } + + dev, err := NewDevice(inst, DeviceOptions{}) + if err != nil { + t.Fatalf("NewDevice: %v", err) + } + defer dev.Close() + + idx := NewPeerIndex(inst.Peers) + // Never configured anywhere server-side, same idea as the TCP test. + wantDest := netip.MustParseAddrPort("10.202.9.9:5353") + + identityErrCh := make(chan error, 8) + AttachUDPHandler(dev.Stack, func(src, dst netip.AddrPort, payload []byte) { + if peer, ok := idx.Lookup(src.Addr()); !ok || peer.Email != wantEmail { + identityErrCh <- fmt.Errorf("peer identity lookup for src %v: ok=%v email=%q, want %q", src, ok, peer.Email, wantEmail) + return + } + if dst != wantDest { + identityErrCh <- fmt.Errorf("recovered dest = %v, want %v", dst, wantDest) + return + } + // Echo the payload back, posing as a reply from the destination the + // client dialed -- exactly what a real relay's downstream reply + // would look like from the tunnel's point of view. + if err := WriteUDPReply(dev.Stack, dst, src, payload); err != nil { + identityErrCh <- fmt.Errorf("WriteUDPReply: %w", err) + } + }) + + clientTun, clientNet, err := netstack.CreateNetTUN( + []netip.Addr{netip.MustParseAddr("10.202.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) + } + + conn, err := clientNet.DialUDPAddrPort(netip.AddrPort{}, wantDest) + if err != nil { + t.Fatalf("client DialUDPAddrPort: %v", err) + } + defer conn.Close() + + deadline := time.Now().Add(5 * time.Second) + var buf [256]byte + for { + select { + case err := <-identityErrCh: + t.Fatal(err) + default: + } + + _ = conn.SetWriteDeadline(time.Now().Add(200 * time.Millisecond)) + if _, err := conn.Write([]byte(echoPayload)); err != nil { + if time.Now().After(deadline) { + t.Fatalf("client write never succeeded: %v", err) + } + continue + } + + _ = conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + n, err := conn.Read(buf[:]) + if err != nil { + if time.Now().After(deadline) { + t.Fatalf("client never received a reply: %v", err) + } + continue + } + if got := string(buf[:n]); got != echoPayload { + t.Fatalf("echoed payload = %q, want %q", got, echoPayload) + } + return + } +}