Pull request 2159: Upd proxy

Squashed commit of the following:

commit 4468e82
Author: Eugene Burkov <E.Burkov@AdGuard.COM>
Date:   Mon Mar 11 13:50:36 2024 +0300

    all: upd dnsproxy

commit 7887f52
Merge: 9120da6 36f9fec
Author: Eugene Burkov <E.Burkov@AdGuard.COM>
Date:   Mon Mar 11 13:50:09 2024 +0300

    Merge branch 'master' into upd-proxy

commit 9120da6
Author: Eugene Burkov <E.Burkov@AdGuard.COM>
Date:   Tue Mar 5 11:50:16 2024 +0300

    all: upd proxy

go.mod

@@ -3,7 +3,7 @@ module github.com/AdguardTeam/AdGuardHome
 go 1.21.8
 
 require (
-	github.com/AdguardTeam/dnsproxy v0.65.2
+	github.com/AdguardTeam/dnsproxy v0.66.0
 	github.com/AdguardTeam/golibs v0.20.1
 	github.com/AdguardTeam/urlfilter v0.18.0
 	github.com/NYTimes/gziphandler v1.1.1

go.sum

@@ -1,5 +1,5 @@
-github.com/AdguardTeam/dnsproxy v0.65.2 h1:D+BMw0Vu2lbQrYpoPctG2Xr+24KdfhgkzZb6QgPZheM=
-github.com/AdguardTeam/dnsproxy v0.65.2/go.mod h1:8NQTTNZY+qR9O1Fzgz3WQv30knfSgms68SRlzSnX74A=
+github.com/AdguardTeam/dnsproxy v0.66.0 h1:RyUbyDxRSXBFjVG1l2/4HV3I98DtfIgpnZkgXkgHKnc=
+github.com/AdguardTeam/dnsproxy v0.66.0/go.mod h1:ZThEXbMUlP1RxfwtNW30ItPAHE6OF4YFygK8qjU/cvY=
 github.com/AdguardTeam/golibs v0.20.1 h1:ol8qLjWGZhU9paMMwN+OLWVTUigGsXa29iVTyd62VKY=
 github.com/AdguardTeam/golibs v0.20.1/go.mod h1:bgcMgRviCKyU6mkrX+RtT/OsKPFzyppelfRsksMG3KU=
 github.com/AdguardTeam/urlfilter v0.18.0 h1:ZZzwODC/ADpjJSODxySrrUnt/fvOCfGFaCW6j+wsGfQ=

internal/aghtest/aghtest.go

@@ -9,8 +9,13 @@ import (
 	"net/netip"
 	"net/url"
 	"testing"
+	"time"
 
+	"github.com/AdguardTeam/dnsproxy/proxy"
 	"github.com/AdguardTeam/golibs/log"
+	"github.com/AdguardTeam/golibs/netutil"
+	"github.com/AdguardTeam/golibs/testutil"
+	"github.com/miekg/dns"
 	"github.com/stretchr/testify/require"
 )
 
@@ -71,3 +76,49 @@ func StartHTTPServer(t testing.TB, data []byte) (c *http.Client, u *url.URL) {
 
 	return srv.Client(), u
 }
+
+// testTimeout is a timeout for tests.
+//
+// TODO(e.burkov):  Move into agdctest.
+const testTimeout = 1 * time.Second
+
+// StartLocalhostUpstream is a test helper that starts a DNS server on
+// localhost.
+func StartLocalhostUpstream(t *testing.T, h dns.Handler) (addr *url.URL) {
+	t.Helper()
+
+	startCh := make(chan netip.AddrPort)
+	defer close(startCh)
+	errCh := make(chan error)
+
+	srv := &dns.Server{
+		Addr:         "127.0.0.1:0",
+		Net:          string(proxy.ProtoTCP),
+		Handler:      h,
+		ReadTimeout:  testTimeout,
+		WriteTimeout: testTimeout,
+	}
+	srv.NotifyStartedFunc = func() {
+		addrPort := srv.Listener.Addr()
+		startCh <- netutil.NetAddrToAddrPort(addrPort)
+	}
+
+	go func() { errCh <- srv.ListenAndServe() }()
+
+	select {
+	case addrPort := <-startCh:
+		addr = &url.URL{
+			Scheme: string(proxy.ProtoTCP),
+			Host:   addrPort.String(),
+		}
+
+		testutil.CleanupAndRequireSuccess(t, func() (err error) { return <-errCh })
+		testutil.CleanupAndRequireSuccess(t, srv.Shutdown)
+	case err := <-errCh:
+		require.NoError(t, err)
+	case <-time.After(testTimeout):
+		require.FailNow(t, "timeout exceeded")
+	}
+
+	return addr
+}

internal/dnsforward/config.go

@@ -357,10 +357,6 @@ func (s *Server) newProxyConfig() (conf *proxy.Config, err error) {
 		conf.DNSCryptResolverCert = c.ResolverCert
 	}
 
-	if conf.UpstreamConfig == nil || len(conf.UpstreamConfig.Upstreams) == 0 {
-		return nil, errors.Error("no default upstream servers configured")
-	}
-
 	conf, err = prepareCacheConfig(conf,
 		srvConf.CacheSize,
 		srvConf.CacheMinTTL,

internal/dnsforward/dns64_test.go

@@ -8,7 +8,6 @@ import (
 	"github.com/AdguardTeam/AdGuardHome/internal/aghtest"
 	"github.com/AdguardTeam/AdGuardHome/internal/filtering"
 	"github.com/AdguardTeam/dnsproxy/proxy"
-	"github.com/AdguardTeam/dnsproxy/upstream"
 	"github.com/AdguardTeam/golibs/netutil"
 	"github.com/AdguardTeam/golibs/testutil"
 	"github.com/miekg/dns"
@@ -101,21 +100,6 @@ func TestServer_HandleDNSRequest_dns64(t *testing.T) {
 	type answerMap = map[uint16][sectionsNum][]dns.RR
 
 	pt := testutil.PanicT{}
-	newUps := func(answers answerMap) (u upstream.Upstream) {
-		return aghtest.NewUpstreamMock(func(req *dns.Msg) (resp *dns.Msg, err error) {
-			q := req.Question[0]
-			require.Contains(pt, answers, q.Qtype)
-
-			answer := answers[q.Qtype]
-
-			resp = (&dns.Msg{}).SetReply(req)
-			resp.Answer = answer[sectionAnswer]
-			resp.Ns = answer[sectionAuthority]
-			resp.Extra = answer[sectionAdditional]
-
-			return resp, nil
-		})
-	}
 
 	testCases := []struct {
 		name    string
@@ -265,39 +249,61 @@ func TestServer_HandleDNSRequest_dns64(t *testing.T) {
 	}}
 
 	localRR := newRR(t, ptr64Domain, dns.TypePTR, 3600, pointedDomain)
-	localUps := aghtest.NewUpstreamMock(func(req *dns.Msg) (resp *dns.Msg, err error) {
-		require.Equal(pt, req.Question[0].Name, ptr64Domain)
-		resp = (&dns.Msg{}).SetReply(req)
-		resp.Answer = []dns.RR{localRR}
+	localUpsHdlr := dns.HandlerFunc(func(w dns.ResponseWriter, m *dns.Msg) {
+		require.Len(pt, m.Question, 1)
+		require.Equal(pt, m.Question[0].Name, ptr64Domain)
+		resp := (&dns.Msg{
+			Answer: []dns.RR{localRR},
+		}).SetReply(m)
 
-		return resp, nil
+		require.NoError(t, w.WriteMsg(resp))
 	})
+	localUpsAddr := aghtest.StartLocalhostUpstream(t, localUpsHdlr).String()
 
 	client := &dns.Client{
 		Net:     "tcp",
 		Timeout: 1 * time.Second,
 	}
 
 	for _, tc := range testCases {
-		// TODO(e.burkov):  It seems [proxy.Proxy] isn't intended to be reused
-		// right after stop, due to a data race in [proxy.Proxy.Init] method
-		// when setting an OOB size.  As a temporary workaround, recreate the
-		// whole server for each test case.
-		s := createTestServer(t, &filtering.Config{
-			BlockingMode: filtering.BlockingModeDefault,
-		}, ServerConfig{
-			UDPListenAddrs: []*net.UDPAddr{{}},
-			TCPListenAddrs: []*net.TCPAddr{{}},
-			UseDNS64:       true,
-			Config: Config{
-				UpstreamMode:     UpstreamModeLoadBalance,
-				EDNSClientSubnet: &EDNSClientSubnet{Enabled: false},
-			},
-			ServePlainDNS: true,
-		}, localUps)
-
+		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
-			s.conf.UpstreamConfig.Upstreams = []upstream.Upstream{newUps(tc.upsAns)}
+			upsHdlr := dns.HandlerFunc(func(w dns.ResponseWriter, req *dns.Msg) {
+				q := req.Question[0]
+				require.Contains(pt, tc.upsAns, q.Qtype)
+
+				answer := tc.upsAns[q.Qtype]
+
+				resp := (&dns.Msg{
+					Answer: answer[sectionAnswer],
+					Ns:     answer[sectionAuthority],
+					Extra:  answer[sectionAdditional],
+				}).SetReply(req)
+
+				require.NoError(pt, w.WriteMsg(resp))
+			})
+			upsAddr := aghtest.StartLocalhostUpstream(t, upsHdlr).String()
+
+			// TODO(e.burkov):  It seems [proxy.Proxy] isn't intended to be
+			// reused right after stop, due to a data race in [proxy.Proxy.Init]
+			// method when setting an OOB size.  As a temporary workaround,
+			// recreate the whole server for each test case.
+			s := createTestServer(t, &filtering.Config{
+				BlockingMode: filtering.BlockingModeDefault,
+			}, ServerConfig{
+				UDPListenAddrs: []*net.UDPAddr{{}},
+				TCPListenAddrs: []*net.TCPAddr{{}},
+				UseDNS64:       true,
+				Config: Config{
+					UpstreamMode:     UpstreamModeLoadBalance,
+					EDNSClientSubnet: &EDNSClientSubnet{Enabled: false},
+					UpstreamDNS:      []string{upsAddr},
+				},
+				UsePrivateRDNS:    true,
+				LocalPTRResolvers: []string{localUpsAddr},
+				ServePlainDNS:     true,
+			})
+
 			startDeferStop(t, s)
 
 			req := (&dns.Msg{}).SetQuestion(tc.qname, tc.qtype)

internal/dnsforward/dnsforward.go

@@ -464,7 +464,8 @@ func (s *Server) Start() error {
 // startLocked starts the DNS server without locking.  s.serverLock is expected
 // to be locked.
 func (s *Server) startLocked() error {
-	err := s.dnsProxy.Start()
+	// TODO(e.burkov):  Use context properly.
+	err := s.dnsProxy.Start(context.Background())
 	if err == nil {
 		s.isRunning = true
 	}
@@ -518,34 +519,30 @@ func (s *Server) prepareLocalResolvers(
 }
 
 // setupLocalResolvers initializes and sets the resolvers for local addresses.
-// It assumes s.serverLock is locked or s not running.
-func (s *Server) setupLocalResolvers(boot upstream.Resolver) (err error) {
-	uc, err := s.prepareLocalResolvers(boot)
-	if err != nil {
-		// Don't wrap the error because it's informative enough as is.
-		return err
+// It assumes s.serverLock is locked or s not running.  It returns the upstream
+// configuration used for private PTR resolving, or nil if it's disabled.  Note,
+// that it's safe to put nil into [proxy.Config.PrivateRDNSUpstreamConfig].
+func (s *Server) setupLocalResolvers(boot upstream.Resolver) (uc *proxy.UpstreamConfig, err error) {
+	if !s.conf.UsePrivateRDNS {
+		// It's safe to put nil into [proxy.Config.PrivateRDNSUpstreamConfig].
+		return nil, nil
 	}
 
-	s.localResolvers = &proxy.Proxy{
-		Config: proxy.Config{
-			UpstreamConfig: uc,
-		},
+	uc, err = s.prepareLocalResolvers(boot)
+	if err != nil {
+		// Don't wrap the error because it's informative enough as is.
+		return nil, err
 	}
 
-	err = s.localResolvers.Init()
+	s.localResolvers, err = proxy.New(&proxy.Config{
+		UpstreamConfig: uc,
+	})
 	if err != nil {
-		return fmt.Errorf("initializing proxy: %w", err)
+		return nil, fmt.Errorf("creating local resolvers: %w", err)
 	}
 
 	// TODO(e.burkov):  Should we also consider the DNS64 usage?
-	if s.conf.UsePrivateRDNS &&
-		// Only set the upstream config if there are any upstreams.  It's safe
-		// to put nil into [proxy.Config.PrivateRDNSUpstreamConfig].
-		len(uc.Upstreams)+len(uc.DomainReservedUpstreams)+len(uc.SpecifiedDomainUpstreams) > 0 {
-		s.dnsProxy.PrivateRDNSUpstreamConfig = uc
-	}
-
-	return nil
+	return uc, nil
 }
 
 // Prepare initializes parameters of s using data from conf.  conf must not be
@@ -586,21 +583,22 @@ func (s *Server) Prepare(conf *ServerConfig) (err error) {
 		return fmt.Errorf("preparing access: %w", err)
 	}
 
-	// Set the proxy here because [setupLocalResolvers] sets its values.
-	//
 	// TODO(e.burkov):  Remove once the local resolvers logic moved to dnsproxy.
-	s.dnsProxy = &proxy.Proxy{Config: *proxyConfig}
-
-	err = s.setupLocalResolvers(boot)
+	proxyConfig.PrivateRDNSUpstreamConfig, err = s.setupLocalResolvers(boot)
 	if err != nil {
 		return fmt.Errorf("setting up resolvers: %w", err)
 	}
 
-	err = s.setupFallbackDNS()
+	proxyConfig.Fallbacks, err = s.setupFallbackDNS()
 	if err != nil {
 		return fmt.Errorf("setting up fallback dns servers: %w", err)
 	}
 
+	s.dnsProxy, err = proxy.New(proxyConfig)
+	if err != nil {
+		return fmt.Errorf("creating proxy: %w", err)
+	}
+
 	s.recDetector.clear()
 
 	s.setupAddrProc()
@@ -643,26 +641,25 @@ func (s *Server) prepareInternalDNS() (boot upstream.Resolver, err error) {
 }
 
 // setupFallbackDNS initializes the fallback DNS servers.
-func (s *Server) setupFallbackDNS() (err error) {
+func (s *Server) setupFallbackDNS() (uc *proxy.UpstreamConfig, err error) {
 	fallbacks := s.conf.FallbackDNS
 	fallbacks = stringutil.FilterOut(fallbacks, IsCommentOrEmpty)
 	if len(fallbacks) == 0 {
-		return nil
+		return nil, nil
 	}
 
-	uc, err := proxy.ParseUpstreamsConfig(fallbacks, &upstream.Options{
+	uc, err = proxy.ParseUpstreamsConfig(fallbacks, &upstream.Options{
 		// TODO(s.chzhen):  Investigate if other options are needed.
 		Timeout:    s.conf.UpstreamTimeout,
 		PreferIPv6: s.conf.BootstrapPreferIPv6,
+		// TODO(e.burkov):  Use bootstrap.
 	})
 	if err != nil {
 		// Do not wrap the error because it's informative enough as is.
-		return err
+		return nil, err
 	}
 
-	s.dnsProxy.Fallbacks = uc
-
-	return nil
+	return uc, nil
 }
 
 // setupAddrProc initializes the address processor.  It assumes s.serverLock is
@@ -730,19 +727,9 @@ func (s *Server) prepareInternalProxy() (err error) {
 		return fmt.Errorf("invalid upstream mode: %w", err)
 	}
 
-	// TODO(a.garipov): Make a proper constructor for proxy.Proxy.
-	p := &proxy.Proxy{
-		Config: *conf,
-	}
-
-	err = p.Init()
-	if err != nil {
-		return err
-	}
-
-	s.internalProxy = p
+	s.internalProxy, err = proxy.New(conf)
 
-	return nil
+	return err
 }
 
 // Stop stops the DNS server.
@@ -761,14 +748,17 @@ func (s *Server) stopLocked() (err error) {
 	// [upstream.Upstream] implementations.
 
 	if s.dnsProxy != nil {
-		err = s.dnsProxy.Stop()
+		// TODO(e.burkov):  Use context properly.
+		err = s.dnsProxy.Shutdown(context.Background())
 		if err != nil {
 			log.Error("dnsforward: closing primary resolvers: %s", err)
 		}
 	}
 
 	logCloserErr(s.internalProxy.UpstreamConfig, "dnsforward: closing internal resolvers: %s")
-	logCloserErr(s.localResolvers.UpstreamConfig, "dnsforward: closing local resolvers: %s")
+	if s.localResolvers != nil {
+		logCloserErr(s.localResolvers.UpstreamConfig, "dnsforward: closing local resolvers: %s")
+	}
 
 	for _, b := range s.bootResolvers {
 		logCloserErr(b, "dnsforward: closing bootstrap %s: %s", b.Address())