From bc2c64d901c28b994bc337a4390c18b7cc99876a Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sun, 1 Nov 2026 00:19:34 +0100 Subject: [PATCH] fix(amneziawg): bound the egress relay's pending tunnel DNS lookups relayUDP started a goroutine for every domain-addressed datagram and each lookup opened its own netstack UDP endpoint, so the lookups in flight were bounded only by the client's send rate: one proxy user could grow the panel by gigabytes and exhaust the outbound stack's ports, stalling every user of that outbound. One association may now have 32 datagrams waiting on DNS and drops the rest, a cached name is delivered from the reader loop without a lookup, and resolveTunnelVia takes one of 1024 process-wide slots before it opens an endpoint, which also covers the TCP CONNECT path. --- internal/amneziawgnet/dns.go | 41 +++++-- internal/amneziawgnet/egress.go | 17 ++- internal/amneziawgnet/egress_domain_test.go | 114 ++++++++++++++++++++ 3 files changed, 162 insertions(+), 10 deletions(-) diff --git a/internal/amneziawgnet/dns.go b/internal/amneziawgnet/dns.go index 15f65414a..b70bcbbfa 100644 --- a/internal/amneziawgnet/dns.go +++ b/internal/amneziawgnet/dns.go @@ -71,26 +71,49 @@ var tunnelDNSCache = struct { const ( tunnelDNSCacheTTL = 60 * time.Second tunnelDNSCacheMaxSize = 1024 + tunnelDNSMaxInFlight = 1024 ) +// tunnelDNSSlots caps lookups in flight across all outbounds: each one holds a +// netstack port until the resolver answers or the lookup times out. +var tunnelDNSSlots = make(chan struct{}, tunnelDNSMaxInFlight) + // dnsCacheKey computes cache key scoped by outbound tag, server, and host. func dnsCacheKey(tag, dnsServer, host string) string { return tag + "|" + dnsServer + "|" + host } -func resolveTunnelVia(ctx context.Context, dev *Device, tag string, dnsServer string, host string) (netip.Addr, error) { - normDNS := amneziawg.NormalizeDNSServer(dnsServer) - if normDNS == "" { - normDNS = defaultDNSFor(dev.LocalAddresses()) +// tunnelDNSServerFor normalizes dnsServer, empty meaning the family default. +func tunnelDNSServerFor(dev *Device, dnsServer string) string { + if normDNS := amneziawg.NormalizeDNSServer(dnsServer); normDNS != "" { + return normDNS } + return defaultDNSFor(dev.LocalAddresses()) +} + +// cachedTunnelAddr returns host's unexpired cache entry; it never queries. +func cachedTunnelAddr(dev *Device, tag string, dnsServer string, host string) (netip.Addr, bool) { + key := dnsCacheKey(tag, tunnelDNSServerFor(dev, dnsServer), host) + tunnelDNSCache.mu.Lock() + defer tunnelDNSCache.mu.Unlock() + e, ok := tunnelDNSCache.m[key] + return e.addr, ok && time.Now().Before(e.exp) +} + +func resolveTunnelVia(ctx context.Context, dev *Device, tag string, dnsServer string, host string) (netip.Addr, error) { + if addr, ok := cachedTunnelAddr(dev, tag, dnsServer, host); ok { + return addr, nil + } + normDNS := tunnelDNSServerFor(dev, dnsServer) key := dnsCacheKey(tag, normDNS, host) now := time.Now() - tunnelDNSCache.mu.Lock() - if e, ok := tunnelDNSCache.m[key]; ok && now.Before(e.exp) { - tunnelDNSCache.mu.Unlock() - return e.addr, nil + + select { + case tunnelDNSSlots <- struct{}{}: + defer func() { <-tunnelDNSSlots }() + case <-ctx.Done(): + return netip.Addr{}, fmt.Errorf("dns lookup %q: too many lookups in flight: %w", host, ctx.Err()) } - tunnelDNSCache.mu.Unlock() server, err := netip.ParseAddrPort(normDNS) if err != nil { diff --git a/internal/amneziawgnet/egress.go b/internal/amneziawgnet/egress.go index ab7816696..0507f447e 100644 --- a/internal/amneziawgnet/egress.go +++ b/internal/amneziawgnet/egress.go @@ -529,6 +529,10 @@ func (b *egressUDPBudget) release(sess *egressUDPSession) { b.mu.Unlock() } +// maxPendingUDPResolves caps the domain datagrams one association may have +// waiting on tunnel DNS; each holds a goroutine and a copy of the datagram. +const maxPendingUDPResolves = 32 + // relayUDP answers the associate request and relays datagrams to // per-destination tunnel endpoints until the control connection closes. func (s *socks5EgressServer) relayUDP(dev *Device, tag string, ctl udpControl, _ socksTarget) { @@ -551,6 +555,7 @@ func (s *socks5EgressServer) relayUDP(dev *Device, tag string, ctl udpControl, _ } sessions := &udpEgressSessions{m: map[netip.AddrPort]*egressUDPSession{}} + pendingResolves := make(chan struct{}, maxPendingUDPResolves) // Reader: strip per-datagram SOCKS5 headers and forward into the tunnel; // only the associated client's address is accepted. @@ -580,10 +585,20 @@ func (s *socks5EgressServer) relayUDP(dev *Device, tag string, ctl udpControl, _ if perr != nil { continue } + dnsSrv := s.currentDNSServer(tag) + if daddr, ok := cachedTunnelAddr(dev, tag, dnsSrv, name); ok { + s.deliverUDPDatagram(dev, tag, udpConn, client, sessions, netip.AddrPortFrom(daddr, port), data[hdrLen:]) + continue + } + select { + case pendingResolves <- struct{}{}: + default: + continue // enough lookups pending; UDP may drop + } // Resolve off reader loop so slow tunnel DNS lookups do not // stall other destinations on this association. go func(client netip.AddrPort, name string, port uint16, hdrLen int, datagram []byte) { - dnsSrv := s.currentDNSServer(tag) + defer func() { <-pendingResolves }() rctx, rcancel := context.WithTimeout(context.Background(), tunnelResolveTimeout) daddr, rerr := resolveTunnelVia(rctx, dev, tag, dnsSrv, name) rcancel() diff --git a/internal/amneziawgnet/egress_domain_test.go b/internal/amneziawgnet/egress_domain_test.go index f88dcab07..164cc598d 100644 --- a/internal/amneziawgnet/egress_domain_test.go +++ b/internal/amneziawgnet/egress_domain_test.go @@ -2,11 +2,14 @@ package amneziawgnet import ( "bytes" + "context" "encoding/binary" + "errors" "fmt" "io" "net" "net/netip" + "runtime" "strconv" "strings" "testing" @@ -643,6 +646,117 @@ func TestEgressUDPDatagramDomainInterleavedClients(t *testing.T) { } } +// TestEgressUDPDatagramDomainFloodBoundsPendingLookups ensures datagrams to a +// name whose lookup never completes do not pin a goroutine and a port each. +func TestEgressUDPDatagramDomainFloodBoundsPendingLookups(t *testing.T) { + tun := newPairedTunnelForTest(t) + registerEgressDeviceForTest(t, tun.client) + srv := GetEgressServer() + prev := srv.currentDNSServer() + srv.SetDNSServer(net.JoinHostPort(tun.serverIP.String(), "53")) + t.Cleanup(func() { srv.SetDNSServer(prev) }) + resetTunnelDNSCacheForTest() + + silentDNS, err := gonet.DialUDP(tun.server.Stack, &tcpip.FullAddress{NIC: 1, Port: 53}, nil, ipv4.ProtocolNumber) + if err != nil { + t.Fatal(err) + } + defer silentDNS.Close() + + ctl, err := (&net.Dialer{Timeout: egressTestDialTimeout}).Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(srv.Port()))) + if err != nil { + t.Fatal(err) + } + defer ctl.Close() + + socksAuth(t, ctl) + if _, err := ctl.Write([]byte{0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0}); err != nil { + t.Fatal(err) + } + reply := make([]byte, 10) + if _, err := io.ReadFull(ctl, reply); err != nil || reply[1] != 0x00 { + t.Fatalf("associate failed: %v %v", err, reply) + } + bindPort := binary.BigEndian.Uint16(reply[8:10]) + + udp, err := net.DialUDP("udp", nil, &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: int(bindPort)}) + if err != nil { + t.Fatal(err) + } + defer udp.Close() + + name := "never-answers.internal" + dgram := make([]byte, 0, 5+len(name)+2+1) + dgram = append(dgram, 0x00, 0x00, 0x00, 0x03, byte(len(name))) + dgram = append(dgram, name...) + dgram = append(dgram, 0x27, 0x0f, 'x') + + const flood = 2000 + before := runtime.NumGoroutine() + for i := 0; i < flood; i++ { + if _, err := udp.Write(dgram); err != nil { + t.Fatal(err) + } + if i%50 == 49 { + time.Sleep(time.Millisecond) + } + } + time.Sleep(100 * time.Millisecond) + if grown := runtime.NumGoroutine() - before; grown > maxPendingUDPResolves+8 { + t.Fatalf("%d datagrams left %d goroutines behind, want at most %d pending lookups", flood, grown, maxPendingUDPResolves) + } + + // A cached name takes no lookup slot, so it still gets through. + in, err := gonet.DialUDP(tun.server.Stack, &tcpip.FullAddress{NIC: 1, Port: 9999}, nil, ipv4.ProtocolNumber) + if err != nil { + t.Fatal(err) + } + defer in.Close() + cached := "cached.internal" + tunnelDNSCache.mu.Lock() + tunnelDNSCache.m[dnsCacheKey("awg-dom-test", srv.currentDNSServer(), cached)] = tunnelDNSCacheEntry{addr: tun.serverIP, exp: time.Now().Add(time.Hour)} + tunnelDNSCache.mu.Unlock() + dgram = append([]byte{0x00, 0x00, 0x00, 0x03, byte(len(cached))}, cached...) + dgram = append(dgram, 0x27, 0x0f, 'p', 'i', 'n', 'g') + rcv := make([]byte, 64) + var nr int + var rerr error + for attempt := 0; attempt < 3; attempt++ { + if _, err := udp.Write(dgram); err != nil { + t.Fatal(err) + } + in.SetReadDeadline(time.Now().Add(3 * time.Second)) + nr, _, rerr = in.ReadFrom(rcv) + if rerr == nil { + break + } + } + if rerr != nil || string(rcv[:nr]) != "ping" { + t.Fatalf("cached-name datagram behind the flood: payload = %q, err = %v", rcv[:nr], rerr) + } +} + +// TestResolveTunnelViaWaitsForLookupSlot ensures a cache miss opens no tunnel +// endpoint (the nil Stack would panic) while every lookup slot is taken. +func TestResolveTunnelViaWaitsForLookupSlot(t *testing.T) { + resetTunnelDNSCacheForTest() + for i := 0; i < cap(tunnelDNSSlots); i++ { + tunnelDNSSlots <- struct{}{} + } + t.Cleanup(func() { + for i := 0; i < cap(tunnelDNSSlots); i++ { + <-tunnelDNSSlots + } + }) + + dev := &Device{localAddrs: []netip.Addr{netip.MustParseAddr(tunnelTestClientAddr)}} + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if _, err := resolveTunnelVia(ctx, dev, "awg-dom-test", "", "slots.internal"); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want context.DeadlineExceeded", err) + } +} + func TestDefaultDNSFor(t *testing.T) { v4 := netip.MustParseAddr("10.8.0.2") v6 := netip.MustParseAddr("2001:db8::2")