transport,grpc: Integrate delegating resolver and introduce dial options for target host resolution

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

@@ -34,6 +34,7 @@ const proxyOptionsKey = keyType("grpc.resolver.delegatingresolver.proxyOptions")
 // handshake.
 type Options struct {
 	User        url.Userinfo
+	UserSet     bool
 	ConnectAddr string
 }
 

internal/proxyattributes/proxyattributes_test.go

@@ -43,6 +43,7 @@ func (s) TestGet(t *testing.T) {
 		addr            resolver.Address
 		wantConnectAddr string
 		wantUser        url.Userinfo
+		wantUserSet     bool
 		wantAttrPresent bool
 	}{
 		{
@@ -61,10 +62,12 @@ func (s) TestGet(t *testing.T) {
 			addr: resolver.Address{
 				Addr: "test-address",
 				Attributes: attributes.New(proxyOptionsKey, Options{
-					User: *user,
+					User:    *user,
+					UserSet: true,
 				}),
 			},
 			wantUser:        *user,
+			wantUserSet:     true,
 			wantAttrPresent: true,
 		},
 		{
@@ -88,6 +91,9 @@ func (s) TestGet(t *testing.T) {
 			if gotOption.User != tt.wantUser {
 				t.Errorf("User(%v) = %v, want %v", tt.addr, gotOption.User, tt.wantUser)
 			}
+			if gotOption.UserSet != tt.wantUserSet {
+				t.Errorf("UserSet(%v) = %v, want %v", tt.addr, gotOption.UserSet, tt.wantUserSet)
+			}
 		})
 	}
 }
@@ -98,6 +104,7 @@ func (s) TestSet(t *testing.T) {
 	addr := resolver.Address{Addr: "test-address"}
 	pOpts := Options{
 		User:        *url.UserPassword("username", "password"),
+		UserSet:     true,
 		ConnectAddr: "proxy-address",
 	}
 
@@ -108,9 +115,12 @@ 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)
 	}
+	if got, want := gotOption.UserSet, pOpts.UserSet; got != want {
+		t.Errorf("unexpected UserSet proxy attribute = %v, want %v", got, want)
+	}
 }

internal/resolver/delegatingresolver/delegatingresolver.go

@@ -206,12 +206,15 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 	}
 	var addresses []resolver.Address
 	var user url.Userinfo
+	var userSet bool
 	if r.proxyURL.User != nil {
 		user = *r.proxyURL.User
+		userSet = true
 	}
 	for _, targetAddr := range (*r.targetResolverState).Addresses {
 		addresses = append(addresses, proxyattributes.Set(proxyAddr, proxyattributes.Options{
 			User:        user,
+			UserSet:     userSet,
 			ConnectAddr: targetAddr.Addr,
 		}))
 	}
@@ -230,6 +233,7 @@ func (r *delegatingResolver) updateClientConnStateLocked() error {
 			for _, targetAddr := range endpt.Addresses {
 				addrs = append(addrs, proxyattributes.Set(proxyAddr, proxyattributes.Options{
 					User:        user,
+					UserSet:     userSet,
 					ConnectAddr: targetAddr.Addr,
 				}))
 			}

internal/testutils/proxyserver/proxyserver.go

@@ -0,0 +1,135 @@
+/*
+ *
+ * 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"
+	"fmt"
+	"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)
+}
+
+// HTTPProxy initializes and starts a proxy server, registers a cleanup to
+// stop it, and returns a ProxyServer.
+func HTTPProxy(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:      fmt.Sprintf("localhost:%d", testutils.ParsePort(t, 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)