xds: propagate audience from cluster resource in gcp auth filter

xds/src/main/java/io/grpc/xds/GcpAuthenticationFilter.java

@@ -16,8 +16,13 @@
 
 package io.grpc.xds;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+import static io.grpc.xds.XdsNameResolver.CLUSTER_SELECTION_KEY;
+import static io.grpc.xds.XdsNameResolver.XDS_CONFIG_CALL_OPTION_KEY;
+
 import com.google.auth.oauth2.ComputeEngineCredentials;
 import com.google.auth.oauth2.IdTokenCredentials;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.primitives.UnsignedLongs;
 import com.google.protobuf.Any;
 import com.google.protobuf.InvalidProtocolBufferException;
@@ -34,8 +39,11 @@
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
+import io.grpc.StatusOr;
 import io.grpc.auth.MoreCallCredentials;
+import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser.AudienceWrapper;
 import io.grpc.xds.MetadataRegistry.MetadataValueParser;
+import io.grpc.xds.XdsConfig.XdsClusterConfig;
 import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -52,6 +60,13 @@ final class GcpAuthenticationFilter implements Filter {
   static final String TYPE_URL =
       "type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.GcpAuthnFilterConfig";
 
+  final String filterInstanceName;
+
+  GcpAuthenticationFilter(String name) {
+    filterInstanceName = checkNotNull(name, "name");
+  }
+
+
   static final class Provider implements Filter.Provider {
     @Override
     public String[] typeUrls() {
@@ -65,7 +80,7 @@ public boolean isClientFilter() {
 
     @Override
     public GcpAuthenticationFilter newInstance(String name) {
-      return new GcpAuthenticationFilter();
+      return new GcpAuthenticationFilter(name);
     }
 
     @Override
@@ -119,22 +134,66 @@ public ClientInterceptor buildClientInterceptor(FilterConfig config,
       public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
           MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
 
-        /*String clusterName = callOptions.getOption(XdsAttributes.ATTR_CLUSTER_NAME);
+        String clusterName = callOptions.getOption(CLUSTER_SELECTION_KEY);
         if (clusterName == null) {
+          return new FailingClientCall<>(
+              Status.UNAVAILABLE.withDescription(
+                  String.format(
+                      "GCP Authn for %s does not contain cluster resource", filterInstanceName)));
+        }
+
+        if (!clusterName.startsWith("cluster:")) {
           return next.newCall(method, callOptions);
-        }*/
+        }
+        XdsConfig xdsConfig = callOptions.getOption(XDS_CONFIG_CALL_OPTION_KEY);
+        if (xdsConfig == null) {
+          return new FailingClientCall<>(
+              Status.UNAVAILABLE.withDescription(
+                  String.format(
+                      "GCP Authn for %s with %s does not contain xds configuration",
+                      filterInstanceName, clusterName)));
+        }
 
-        // TODO: Fetch the CDS resource for the cluster.
-        // If the CDS resource is not available, fail the RPC with Status.UNAVAILABLE.
+        StatusOr<XdsClusterConfig> xdsCluster =
+            xdsConfig.getClusters().get(clusterName.substring(8)); // get rid of prefix "cluster:"
+        if (xdsCluster == null) {
+          return new FailingClientCall<>(
+              Status.UNAVAILABLE.withDescription(
+                  String.format(
+                      "GCP Authn for %s with %s does not contain xds cluster",
+                      filterInstanceName, clusterName)));
+        }
 
-        // TODO: Extract the audience from the CDS resource metadata.
-        // If the audience is not found or is in the wrong format, fail the RPC.
-        String audience = "TEST_AUDIENCE";
+        if (!xdsCluster.hasValue()) {
+          return next.newCall(method, callOptions);
+        }
+        ImmutableMap<String, Object> parsedMetadata = xdsCluster.getValue().getClusterResource()
+            .parsedMetadata();
+
+        if (parsedMetadata == null || !parsedMetadata.containsKey(filterInstanceName)) {
+          return next.newCall(method, callOptions);
+        }
+
+        AudienceWrapper audience;
+        if (parsedMetadata.get(filterInstanceName) instanceof AudienceWrapper) {
+          audience = (AudienceWrapper) parsedMetadata.get(filterInstanceName);
+          if (audience.audience == null) {
+            return next.newCall(method, callOptions);
+          }
+        }
+        else {
+          return new FailingClientCall<>(
+              Status.UNAVAILABLE.withDescription(
+                  String.format("GCP Authn found wrong type in %s metadata: %s=%s",
+                      clusterName, filterInstanceName,
+                      parsedMetadata.get(filterInstanceName) == null
+                          ? null : parsedMetadata.get(filterInstanceName))));
+        }
 
         try {
           CallCredentials existingCallCredentials = callOptions.getCredentials();
           CallCredentials newCallCredentials =
-              getCallCredentials(callCredentialsCache, audience, credentials);
+              getCallCredentials(callCredentialsCache, audience.audience, credentials);
           if (existingCallCredentials != null) {
             callOptions = callOptions.withCallCredentials(
                 new CompositeCallCredentials(existingCallCredentials, newCallCredentials));
@@ -235,13 +294,21 @@ V getOrInsert(K key, Function<K, V> create) {
 
   static class AudienceMetadataParser implements MetadataValueParser {
 
+    static final class AudienceWrapper {
+      final String audience;
+
+      AudienceWrapper(String audience) {
+        this.audience = audience;
+      }
+    }
+
     @Override
     public String getTypeUrl() {
       return "type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.Audience";
     }
 
     @Override
-    public String parse(Any any) throws ResourceInvalidException {
+    public AudienceWrapper parse(Any any) throws ResourceInvalidException {
       Audience audience;
       try {
         audience = any.unpack(Audience.class);
@@ -253,7 +320,7 @@ public String parse(Any any) throws ResourceInvalidException {
         throw new ResourceInvalidException(
             "Audience URL is empty. Metadata value must contain a valid URL.");
       }
-      return url;
+      return new AudienceWrapper(url);
     }
   }
 }

xds/src/test/java/io/grpc/xds/GcpAuthenticationFilterTest.java

@@ -17,12 +17,15 @@
 package io.grpc.xds;
 
 import static com.google.common.truth.Truth.assertThat;
+import static io.grpc.xds.XdsNameResolver.CLUSTER_SELECTION_KEY;
+import static io.grpc.xds.XdsNameResolver.XDS_CONFIG_CALL_OPTION_KEY;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
 
 import com.google.protobuf.Any;
 import com.google.protobuf.Empty;
@@ -34,6 +37,7 @@
 import io.grpc.Channel;
 import io.grpc.ClientInterceptor;
 import io.grpc.MethodDescriptor;
+import io.grpc.inprocess.InProcessServerBuilder;
 import io.grpc.testing.TestMethodDescriptors;
 import io.grpc.xds.GcpAuthenticationFilter.GcpAuthenticationConfig;
 import org.junit.Test;
@@ -92,21 +96,29 @@ public void testParseFilterConfig_withInvalidMessageType() {
   }
 
   @Test
-  public void testClientInterceptor_createsAndReusesCachedCredentials() {
+  public void testClientInterceptor_createsAndReusesCachedCredentials() throws Exception {
+    String serverName = InProcessServerBuilder.generateName();
+    XdsConfig defaultXdsConfig = XdsTestUtils.getDefaultXdsConfigWithCdsUpdate(serverName);
+
     GcpAuthenticationConfig config = new GcpAuthenticationConfig(10);
-    GcpAuthenticationFilter filter = new GcpAuthenticationFilter();
+    GcpAuthenticationFilter filter = new GcpAuthenticationFilter("FILTER_INSTANCE_NAME");
 
     // Create interceptor
     ClientInterceptor interceptor = filter.buildClientInterceptor(config, null, null);
     MethodDescriptor<Void, Void> methodDescriptor = TestMethodDescriptors.voidMethod();
 
     // Mock channel and capture CallOptions
-    Channel mockChannel = Mockito.mock(Channel.class);
+    Channel mockChannel = mock(Channel.class);
     ArgumentCaptor<CallOptions> callOptionsCaptor = ArgumentCaptor.forClass(CallOptions.class);
 
+    // Set CallOptions with required keys
+    CallOptions callOptionsWithXds = CallOptions.DEFAULT
+        .withOption(CLUSTER_SELECTION_KEY, "cluster:cluster0")
+        .withOption(XDS_CONFIG_CALL_OPTION_KEY, defaultXdsConfig);
+
     // Execute interception twice to check caching
-    interceptor.interceptCall(methodDescriptor, CallOptions.DEFAULT, mockChannel);
-    interceptor.interceptCall(methodDescriptor, CallOptions.DEFAULT, mockChannel);
+    interceptor.interceptCall(methodDescriptor, callOptionsWithXds, mockChannel);
+    interceptor.interceptCall(methodDescriptor, callOptionsWithXds, mockChannel);
 
     // Capture and verify CallOptions for CallCredentials presence
     Mockito.verify(mockChannel, Mockito.times(2))

xds/src/test/java/io/grpc/xds/GrpcXdsClientImplDataTest.java

@@ -129,6 +129,7 @@
 import io.grpc.xds.Endpoints.LbEndpoint;
 import io.grpc.xds.Endpoints.LocalityLbEndpoints;
 import io.grpc.xds.Filter.FilterConfig;
+import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser.AudienceWrapper;
 import io.grpc.xds.MetadataRegistry.MetadataValueParser;
 import io.grpc.xds.RouteLookupServiceClusterSpecifierPlugin.RlsPluginConfig;
 import io.grpc.xds.VirtualHost.Route;
@@ -2417,8 +2418,7 @@ public Object parse(Any value) {
   }
 
   @Test
-  public void processCluster_parsesAudienceMetadata()
-      throws ResourceInvalidException, InvalidProtocolBufferException {
+  public void processCluster_parsesAudienceMetadata() throws Exception {
     MetadataRegistry.getInstance();
 
     Audience audience = Audience.newBuilder()
@@ -2462,7 +2462,10 @@ public void processCluster_parsesAudienceMetadata()
         "FILTER_METADATA", ImmutableMap.of(
             "key1", "value1",
             "key2", 42.0));
-    assertThat(update.parsedMetadata()).isEqualTo(expectedParsedMetadata);
+    assertThat(update.parsedMetadata().get("FILTER_METADATA"))
+        .isEqualTo(expectedParsedMetadata.get("FILTER_METADATA"));
+    assertThat(update.parsedMetadata().get("AUDIENCE_METADATA"))
+        .isInstanceOf(AudienceWrapper.class);
   }
 
   @Test
@@ -2519,8 +2522,7 @@ public void processCluster_parsesAddressMetadata() throws Exception {
   }
 
   @Test
-  public void processCluster_metadataKeyCollision_resolvesToTypedMetadata()
-      throws ResourceInvalidException, InvalidProtocolBufferException {
+  public void processCluster_metadataKeyCollision_resolvesToTypedMetadata() throws Exception {
     MetadataRegistry metadataRegistry = MetadataRegistry.getInstance();
 
     MetadataValueParser testParser =
@@ -2575,8 +2577,7 @@ public Object parse(Any value) {
   }
 
   @Test
-  public void parseNonAggregateCluster_withHttp11ProxyTransportSocket()
-      throws ResourceInvalidException, InvalidProtocolBufferException {
+  public void parseNonAggregateCluster_withHttp11ProxyTransportSocket() throws Exception {
     XdsClusterResource.isEnabledXdsHttpConnect = true;
 
     Http11ProxyUpstreamTransport http11ProxyUpstreamTransport =

xds/src/test/java/io/grpc/xds/XdsTestUtils.java

@@ -56,6 +56,8 @@
 import io.grpc.stub.StreamObserver;
 import io.grpc.xds.Endpoints.LbEndpoint;
 import io.grpc.xds.Endpoints.LocalityLbEndpoints;
+import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser.AudienceWrapper;
+import io.grpc.xds.XdsClusterResource.CdsUpdate;
 import io.grpc.xds.XdsConfig.XdsClusterConfig.EndpointConfig;
 import io.grpc.xds.client.Bootstrapper;
 import io.grpc.xds.client.Locality;
@@ -280,6 +282,63 @@ static XdsConfig getDefaultXdsConfig(String serverHostName)
     return builder.build();
   }
 
+  static XdsConfig getDefaultXdsConfigWithCdsUpdate(String serverHostName)
+      throws XdsResourceType.ResourceInvalidException, IOException {
+    XdsConfig.XdsConfigBuilder builder = new XdsConfig.XdsConfigBuilder();
+
+    Filter.NamedFilterConfig routerFilterConfig = new Filter.NamedFilterConfig(
+        serverHostName, RouterFilter.ROUTER_CONFIG);
+
+    HttpConnectionManager httpConnectionManager = HttpConnectionManager.forRdsName(
+        0L, RDS_NAME, Collections.singletonList(routerFilterConfig));
+    XdsListenerResource.LdsUpdate ldsUpdate =
+        XdsListenerResource.LdsUpdate.forApiListener(httpConnectionManager);
+
+    RouteConfiguration routeConfiguration =
+        buildRouteConfiguration(serverHostName, RDS_NAME, CLUSTER_NAME);
+    Bootstrapper.ServerInfo serverInfo = null;
+    XdsResourceType.Args args = new XdsResourceType.Args(serverInfo, "0", "0", null, null, null);
+    XdsRouteConfigureResource.RdsUpdate rdsUpdate =
+        XdsRouteConfigureResource.getInstance().doParse(args, routeConfiguration);
+
+    // Take advantage of knowing that there is only 1 virtual host in the route configuration
+    assertThat(rdsUpdate.virtualHosts).hasSize(1);
+    VirtualHost virtualHost = rdsUpdate.virtualHosts.get(0);
+
+    // Need to create endpoints to create locality endpoints map to create edsUpdate
+    Map<Locality, LocalityLbEndpoints> lbEndpointsMap = new HashMap<>();
+    LbEndpoint lbEndpoint = LbEndpoint.create(
+        serverHostName, ENDPOINT_PORT, 0, true, ENDPOINT_HOSTNAME, ImmutableMap.of());
+    lbEndpointsMap.put(
+        Locality.create("", "", ""),
+        LocalityLbEndpoints.create(ImmutableList.of(lbEndpoint), 10, 0, ImmutableMap.of()));
+
+    // Need to create EdsUpdate to create CdsUpdate to create XdsClusterConfig for builder
+    XdsEndpointResource.EdsUpdate edsUpdate = new XdsEndpointResource.EdsUpdate(
+        EDS_NAME, lbEndpointsMap, Collections.emptyList());
+
+    // Use ImmutableMap.Builder to construct the map
+    ImmutableMap.Builder<String, Object> parsedMetadata = ImmutableMap.builder();
+    parsedMetadata.put("FILTER_INSTANCE_NAME", new AudienceWrapper("TEST_AUDIENCE"));
+
+    CdsUpdate.Builder cdsUpdate = CdsUpdate.forEds(
+            CLUSTER_NAME, EDS_NAME, null, null, null, null, false)
+        .lbPolicyConfig(getWrrLbConfigAsMap());
+    cdsUpdate.parsedMetadata(parsedMetadata.build());
+    XdsConfig.XdsClusterConfig clusterConfig = new XdsConfig.XdsClusterConfig(
+        CLUSTER_NAME,
+        cdsUpdate.build(),
+        new EndpointConfig(StatusOr.fromValue(edsUpdate)));
+
+    builder
+        .setListener(ldsUpdate)
+        .setRoute(rdsUpdate)
+        .setVirtualHost(virtualHost)
+        .addCluster(CLUSTER_NAME, StatusOr.fromValue(clusterConfig));
+
+    return builder.build();
+  }
+
   static Map<Locality, LocalityLbEndpoints> createMinimalLbEndpointsMap(String serverHostName) {
     Map<Locality, LocalityLbEndpoints> lbEndpointsMap = new HashMap<>();
     LbEndpoint lbEndpoint = LbEndpoint.create(