xds/gcp_authn: implement GCP Authentication filter (gRFC A83). (#9119)

This PR implements the xDS GCP Authentication HTTP filter for gRPC-Go,
as specified in [gRFC
A83](https://github.com/grpc/proposal/blob/master/A83-xds-gcp-authn-filter.md).

The implementation provides a mechanism for attaching service account
identity JWT tokens as gRPC call credentials on GCP, featuring a
two-level caching strategy (filter-level cache of credentials objects
and credential-level cache of tokens) to minimize latency and metadata
server load.

RELEASE NOTES:
* xds: add support for the GCP Authentication HTTP Filter. It
automatically fetches and attaches GCP Service Account Identity JWT
tokens to outgoing RPCs. This feature is currently gated behind the
`GRPC_EXPERIMENTAL_XDS_GCP_AUTHENTICATION_FILTER` environment variable.

Author: Pranjali-2501

internal/xds/balancer/clustermanager/picker.go

@@ -43,7 +43,7 @@ func newPickerGroup(idToPickerState map[string]*subBalancerState) *pickerGroup {
 }
 
 func (pg *pickerGroup) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
-	cluster := getPickedCluster(info.Ctx)
+	cluster := PickedCluster(info.Ctx)
 	if p := pg.pickers[cluster]; p != nil {
 		return p.Pick(info)
 	}
@@ -52,17 +52,12 @@ func (pg *pickerGroup) Pick(info balancer.PickInfo) (balancer.PickResult, error)
 
 type clusterKey struct{}
 
-func getPickedCluster(ctx context.Context) string {
+// PickedCluster returns the cluster name stored in the context.
+func PickedCluster(ctx context.Context) string {
 	cluster, _ := ctx.Value(clusterKey{}).(string)
 	return cluster
 }
 
-// GetPickedClusterForTesting returns the cluster in the context; to be used
-// for testing only.
-func GetPickedClusterForTesting(ctx context.Context) string {
-	return getPickedCluster(ctx)
-}
-
 // SetPickedCluster adds the selected cluster to the context for the
 // xds_cluster_manager LB policy to pick.
 func SetPickedCluster(ctx context.Context, cluster string) context.Context {

internal/xds/httpfilter/extproc/config_test.go

@@ -735,7 +735,7 @@ func (s) TestBuildClientInterceptor_Success(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			builder := builder{}
-			filter := builder.BuildClientFilter()
+			filter := builder.BuildClientFilter(httpfilter.ClientFilterOptions{})
 			defer filter.Close()
 
 			intptr, err := filter.BuildClientInterceptor(tc.cfg, tc.override)
@@ -818,7 +818,7 @@ func (s) TestBuildClientInterceptor_Failure(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			builder := builder{}
-			filter := builder.BuildClientFilter()
+			filter := builder.BuildClientFilter(httpfilter.ClientFilterOptions{})
 			defer filter.Close()
 
 			_, err := filter.BuildClientInterceptor(tc.cfg, tc.override)

internal/xds/httpfilter/extproc/ext_proc.go

@@ -190,7 +190,7 @@ func (builder) IsTerminal() bool {
 	return false
 }
 
-func (builder) BuildClientFilter() httpfilter.ClientFilter {
+func (builder) BuildClientFilter(httpfilter.ClientFilterOptions) httpfilter.ClientFilter {
 	return clientFilter{}
 }
 

internal/xds/httpfilter/fault/fault.go

@@ -105,7 +105,7 @@ func (builder) IsTerminal() bool {
 	return false
 }
 
-func (builder) BuildClientFilter() httpfilter.ClientFilter {
+func (builder) BuildClientFilter(httpfilter.ClientFilterOptions) httpfilter.ClientFilter {
 	return clientFilter{}
 }
 

internal/xds/httpfilter/gcp_authn/gcp_authn_filter.go

@@ -0,0 +1,333 @@
+/*
+ *
+ * Copyright 2026 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 gcpauthn implements the GCP Authentication HTTP filter.
+package gcpauthn
+
+import (
+	"container/list"
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+
+	"golang.org/x/sync/singleflight"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/google"
+	"google.golang.org/grpc/internal/resolver"
+	"google.golang.org/grpc/internal/xds/balancer/clustermanager"
+	"google.golang.org/grpc/internal/xds/httpfilter"
+	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+
+	v3gcpauthnpb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/gcp_authn/v3"
+	anypb "google.golang.org/protobuf/types/known/anypb"
+)
+
+const defaultCacheSize = 10 // default capacity of the LRU credentials cache.
+
+func init() {
+	httpfilter.Register(builder{})
+}
+
+type builder struct{}
+
+type config struct {
+	httpfilter.FilterConfig
+	cacheSize uint64
+}
+
+func (builder) TypeURLs() []string {
+	return []string{"type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.GcpAuthnFilterConfig"}
+}
+
+func (builder) ParseFilterConfig(cfg proto.Message) (httpfilter.FilterConfig, error) {
+	m, ok := cfg.(*anypb.Any)
+	if !ok {
+		return nil, fmt.Errorf("gcpauthn: invalid filter config type %T", cfg)
+	}
+	msg := &v3gcpauthnpb.GcpAuthnFilterConfig{}
+	if err := m.UnmarshalTo(msg); err != nil {
+		return nil, fmt.Errorf("gcpauthn: failed to unmarshal filter config: %v", err)
+	}
+
+	cacheSize := uint64(defaultCacheSize)
+	if cacheSizeConfig := msg.GetCacheConfig().GetCacheSize(); cacheSizeConfig != nil {
+		if cacheSize = cacheSizeConfig.GetValue(); cacheSize == 0 {
+			return nil, fmt.Errorf("gcpauthn: cache_config.cache_size must be greater than zero")
+		}
+	}
+
+	return config{cacheSize: cacheSize}, nil
+}
+
+// ParseFilterConfigOverride parses the provided override configuration.
+//
+// Note that we don't support overrides for this filter configuration,
+// but still validate it as part of the normal resource validation.
+func (b builder) ParseFilterConfigOverride(cfg proto.Message) (httpfilter.FilterConfig, error) {
+	return b.ParseFilterConfig(cfg)
+}
+
+func (builder) IsTerminal() bool {
+	return false
+}
+
+func (builder) BuildClientFilter(opts httpfilter.ClientFilterOptions) httpfilter.ClientFilter {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &clientFilter{
+		ctx:        ctx,
+		cancel:     cancel,
+		filterName: opts.FilterName,
+	}
+}
+
+var _ httpfilter.ClientFilterBuilder = builder{}
+
+// clientFilter implements the httpfilter.ClientFilter interface.
+type clientFilter struct {
+	// ctx is initialized using context.Background() and is scoped to the
+	// lifetime of the filter. It is used as the parent context for fetching
+	// service account identity credentials, ensuring that token fetch requests
+	// are not terminated by individual RPC's context.
+	ctx context.Context
+
+	// cancel is the cancellation function for ctx, called when the filter
+	// is closed.
+	cancel context.CancelFunc
+
+	// filterName is the name of the HTTP filter instance in the xDS
+	// configuration. It is used as the key in the cluster metadata to look up
+	// the audience value for the cluster to which the RPC is destined.
+	filterName string
+
+	// cache is the LRU cache of PerRPCCredentials instances, keyed by audience
+	// and is initialized or resized when BuildClientInterceptor is called.
+	cache *lruCache
+}
+
+// BuildClientInterceptor builds a client interceptor for the GCP
+// Authentication filter.
+func (cf *clientFilter) BuildClientInterceptor(cfg, _ httpfilter.FilterConfig) (httpfilter.ClientInterceptor, error) {
+	c, ok := cfg.(config)
+	if !ok {
+		return nil, fmt.Errorf("gcpauthn: invalid filter config type %T", cfg)
+	}
+
+	if cf.cache == nil {
+		cf.cache = newLRUCache(c.cacheSize)
+	} else {
+		cf.cache.resizeCache(c.cacheSize)
+	}
+
+	return &interceptor{
+		ctx:        cf.ctx,
+		filterName: cf.filterName,
+		cache:      cf.cache,
+	}, nil
+}
+
+// Close closes the client filter.
+func (cf *clientFilter) Close() {
+	cf.cancel()
+}
+
+type interceptor struct {
+	ctx        context.Context
+	filterName string
+	cache      *lruCache
+}
+
+func (i *interceptor) NewStream(ctx context.Context, _ resolver.RPCInfo, newStream func(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStream, error), opts ...grpc.CallOption) (grpc.ClientStream, error) {
+	clusterName := clustermanager.PickedCluster(ctx)
+	// The picked cluster name in the context is formatted by the xDS
+	// resolver with a prefix to distinguish between standard CDS clusters and
+	// cluster_specifier_plugins:
+	// - cluster_specifier_plugin: Since the target cluster is dynamically
+	//   resolved later at the load balancing layer, the final cluster name is
+	//   not yet known at this stage, hence we bypass this filter.
+	// - cluster: Standard CDS routing where the destination cluster is
+	//   known at routing time. We strip the prefix to get the raw cluster name,
+	//   which is used to look up the cluster configuration in the CDS response.
+	if strings.HasPrefix(clusterName, "cluster_specifier_plugin:") {
+		return newStream(ctx, opts...)
+	}
+	clusterName = strings.TrimPrefix(clusterName, "cluster:")
+
+	cfg := xdsresource.XDSConfigFromContext(ctx)
+	if cfg == nil {
+		return nil, status.Errorf(codes.Unavailable, "gcpauthn: xDS config not found in context")
+	}
+
+	clusterResult, ok := cfg.Clusters[clusterName]
+	if !ok {
+		return nil, status.Errorf(codes.Unavailable, "gcpauthn: cluster %q not found in xDS config", clusterName)
+	}
+
+	if clusterResult.Err != nil {
+		return nil, status.Errorf(codes.Unavailable, "gcpauthn: cluster config for %q is invalid or missing: %v", clusterName, clusterResult.Err)
+	}
+
+	val, ok := clusterResult.Config.Cluster.Metadata[i.filterName]
+	if !ok {
+		return newStream(ctx, opts...)
+	}
+
+	audienceMetadata, ok := val.(xdsresource.AudienceMetadataValue)
+	if !ok {
+		return nil, status.Errorf(codes.Unavailable, "gcpauthn: cluster metadata for key %q is not of type AudienceMetadataValue, got %T", i.filterName, val)
+	}
+
+	creds, err := i.cache.getOrCreate(i.ctx, audienceMetadata.Audience)
+	if err != nil {
+		return nil, status.Errorf(codes.Unavailable, "gcpauthn: failed to create credentials: %v", err)
+	}
+
+	// We pass the credentials via a PerRPCCredentials call option rather than
+	// directly attaching the token here. Since this filter runs before load
+	// balancer has selected a connection, it cannot check if the connection is
+	// secure. Passing it as a call option defers token retrieval and injection
+	// to the credentials package, which can verify transport security after
+	// the connection has been established.
+	opts = append(opts, grpc.PerRPCCredentials(creds))
+
+	return newStream(ctx, opts...)
+}
+
+func (i *interceptor) Close() {}
+
+// cacheEntry represents a cached PerRPCCredentials instance and its position
+// in the LRU list.
+type cacheEntry struct {
+	creds credentials.PerRPCCredentials
+	elem  *list.Element
+}
+
+// lruCache is a thread-safe LRU cache that stores PerRPCCredentials instances
+// by their target audience string.
+type lruCache struct {
+	// The following fields are protected by mu.
+	mu sync.Mutex
+
+	// cacheSize is the maximum capacity of the lruList.
+	cacheSize uint64
+
+	// lruList is a doubly linked list tracking access order. The front of the
+	// list holds the most recently used entry, and the back holds the least
+	// recently used entry. Elements in the list store values of type
+	// string (representing the target audience).
+	lruList *list.List
+
+	// cache maps audience keys to their corresponding cacheEntry pointers,
+	// allowing O(1) lookups and updates in the LRU list.
+	cache map[string]*cacheEntry
+
+	// sf is used to deduplicate credential creation for the same audience.
+	sf singleflight.Group
+}
+
+// newLRUCache instantiates a new lruCache with the specified capacity.
+func newLRUCache(size uint64) *lruCache {
+	return &lruCache{
+		cacheSize: size,
+		lruList:   list.New(),
+		cache:     make(map[string]*cacheEntry),
+	}
+}
+
+// resizeCache dynamically updates the capacity of the LRU cache,
+// immediately evicting Least Recently Used entries if the new size is
+// smaller than the current cache size.
+func (c *lruCache) resizeCache(newCacheSize uint64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.cacheSize == newCacheSize {
+		return
+	}
+
+	c.cacheSize = newCacheSize
+	for uint64(len(c.cache)) > c.cacheSize {
+		c.removeOldestLocked()
+	}
+}
+
+// getOrCreate retrieves or constructs the PerRPCCredentials for a specified
+// audience. If the audience is not found in the cache, it creates new
+// credentials using the configured creator, adds it to the cache, and evicts
+// the least recently used entry if the cache size is at capacity.
+func (c *lruCache) getOrCreate(ctx context.Context, audience string) (credentials.PerRPCCredentials, error) {
+	if creds, ok := c.getExisting(audience); ok {
+		return creds, nil
+	}
+
+	val, err, _ := c.sf.Do(audience, func() (any, error) {
+		// Double-check cache inside singleflight callback in case another call
+		// completed the initial check before the current call updates the cache.
+		if creds, ok := c.getExisting(audience); ok {
+			return creds, nil
+		}
+
+		creds, err := google.NewServiceAccountIdentityCredentials(ctx, audience)
+		if err != nil {
+			return nil, err
+		}
+
+		c.mu.Lock()
+		defer c.mu.Unlock()
+
+		if uint64(len(c.cache)) >= c.cacheSize {
+			c.removeOldestLocked()
+		}
+		c.cache[audience] = &cacheEntry{
+			creds: creds,
+			elem:  c.lruList.PushFront(audience),
+		}
+		return creds, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return val.(credentials.PerRPCCredentials), nil
+}
+
+// getExisting retrieves the PerRPCCredentials for a specified audience if it
+// already exists in the cache and updates the access order in the LRU list.
+//
+// The boolean return value indicates if an entry was found in the cache.
+func (c *lruCache) getExisting(audience string) (credentials.PerRPCCredentials, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if entry, ok := c.cache[audience]; ok {
+		c.lruList.MoveToFront(entry.elem)
+		return entry.creds, ok
+	}
+	return nil, false
+}
+
+// removeOldestLocked evicts the least recently used entry from the cache.
+// It must be called with mu locked.
+func (c *lruCache) removeOldestLocked() {
+	oldest := c.lruList.Back()
+	delete(c.cache, oldest.Value.(string))
+	c.lruList.Remove(oldest)
+}