Skip to content

xds: propagate audience from cluster resource in gcp auth filter #11972

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shivaspeaks merged 8 commits into from
Apr 2, 2025
Merged

xds: propagate audience from cluster resource in gcp auth filter #11972

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
3b9f264
xds: propagate audience from cluster resource in gcp auth filter
shivaspeaks Mar 21, 2025
f2dffe4
address comments
shivaspeaks Mar 25, 2025
cf09105
address comments
shivaspeaks Mar 25, 2025
e6be504
address comments
shivaspeaks Mar 26, 2025
bcec5d0
address comments and add test for failure cases
shivaspeaks Mar 26, 2025
0236a04
Fix unit tests
shivaspeaks Apr 1, 2025
a845f3d
fix unit tests
shivaspeaks Apr 2, 2025
e64df5d
fix unit tests
shivaspeaks Apr 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 75 additions & 13 deletions xds/src/main/java/io/grpc/xds/GcpAuthenticationFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.annotations.VisibleForTesting;
import com.google.common.primitives.UnsignedLongs;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
Expand All @@ -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;
Expand All @@ -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() {
Expand All @@ -65,7 +80,7 @@ public boolean isClientFilter() {

@Override
public GcpAuthenticationFilter newInstance(String name) {
return new GcpAuthenticationFilter();
return new GcpAuthenticationFilter(name);
}

@Override
Expand Down Expand Up @@ -119,22 +134,59 @@ 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);
Comment thread
ejona86 marked this conversation as resolved.
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:")) {
Comment thread
ejona86 marked this conversation as resolved.
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)));
}

StatusOr<XdsClusterConfig> xdsCluster =
xdsConfig.getClusters().get(clusterName.substring("cluster:".length()));
if (xdsCluster == null) {
return new FailingClientCall<>(
Status.UNAVAILABLE.withDescription(
String.format(
"GCP Authn for %s with %s - xds cluster config does not contain xds cluster",
filterInstanceName, clusterName)));
}

if (!xdsCluster.hasValue()) {
Comment thread
ejona86 marked this conversation as resolved.
return new FailingClientCall<>(xdsCluster.getStatus());
}

// TODO: Fetch the CDS resource for the cluster.
// If the CDS resource is not available, fail the RPC with Status.UNAVAILABLE.
Object audienceObj =
xdsCluster.getValue().getClusterResource().parsedMetadata().get(filterInstanceName);
if (audienceObj == null) {
return next.newCall(method, callOptions);
}

// 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 (!(audienceObj instanceof AudienceWrapper)) {
return new FailingClientCall<>(
Status.UNAVAILABLE.withDescription(
String.format("GCP Authn found wrong type in %s metadata: %s=%s",
clusterName, filterInstanceName,
audienceObj == null ? null : audienceObj.getClass())));
}
AudienceWrapper audience = (AudienceWrapper) audienceObj;

try {
CallCredentials existingCallCredentials = callOptions.getCredentials();
Comment thread
kannanjgithub marked this conversation as resolved.
Outdated
CallCredentials newCallCredentials =
getCallCredentials(callCredentialsCache, audience, credentials);
getCallCredentials(callCredentialsCache, audience.audience, credentials);
Comment thread
ejona86 marked this conversation as resolved.
Outdated
if (existingCallCredentials != null) {
callOptions = callOptions.withCallCredentials(
new CompositeCallCredentials(existingCallCredentials, newCallCredentials));
Expand Down Expand Up @@ -186,9 +238,11 @@ public String typeUrl() {
}

/** An implementation of {@link ClientCall} that fails when started. */
private static final class FailingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
@VisibleForTesting
static final class FailingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {

private final Status error;
@VisibleForTesting
final Status error;

public FailingClientCall(Status error) {
this.error = error;
Expand Down Expand Up @@ -235,13 +289,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 = checkNotNull(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);
Expand All @@ -253,7 +315,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);
}
}
}
155 changes: 149 additions & 6 deletions xds/src/test/java/io/grpc/xds/GcpAuthenticationFilterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,25 +17,54 @@
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 io.grpc.xds.XdsTestUtils.CLUSTER_NAME;
import static io.grpc.xds.XdsTestUtils.EDS_NAME;
import static io.grpc.xds.XdsTestUtils.ENDPOINT_HOSTNAME;
import static io.grpc.xds.XdsTestUtils.ENDPOINT_PORT;
import static io.grpc.xds.XdsTestUtils.RDS_NAME;
import static io.grpc.xds.XdsTestUtils.buildRouteConfiguration;
import static io.grpc.xds.XdsTestUtils.getWrrLbConfigAsMap;
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.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.Any;
import com.google.protobuf.Empty;
import com.google.protobuf.Message;
import com.google.protobuf.UInt64Value;
import io.envoyproxy.envoy.config.route.v3.RouteConfiguration;
import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.GcpAuthnFilterConfig;
import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.TokenCacheConfig;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.StatusOr;
import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.testing.TestMethodDescriptors;
import io.grpc.xds.Endpoints.LbEndpoint;
import io.grpc.xds.Endpoints.LocalityLbEndpoints;
import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser.AudienceWrapper;
import io.grpc.xds.GcpAuthenticationFilter.FailingClientCall;
import io.grpc.xds.GcpAuthenticationFilter.GcpAuthenticationConfig;
import io.grpc.xds.XdsClusterResource.CdsUpdate;
import io.grpc.xds.XdsConfig.XdsClusterConfig;
import io.grpc.xds.XdsConfig.XdsClusterConfig.EndpointConfig;
import io.grpc.xds.client.Locality;
import io.grpc.xds.client.XdsResourceType;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
Expand Down Expand Up @@ -92,22 +121,136 @@ public void testParseFilterConfig_withInvalidMessageType() {
}

