xds: implement :authority header rewriting in xds_cluster_impl LB policy (gRFC A81)

internal/testutils/xds/e2e/clientresources.go

@@ -650,6 +650,8 @@ type BackendOptions struct {
 	HealthStatus v3corepb.HealthStatus
 	// Weight sets the backend weight. Defaults to 1.
 	Weight uint32
+	// Hostname sets the endpoint hostname for authority rewriting.
+	Hostname string
 	// Metadata sets the LB endpoint metadata (envoy.lb FilterMetadata field).
 	// See https://www.envoyproxy.io/docs/envoy/latest/api-v3/config/core/v3/base.proto#envoy-v3-api-msg-config-core-v3-metadata
 	Metadata map[string]any
@@ -721,6 +723,7 @@ func EndpointResourceWithOptions(opts EndpointOptions) *v3endpointpb.ClusterLoad
 							PortSpecifier: &v3corepb.SocketAddress_PortValue{PortValue: b.Ports[0]},
 						},
 					}},
+					Hostname:            b.Hostname,
 					AdditionalAddresses: additionalAddresses,
 				}},
 				HealthStatus:        b.HealthStatus,

internal/xds/balancer/clusterimpl/clusterimpl.go

@@ -45,6 +45,7 @@ import (
 	"google.golang.org/grpc/internal/xds/clients"
 	"google.golang.org/grpc/internal/xds/clients/lrsclient"
 	"google.golang.org/grpc/internal/xds/xdsclient"
+	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/serviceconfig"
 )
