transport,grpc: Integrate delegating resolver and introduce dial options for target host resolution (#7881)

* Change proxy behaviour

Author: eshitachandwani

clientconn.go

@@ -225,7 +225,12 @@ func Dial(target string, opts ...DialOption) (*ClientConn, error) {
 func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
 	// At the end of this method, we kick the channel out of idle, rather than
 	// waiting for the first rpc.
-	opts = append([]DialOption{withDefaultScheme("passthrough")}, opts...)
+	//
+	// WithLocalDNSResolution dial option in `grpc.Dial` ensures that it
+	// preserves behavior: when default scheme passthrough is used, skip
+	// hostname resolution, when "dns" is used for resolution, perform
+	// resolution on the client.
+	opts = append([]DialOption{withDefaultScheme("passthrough"), WithLocalDNSResolution()}, opts...)
 	cc, err := NewClient(target, opts...)
 	if err != nil {
 		return nil, err

dialoptions.go

@@ -94,6 +94,8 @@ type dialOptions struct {
 	idleTimeout                 time.Duration
 	defaultScheme               string
 	maxCallAttempts             int
+	enableLocalDNSResolution    bool // Specifies if target hostnames should be resolved when proxying is enabled.
+	useProxy                    bool // Specifies if a server should be connected via proxy.
 }
 
 // DialOption configures how we set up the connection.
@@ -377,7 +379,22 @@ func WithInsecure() DialOption {
 // later release.
 func WithNoProxy() DialOption {
 	return newFuncDialOption(func(o *dialOptions) {
-		o.copts.UseProxy = false
+		o.useProxy = false
+	})
+}
+
+// WithLocalDNSResolution forces local DNS name resolution even when a proxy is
+// specified in the environment.  By default, the server name is provided
+// directly to the proxy as part of the CONNECT handshake. This is ignored if
+// WithNoProxy is used.
+//
+// # Experimental
+//
+// Notice: This API is EXPERIMENTAL and may be changed or removed in a
+// later release.
+func WithLocalDNSResolution() DialOption {
+	return newFuncDialOption(func(o *dialOptions) {
+		o.enableLocalDNSResolution = true
 	})
 }
 
@@ -667,14 +684,15 @@ func defaultDialOptions() dialOptions {
 		copts: transport.ConnectOptions{
 			ReadBufferSize:  defaultReadBufSize,
 			WriteBufferSize: defaultWriteBufSize,
-			UseProxy:        true,
 			UserAgent:       grpcUA,
 			BufferPool:      mem.DefaultBufferPool(),
 		},
-		bs:              internalbackoff.DefaultExponential,
-		idleTimeout:     30 * time.Minute,
-		defaultScheme:   "dns",
-		maxCallAttempts: defaultMaxCallAttempts,
+		bs:                       internalbackoff.DefaultExponential,
+		idleTimeout:              30 * time.Minute,
+		defaultScheme:            "dns",
+		maxCallAttempts:          defaultMaxCallAttempts,
+		useProxy:                 true,
+		enableLocalDNSResolution: false,
 	}
 }
 

internal/proxyattributes/proxyattributes.go

@@ -33,7 +33,7 @@ const proxyOptionsKey = keyType("grpc.resolver.delegatingresolver.proxyOptions")
 // Options holds the proxy connection details needed during the CONNECT
 // handshake.
 type Options struct {
-	User        url.Userinfo
+	User        *url.Userinfo
 	ConnectAddr string
 }
 
@@ -44,7 +44,8 @@ func Set(addr resolver.Address, opts Options) resolver.Address {
 }
 
 // Get returns the Options for the proxy [resolver.Address] and a boolean
-// value representing if the attribute is present or not.
+// value representing if the attribute is present or not. The returned data
+// should not be mutated.
 func Get(addr resolver.Address) (Options, bool) {
 	if a := addr.Attributes.Value(proxyOptionsKey); a != nil {
 		return a.(Options), true

internal/proxyattributes/proxyattributes_test.go

@@ -42,7 +42,7 @@ func (s) TestGet(t *testing.T) {
 		name            string
 		addr            resolver.Address
 		wantConnectAddr string
-		wantUser        url.Userinfo
+		wantUser        *url.Userinfo
 		wantAttrPresent bool
 	}{
 		{
@@ -61,10 +61,10 @@ func (s) TestGet(t *testing.T) {
 			addr: resolver.Address{
 				Addr: "test-address",
 				Attributes: attributes.New(proxyOptionsKey, Options{
-					User: *user,
+					User: user,
 				}),
 			},
-			wantUser:        *user,
+			wantUser:        user,
 			wantAttrPresent: true,
 		},
 		{
@@ -97,7 +97,7 @@ func (s) TestGet(t *testing.T) {
 func (s) TestSet(t *testing.T) {
 	addr := resolver.Address{Addr: "test-address"}
 	pOpts := Options{
-		User:        *url.UserPassword("username", "password"),
+		User:        url.UserPassword("username", "password"),
 		ConnectAddr: "proxy-address",
 	}
 
@@ -108,7 +108,7 @@ func (s) TestSet(t *testing.T) {
 		t.Errorf("Get(%v) = %v, want %v ", populatedAddr, attrPresent, true)
 	}
 	if got, want := gotOption.ConnectAddr, pOpts.ConnectAddr; got != want {
-		t.Errorf("Unexpected ConnectAddr proxy atrribute = %v, want %v", got, want)
+		t.Errorf("unexpected ConnectAddr proxy atrribute = %v, want %v", got, want)
 	}
 	if got, want := gotOption.User, pOpts.User; got != want {
 		t.Errorf("unexpected User proxy attribute = %v, want %v", got, want)

internal/resolver/delegatingresolver/delegatingresolver.go

@@ -205,13 +205,9 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 		proxyAddr = resolver.Address{Addr: r.proxyURL.Host}
 	}
 	var addresses []resolver.Address
-	var user url.Userinfo
-	if r.proxyURL.User != nil {
-		user = *r.proxyURL.User
-	}
 	for _, targetAddr := range (*r.targetResolverState).Addresses {
 		addresses = append(addresses, proxyattributes.Set(proxyAddr, proxyattributes.Options{
-			User:        user,
+			User:        r.proxyURL.User,
 			ConnectAddr: targetAddr.Addr,
 		}))
 	}
@@ -229,7 +225,7 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 		for _, proxyAddr := range r.proxyAddrs {
 			for _, targetAddr := range endpt.Addresses {
 				addrs = append(addrs, proxyattributes.Set(proxyAddr, proxyattributes.Options{
-					User:        user,
+					User:        r.proxyURL.User,
 					ConnectAddr: targetAddr.Addr,
 				}))
 			}

internal/testutils/proxyserver/proxyserver.go

@@ -0,0 +1,134 @@
+/*
+ *
+ * Copyright 2024 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+// Package proxyserver provides an implementation of a proxy server for testing purposes.
+// The server supports only a single incoming connection at a time and is not concurrent.
+// It handles only HTTP CONNECT requests; other HTTP methods are not supported.
+package proxyserver
+
+import (
+	"bufio"
+	"bytes"
+	"io"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc/internal/testutils"
+)
+
+// ProxyServer represents a test proxy server.
+type ProxyServer struct {
+	lis       net.Listener
+	in        net.Conn            // Connection from the client to the proxy.
+	out       net.Conn            // Connection from the proxy to the backend.
+	onRequest func(*http.Request) // Function to check the request sent to proxy.
+	Addr      string              // Address of the proxy
+}
+
+const defaultTestTimeout = 10 * time.Second
+
+// Stop closes the ProxyServer and its connections to client and server.
+func (p *ProxyServer) stop() {
+	p.lis.Close()
+	if p.in != nil {
+		p.in.Close()
+	}
+	if p.out != nil {
+		p.out.Close()
+	}
+}
+
+func (p *ProxyServer) handleRequest(t *testing.T, in net.Conn, waitForServerHello bool) {
+	req, err := http.ReadRequest(bufio.NewReader(in))
+	if err != nil {
+		t.Errorf("failed to read CONNECT req: %v", err)
+		return
+	}
+	if req.Method != http.MethodConnect {
+		t.Errorf("unexpected Method %q, want %q", req.Method, http.MethodConnect)
+	}
+	p.onRequest(req)
+
+	t.Logf("Dialing to %s", req.URL.Host)
+	out, err := net.Dial("tcp", req.URL.Host)
+	if err != nil {
+		in.Close()
+		t.Logf("failed to dial to server: %v", err)
+		return
+	}
+	out.SetDeadline(time.Now().Add(defaultTestTimeout))
+	resp := http.Response{StatusCode: http.StatusOK, Proto: "HTTP/1.0"}
+	var buf bytes.Buffer
+	resp.Write(&buf)
+
+	if waitForServerHello {
+		// Batch the first message from the server with the http connect
+		// response. This is done to test the cases in which the grpc client has
+		// the response to the connect request and proxied packets from the
+		// destination server when it reads the transport.
+		b := make([]byte, 50)
+		bytesRead, err := out.Read(b)
+		if err != nil {
+			t.Errorf("Got error while reading server hello: %v", err)
+			in.Close()
+			out.Close()
+			return
+		}
+		buf.Write(b[0:bytesRead])
+	}
+	p.in = in
+	p.in.Write(buf.Bytes())
+	p.out = out
+
+	go io.Copy(p.in, p.out)
+	go io.Copy(p.out, p.in)
+}
+
+// New initializes and starts a proxy server, registers a cleanup to
+// stop it, and returns a ProxyServer.
+func New(t *testing.T, reqCheck func(*http.Request), waitForServerHello bool) *ProxyServer {
+	t.Helper()
+	pLis, err := testutils.LocalTCPListener()
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+
+	p := &ProxyServer{
+		lis:       pLis,
+		onRequest: reqCheck,
+		Addr:      pLis.Addr().String(),
+	}
+
+	// Start the proxy server.
+	go func() {
+		for {
+			in, err := p.lis.Accept()
+			if err != nil {
+				return
+			}
+			// p.handleRequest is not invoked in a goroutine because the test
+			// proxy currently supports handling only one connection at a time.
+			p.handleRequest(t, in, waitForServerHello)
+		}
+	}()
+	t.Logf("Started proxy at: %q", pLis.Addr().String())
+	t.Cleanup(p.stop)
+	return p
+}

internal/transport/http2_client.go

@@ -43,6 +43,7 @@ import (
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/grpcutil"
 	imetadata "google.golang.org/grpc/internal/metadata"
+	"google.golang.org/grpc/internal/proxyattributes"
 	istatus "google.golang.org/grpc/internal/status"
 	isyscall "google.golang.org/grpc/internal/syscall"
 	"google.golang.org/grpc/internal/transport/networktype"
@@ -153,7 +154,7 @@ type http2Client struct {
 	logger       *grpclog.PrefixLogger
 }
 
-func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error), addr resolver.Address, useProxy bool, grpcUA string) (net.Conn, error) {
+func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error), addr resolver.Address, grpcUA string) (net.Conn, error) {
 	address := addr.Addr
 	networkType, ok := networktype.Get(addr)
 	if fn != nil {
@@ -177,8 +178,8 @@ func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error
 	if !ok {
 		networkType, address = parseDialTarget(address)
 	}
-	if networkType == "tcp" && useProxy {
-		return proxyDial(ctx, address, grpcUA)
+	if opts, present := proxyattributes.Get(addr); present {
+		return proxyDial(ctx, addr, grpcUA, opts)
 	}
 	return internal.NetDialerWithTCPKeepalive().DialContext(ctx, networkType, address)
 }
@@ -217,7 +218,7 @@ func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 	// address specific arbitrary data to reach custom dialers and credential handshakers.
 	connectCtx = icredentials.NewClientHandshakeInfoContext(connectCtx, credentials.ClientHandshakeInfo{Attributes: addr.Attributes})
 
-	conn, err := dial(connectCtx, opts.Dialer, addr, opts.UseProxy, opts.UserAgent)
+	conn, err := dial(connectCtx, opts.Dialer, addr, opts.UserAgent)
 	if err != nil {
 		if opts.FailOnNonTempDialError {
 			return nil, connectionErrorf(isTemporary(err), err, "transport: error while dialing: %v", err)