Comment thread
kannanjgithub marked this conversation as resolved.
@Test
public void testClientInterceptor_createsAndReusesCachedCredentials() {
public void testClientInterceptor() throws Exception {
Comment thread
kannanjgithub marked this conversation as resolved.
Outdated
String serverName = InProcessServerBuilder.generateName();
XdsConfig.XdsConfigBuilder builder = new XdsConfig.XdsConfigBuilder();

Filter.NamedFilterConfig routerFilterConfig = new Filter.NamedFilterConfig(
serverName, RouterFilter.ROUTER_CONFIG);

HttpConnectionManager httpConnectionManager = HttpConnectionManager.forRdsName(
0L, RDS_NAME, Collections.singletonList(routerFilterConfig));
XdsListenerResource.LdsUpdate ldsUpdate =
XdsListenerResource.LdsUpdate.forApiListener(httpConnectionManager);

RouteConfiguration routeConfiguration =
buildRouteConfiguration(serverName, RDS_NAME, CLUSTER_NAME);
XdsResourceType.Args args = new XdsResourceType.Args(null, "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(
serverName, 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
Comment thread
kannanjgithub marked this conversation as resolved.
Outdated
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)));

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);
ArgumentCaptor<CallOptions> callOptionsCaptor = ArgumentCaptor.forClass(CallOptions.class);
Channel mockChannel = mock(Channel.class);

// Set CallOptions with required keys
CallOptions callOptionsWithXds = CallOptions.DEFAULT;

// Execute interception twice to check caching
ClientCall<Void, Void> call = interceptor.interceptCall(
Comment thread
kannanjgithub marked this conversation as resolved.
Outdated
Comment thread
kannanjgithub marked this conversation as resolved.
methodDescriptor, callOptionsWithXds, mockChannel);
assertTrue(call instanceof FailingClientCall);
FailingClientCall<Void, Void> clientCall = (FailingClientCall<Void, Void>) call;
assertThat(clientCall.error.getDescription()).contains("does not contain cluster resource");

callOptionsWithXds = CallOptions.DEFAULT
.withOption(CLUSTER_SELECTION_KEY, "cluster:cluster0");

// Execute interception twice to check caching
call = interceptor.interceptCall(methodDescriptor, callOptionsWithXds, mockChannel);
assertTrue(call instanceof FailingClientCall);
clientCall = (FailingClientCall<Void, Void>) call;
assertThat(clientCall.error.getDescription()).contains("does not contain xds configuration");

XdsConfig defaultXdsConfig = builder
.setListener(ldsUpdate)
.setRoute(rdsUpdate)
.setVirtualHost(virtualHost)
.addCluster(CLUSTER_NAME, StatusOr.fromValue(clusterConfig)).build();
callOptionsWithXds = CallOptions.DEFAULT
.withOption(CLUSTER_SELECTION_KEY, "cluster:cluster")
Comment thread
kannanjgithub marked this conversation as resolved.
Outdated
.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);
call = interceptor.interceptCall(methodDescriptor, callOptionsWithXds, mockChannel);
assertTrue(call instanceof FailingClientCall);
clientCall = (FailingClientCall<Void, Void>) call;
assertThat(clientCall.error.getDescription()).contains("does not contain xds cluster");

StatusOr<XdsClusterConfig> errorCluster =
StatusOr.fromStatus(Status.NOT_FOUND.withDescription("Cluster resource not found"));
defaultXdsConfig = builder
.setListener(ldsUpdate)
.setRoute(rdsUpdate)
.setVirtualHost(virtualHost)
.addCluster(CLUSTER_NAME, errorCluster).build();
callOptionsWithXds = CallOptions.DEFAULT
.withOption(CLUSTER_SELECTION_KEY, "cluster:cluster0")
.withOption(XDS_CONFIG_CALL_OPTION_KEY, defaultXdsConfig);

// Create interceptor
interceptor = filter.buildClientInterceptor(config, null, null);
methodDescriptor = TestMethodDescriptors.voidMethod();

// Mock channel and capture CallOptions
mockChannel = mock(Channel.class);
call = interceptor.interceptCall(methodDescriptor, callOptionsWithXds, mockChannel);
assertTrue(call instanceof FailingClientCall);
clientCall = (FailingClientCall<Void, Void>) call;
assertThat(clientCall.error.getDescription())
.contains("Cluster resource not found");

// Success case
defaultXdsConfig = builder
.setListener(ldsUpdate)
.setRoute(rdsUpdate)
.setVirtualHost(virtualHost)
.addCluster(CLUSTER_NAME, StatusOr.fromValue(clusterConfig)).build();
// Set CallOptions with required keys
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, callOptionsWithXds, mockChannel);
interceptor.interceptCall(methodDescriptor, callOptionsWithXds, mockChannel);

ArgumentCaptor<CallOptions> callOptionsCaptor = ArgumentCaptor.forClass(CallOptions.class);
// Capture and verify CallOptions for CallCredentials presence
Mockito.verify(mockChannel, Mockito.times(2))
.newCall(eq(methodDescriptor), callOptionsCaptor.capture());
Expand Down
Loading