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

@@ -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 {

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()

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")