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