@@ -420,6 +421,8 @@ type scWrapper struct {
 	// locality needs to be atomic because it can be updated while being read by
 	// the picker.
 	locality atomic.Pointer[clients.Locality]
+
+	hostname string
 }
 
 func (scw *scWrapper) updateLocalityID(lID clients.Locality) {
@@ -442,6 +445,9 @@ func (b *clusterImplBalancer) NewSubConn(addrs []resolver.Address, opts balancer
 	}
 	var sc balancer.SubConn
 	scw := &scWrapper{}
+	if len(addrs) > 0 {
+		scw.hostname = xdsresource.HostnameFromAddress(addrs[0])
+	}
 	oldListener := opts.StateListener
 	opts.StateListener = func(state balancer.SubConnState) {
 		b.updateSubConnState(sc, state, oldListener)

internal/xds/balancer/clusterimpl/picker.go

@@ -31,6 +31,7 @@ import (
 	xdsinternal "google.golang.org/grpc/internal/xds"
 	"google.golang.org/grpc/internal/xds/clients"
 	"google.golang.org/grpc/internal/xds/xdsclient"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 )
 
@@ -145,6 +146,14 @@ func (d *picker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
 		// If locality ID isn't found in the wrapper, an empty locality ID will
 		// be used.
 		lID = scw.localityID()
+
+		if scw.hostname != "" && autoHostRewrite(info.Ctx) {
+			if pr.Metadata == nil {
+				pr.Metadata = metadata.Pairs(":authority", scw.hostname)
+			} else {
+				pr.Metadata.Set(":authority", scw.hostname)
+			}
+		}
 	}
 
 	if err != nil {

internal/xds/balancer/clusterimpl/tests/balancer_test.go

@@ -20,11 +20,14 @@ package clusterimpl_test
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"math"
 	"net"
+	"os"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -38,19 +41,24 @@ import (
 	"google.golang.org/grpc/balancer/pickfirst"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/balancer/stub"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/internal/testutils/xds/fakeserver"
+	"google.golang.org/grpc/internal/xds/bootstrap"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/serviceconfig"
 	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/testdata"
 	"google.golang.org/protobuf/testing/protocmp"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
@@ -95,7 +103,6 @@ func (s) TestConfigUpdateWithSameLoadReportingServerConfig(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -377,7 +384,6 @@ func (s) TestCircuitBreaking(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -574,7 +580,6 @@ func (s) TestDropByCategory(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -717,7 +722,6 @@ func (s) TestCircuitBreakingLogicalDNS(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -836,7 +840,6 @@ func (s) TestLRSLogicalDNS(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -928,7 +931,6 @@ func (s) TestReResolutionAfterTransientFailure(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -1048,7 +1050,6 @@ func (s) TestUpdateLRSServerToNil(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -1135,7 +1136,6 @@ func (s) TestChildPolicyChangeOnConfigUpdate(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -1258,7 +1258,6 @@ func (s) TestFailedToParseChildPolicyConfig(t *testing.T) {
 	// Create bootstrap configuration pointing to the above management server.
 	nodeID := uuid.New().String()
 	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
-	testutils.CreateBootstrapFileForTesting(t, bc)
 
 	// Create an xDS resolver with the above bootstrap configuration.
 	if internal.NewXDSResolverWithConfigForTesting == nil {
@@ -1316,3 +1315,206 @@ func (s) TestFailedToParseChildPolicyConfig(t *testing.T) {
 		t.Fatal("EmptyCall RPC succeeded when expected to fail")
 	}
 }
+
+// setupManagementServerAndResolver sets up an xDS management server and returns
+// the management server, resolver builder and Node ID.
+func setupManagementServerAndResolver(t *testing.T) (*e2e.ManagementServer, resolver.Builder, string) {
+	t.Helper()
+
+	nodeID := uuid.New().String()
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+	opts := bootstrap.ConfigOptionsForTesting{
+		Servers: []byte(fmt.Sprintf(`[{
+			"server_uri": %q,
+			"channel_creds": [{"type": "insecure"}],
+			"server_features": ["trusted_xds_server"]
+		}]`, mgmtServer.Address)),
+		Node: []byte(fmt.Sprintf(`{"id": "%s"}`, nodeID)),
+	}
+
+	contents, err := bootstrap.NewContentsForTesting(opts)
+	if err != nil {
+		t.Fatalf("Failed to create bootstrap configuration: %v", err)
+	}
+
+	// Create an xDS resolver with the above bootstrap configuration.
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(contents)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+	}
+
+	return mgmtServer, resolverBuilder, nodeID
+}
+
+// configureXDSResources configures the management server with a route that
+// enables auto_host_rewrite and an endpoint with the specified hostname.
+func configureXDSResources(ctx context.Context, t *testing.T, mgmtServer *e2e.ManagementServer, nodeID string, serverAddr string, endpointHostname string) {
+	t.Helper()
+
+	const (
+		serviceName  = "my-test-xds-service"
+		routeName    = "route-my-test-xds-service"
+		clusterName  = "cluster-my-test-xds-service"
+		endpointName = "endpoints-my-test-xds-service"
+	)
+
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: serviceName,
+		NodeID:     nodeID,
+		Host:       "localhost",
+		Port:       testutils.ParsePort(t, serverAddr),
+		SecLevel:   e2e.SecurityLevelNone,
+	})
+
+	// Set the endpoint hostname for authority rewriting.
+	resources.Endpoints[0].Endpoints[0].LbEndpoints[0].GetEndpoint().Hostname = endpointHostname
+
+	// Modify the route to enable AutoHostRewrite.
+	resources.Routes[0].VirtualHosts[0].Routes[0].GetRoute().HostRewriteSpecifier = &v3routepb.RouteAction_AutoHostRewrite{
+		AutoHostRewrite: &wrapperspb.BoolValue{Value: true},
+	}
+
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// TestAuthorityOverriding verifies that the :authority header is correctly
+// rewritten to the endpoint's hostname. Also verifies that CallAuthority
+// call option takes precedence.
+func (s) TestAuthorityOverriding(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSAuthorityRewrite, true)
+	mgmtServer, resolverBuilder, nodeID := setupManagementServerAndResolver(t)
+
+	// Start a server backend exposing the test service.
+	authorityCh := make(chan string, 1)
+	f := &stubserver.StubServer{
+		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+			if md, ok := metadata.FromIncomingContext(ctx); ok {
+				if authVals := md.Get(":authority"); len(authVals) > 0 {
+					authorityCh <- authVals[0]
+				}
+			}
+			return &testpb.Empty{}, nil
+		},
+	}
+	server := stubserver.StartTestService(t, f)
+	defer server.Stop()
+
+	const xdsAuthorityOverride = "rewritten.example.com"
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	configureXDSResources(ctx, t, mgmtServer, nodeID, server.Address, xdsAuthorityOverride)
+
+	// Create a ClientConn and make a successful RPC.
+	cc, err := grpc.NewClient("xds:///my-test-xds-service", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(resolverBuilder))
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+	defer cc.Close()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+		t.Fatalf("client.EmptyCall() failed: %v", err)
+	}
+
+	select {
+	case gotAuth := <-authorityCh:
+		if gotAuth != xdsAuthorityOverride {
+			t.Errorf("invalid authority got: %q, want: %q", gotAuth, xdsAuthorityOverride)
+		}
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for successful RPC after authority rewriting.")
+	}
+
+	// The authority specified via the `CallAuthority` CallOption takes the
+	// highest precedence when determining the `:authority` header.
+	userAuthorityOverride := "user-override.com"
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.CallAuthority(userAuthorityOverride)); err != nil {
+		t.Fatalf("client.EmptyCall() failed: %v", err)
+	}
+
+	select {
+	case got := <-authorityCh:
+		if got != userAuthorityOverride {
+			t.Errorf("Server received authority %q, want %q (user override)", got, userAuthorityOverride)
+		}
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for successful RPC.")
+	}
+}
+
+// TestAuthorityOverridingWithTLS verifies the interaction between xDS Authority
+// Rewriting and TLS Secure Naming. It ensures that when the :authority header
+// is rewritten by the clusterimpl picker, the new authority is correctly
+// validated against the server's TLS certificate before the RPC proceeds.
+func (s) TestAuthorityOverridingWithTLS(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSAuthorityRewrite, true)
+	mgmtServer, resolverBuilder, nodeID := setupManagementServerAndResolver(t)
+
+	serverCert, err := tls.LoadX509KeyPair(testdata.Path("x509/server1_cert.pem"), testdata.Path("x509/server1_key.pem"))
+	if err != nil {
+		t.Fatalf("Failed to load server key pair: %v", err)
+	}
+
+	pemData, err := os.ReadFile(testdata.Path("x509/client_ca_cert.pem"))
+	if err != nil {
+		t.Fatalf("Failed to read client CA cert: %v", err)
+	}
+	roots := x509.NewCertPool()
+	roots.AppendCertsFromPEM(pemData)
+	tlsConfig := &tls.Config{
+		Certificates:       []tls.Certificate{serverCert},
+		ClientCAs:          roots,
+		InsecureSkipVerify: true,
+	}
+	clientCreds := credentials.NewTLS(tlsConfig)
+
+	// Start a server backend exposing the test service.
+	authorityCh := make(chan string, 1)
+	f := &stubserver.StubServer{
+		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+			if md, ok := metadata.FromIncomingContext(ctx); ok {
+				if authVals := md.Get(":authority"); len(authVals) > 0 {
+					authorityCh <- authVals[0]
+				}
+			}
+			return &testpb.Empty{}, nil
+		},
+	}
+	if err := f.StartServer(grpc.Creds(credentials.NewServerTLSFromCert(&serverCert))); err != nil {
+		t.Fatalf("Failed to start the server: %v", err)
+	}
+	defer f.Stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	const xdsAuthorityOverride = "x.test.example.com"
+	configureXDSResources(ctx, t, mgmtServer, nodeID, f.Address, xdsAuthorityOverride)
+
+	// Create ClientConn with TLS
+	cc, err := grpc.NewClient("xds:///my-test-xds-service", grpc.WithTransportCredentials(clientCreds), grpc.WithResolvers(resolverBuilder))
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+	defer cc.Close()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+		t.Fatalf("client.EmptyCall() failed: %v", err)
+	}
+
+	select {
+	case gotAuth := <-authorityCh:
+		if gotAuth != xdsAuthorityOverride {
+			t.Errorf("invalid authority got: %q, want: %q", gotAuth, xdsAuthorityOverride)
+		}
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for successful RPC after authority rewriting.")
+	}
+}