#19671

Committed 30 Jan 2025 08:43PM UTC coverage: 88.581% (-0.01%) from 88.595%

Build # #19671

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Reuse filter interceptors across RPCs

This moves the interceptor creation from the ConfigSelector to the
resource update handling.

The code structure changes will make adding support for filter
lifecycles (for RLQS) a bit easier. The filter lifecycles will allow
filters to share state across interceptors, and constructing all the
interceptors on a single thread will mean filters wouldn't need to be
thread-safe (but their interceptors would be thread-safe).

Run Details

33760 of 38112 relevant lines covered (88.58%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.43

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

/*
 * Copyright 2021 The 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 io.grpc.xds;

import com.google.auth.oauth2.ComputeEngineCredentials;
import com.google.auth.oauth2.IdTokenCredentials;
import com.google.common.primitives.UnsignedLongs;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;
import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.Audience;
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.CallCredentials;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.CompositeCallCredentials;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.auth.MoreCallCredentials;
import io.grpc.xds.Filter.ClientInterceptorBuilder;
import io.grpc.xds.MetadataRegistry.MetadataValueParser;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Function;
import javax.annotation.Nullable;

/**
 * A {@link Filter} that injects a {@link CallCredentials} to handle
 * authentication for xDS credentials.
 */
final class GcpAuthenticationFilter implements Filter, ClientInterceptorBuilder {

  static final String TYPE_URL =
      "type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.GcpAuthnFilterConfig";

  @Override
  public String[] typeUrls() {
    return new String[] { TYPE_URL };
  }

  @Override
  public ConfigOrError<? extends FilterConfig> parseFilterConfig(Message rawProtoMessage) {
    GcpAuthnFilterConfig gcpAuthnProto;
    if (!(rawProtoMessage instanceof Any)) {
      return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());
    }
    Any anyMessage = (Any) rawProtoMessage;

    try {
      gcpAuthnProto = anyMessage.unpack(GcpAuthnFilterConfig.class);
    } catch (InvalidProtocolBufferException e) {
      return ConfigOrError.fromError("Invalid proto: " + e);
    }

    long cacheSize = 10;
    // Validate cache_config
    if (gcpAuthnProto.hasCacheConfig()) {
      TokenCacheConfig cacheConfig = gcpAuthnProto.getCacheConfig();
      cacheSize = cacheConfig.getCacheSize().getValue();
      if (cacheSize == 0) {
        return ConfigOrError.fromError(
            "cache_config.cache_size must be greater than zero");
      }
      // LruCache's size is an int and briefly exceeds its maximum size before evicting entries
      cacheSize = UnsignedLongs.min(cacheSize, Integer.MAX_VALUE - 1);
    }

    GcpAuthenticationConfig config = new GcpAuthenticationConfig((int) cacheSize);
    return ConfigOrError.fromConfig(config);
  }

  @Override
  public ConfigOrError<? extends FilterConfig> parseFilterConfigOverride(Message rawProtoMessage) {
    return parseFilterConfig(rawProtoMessage);
  }

  @Nullable
  @Override
  public ClientInterceptor buildClientInterceptor(FilterConfig config,
      @Nullable FilterConfig overrideConfig, ScheduledExecutorService scheduler) {

    ComputeEngineCredentials credentials = ComputeEngineCredentials.create();
    LruCache<String, CallCredentials> callCredentialsCache =
        new LruCache<>(((GcpAuthenticationConfig) config).getCacheSize());
    return new ClientInterceptor() {
      @Override
      public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
          MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {

        /*String clusterName = callOptions.getOption(XdsAttributes.ATTR_CLUSTER_NAME);
        if (clusterName == null) {
          return next.newCall(method, callOptions);
        }*/

        // TODO: Fetch the CDS resource for the cluster.
        // If the CDS resource is not available, fail the RPC with Status.UNAVAILABLE.

        // 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";

        try {
          CallCredentials existingCallCredentials = callOptions.getCredentials();
          CallCredentials newCallCredentials =
              getCallCredentials(callCredentialsCache, audience, credentials);
          if (existingCallCredentials != null) {
            callOptions = callOptions.withCallCredentials(
                new CompositeCallCredentials(existingCallCredentials, newCallCredentials));
          } else {
            callOptions = callOptions.withCallCredentials(newCallCredentials);
          }
        }
        catch (Exception e) {
          // If we fail to attach CallCredentials due to any reason, return a FailingClientCall
          return new FailingClientCall<>(Status.UNAUTHENTICATED
              .withDescription("Failed to attach CallCredentials.")
              .withCause(e));
        }
        return next.newCall(method, callOptions);
      }
    };
  }

  private CallCredentials getCallCredentials(LruCache<String, CallCredentials> cache,
      String audience, ComputeEngineCredentials credentials) {

    synchronized (cache) {
      return cache.getOrInsert(audience, key -> {
        IdTokenCredentials creds = IdTokenCredentials.newBuilder()
            .setIdTokenProvider(credentials)
            .setTargetAudience(audience)
            .build();
        return MoreCallCredentials.from(creds);
      });
    }
  }

  static final class GcpAuthenticationConfig implements FilterConfig {

    private final int cacheSize;

    public GcpAuthenticationConfig(int cacheSize) {
      this.cacheSize = cacheSize;
    }

    public int getCacheSize() {
      return cacheSize;
    }

    @Override
    public String typeUrl() {
      return GcpAuthenticationFilter.TYPE_URL;
    }
  }

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

    private final Status error;

    public FailingClientCall(Status error) {
      this.error = error;
    }

    @Override
    public void start(ClientCall.Listener<RespT> listener, Metadata headers) {
      listener.onClose(error, new Metadata());
    }

    @Override
    public void request(int numMessages) {}

    @Override
    public void cancel(String message, Throwable cause) {}

    @Override
    public void halfClose() {}

    @Override
    public void sendMessage(ReqT message) {}
  }

  private static final class LruCache<K, V> {

    private final Map<K, V> cache;

    LruCache(int maxSize) {
      this.cache = new LinkedHashMap<K, V>(
          maxSize,
          0.75f,
          true) {
        @Override
        protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
          return size() > maxSize;
        }
      };
    }

    V getOrInsert(K key, Function<K, V> create) {
      return cache.computeIfAbsent(key, create);
    }
  }

  static class AudienceMetadataParser implements MetadataValueParser {

    @Override
    public String getTypeUrl() {
      return "type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.Audience";
    }

    @Override
    public String parse(Any any) throws InvalidProtocolBufferException {
      Audience audience = any.unpack(Audience.class);
      String url = audience.getUrl();
      if (url.isEmpty()) {
        throw new InvalidProtocolBufferException(
            "Audience URL is empty. Metadata value must contain a valid URL.");
      }
      return url;
    }
  }
}

1	/*
2	* Copyright 2021 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.xds;
18
19	import com.google.auth.oauth2.ComputeEngineCredentials;
20	import com.google.auth.oauth2.IdTokenCredentials;
21	import com.google.common.primitives.UnsignedLongs;
22	import com.google.protobuf.Any;
23	import com.google.protobuf.InvalidProtocolBufferException;
24	import com.google.protobuf.Message;
25	import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.Audience;
26	import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.GcpAuthnFilterConfig;
27	import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.TokenCacheConfig;
28	import io.grpc.CallCredentials;
29	import io.grpc.CallOptions;
30	import io.grpc.Channel;
31	import io.grpc.ClientCall;
32	import io.grpc.ClientInterceptor;
33	import io.grpc.CompositeCallCredentials;
34	import io.grpc.Metadata;
35	import io.grpc.MethodDescriptor;
36	import io.grpc.Status;
37	import io.grpc.auth.MoreCallCredentials;
38	import io.grpc.xds.Filter.ClientInterceptorBuilder;
39	import io.grpc.xds.MetadataRegistry.MetadataValueParser;
40	import java.util.LinkedHashMap;
41	import java.util.Map;
42	import java.util.concurrent.ScheduledExecutorService;
43	import java.util.function.Function;
44	import javax.annotation.Nullable;
45
46	/**
47	* A {@link Filter} that injects a {@link CallCredentials} to handle
48	* authentication for xDS credentials.
49	*/
50	final class GcpAuthenticationFilter implements Filter, ClientInterceptorBuilder {	1✔
51
52	static final String TYPE_URL =
53	"type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.GcpAuthnFilterConfig";
54
55	@Override
56	public String[] typeUrls() {
57	return new String[] { TYPE_URL };	×
58	}
59
60	@Override
61	public ConfigOrError<? extends FilterConfig> parseFilterConfig(Message rawProtoMessage) {
62	GcpAuthnFilterConfig gcpAuthnProto;
63	if (!(rawProtoMessage instanceof Any)) {	1✔
64	return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());	1✔
65	}
66	Any anyMessage = (Any) rawProtoMessage;	1✔
67
68	try {
69	gcpAuthnProto = anyMessage.unpack(GcpAuthnFilterConfig.class);	1✔
70	} catch (InvalidProtocolBufferException e) {	×
71	return ConfigOrError.fromError("Invalid proto: " + e);	×
72	}	1✔
73
74	long cacheSize = 10;	1✔
75	// Validate cache_config
76	if (gcpAuthnProto.hasCacheConfig()) {	1✔
77	TokenCacheConfig cacheConfig = gcpAuthnProto.getCacheConfig();	1✔
78	cacheSize = cacheConfig.getCacheSize().getValue();	1✔
79	if (cacheSize == 0) {	1✔
80	return ConfigOrError.fromError(	1✔
81	"cache_config.cache_size must be greater than zero");
82	}
83	// LruCache's size is an int and briefly exceeds its maximum size before evicting entries
84	cacheSize = UnsignedLongs.min(cacheSize, Integer.MAX_VALUE - 1);	1✔
85	}
86
87	GcpAuthenticationConfig config = new GcpAuthenticationConfig((int) cacheSize);	1✔
88	return ConfigOrError.fromConfig(config);	1✔
89	}
90
91	@Override
92	public ConfigOrError<? extends FilterConfig> parseFilterConfigOverride(Message rawProtoMessage) {
93	return parseFilterConfig(rawProtoMessage);	×
94	}
95
96	@Nullable
97	@Override
98	public ClientInterceptor buildClientInterceptor(FilterConfig config,
99	@Nullable FilterConfig overrideConfig, ScheduledExecutorService scheduler) {
100
101	ComputeEngineCredentials credentials = ComputeEngineCredentials.create();	1✔
102	LruCache<String, CallCredentials> callCredentialsCache =	1✔
103	new LruCache<>(((GcpAuthenticationConfig) config).getCacheSize());	1✔
104	return new ClientInterceptor() {	1✔
105	@Override
106	public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
107	MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
108
109	/*String clusterName = callOptions.getOption(XdsAttributes.ATTR_CLUSTER_NAME);
110	if (clusterName == null) {
111	return next.newCall(method, callOptions);
112	}*/
113
114	// TODO: Fetch the CDS resource for the cluster.
115	// If the CDS resource is not available, fail the RPC with Status.UNAVAILABLE.
116
117	// TODO: Extract the audience from the CDS resource metadata.
118	// If the audience is not found or is in the wrong format, fail the RPC.
119	String audience = "TEST_AUDIENCE";	1✔
120
121	try {
122	CallCredentials existingCallCredentials = callOptions.getCredentials();	1✔
123	CallCredentials newCallCredentials =	1✔
124	getCallCredentials(callCredentialsCache, audience, credentials);	1✔
125	if (existingCallCredentials != null) {	1✔
126	callOptions = callOptions.withCallCredentials(	×
127	new CompositeCallCredentials(existingCallCredentials, newCallCredentials));
128	} else {
129	callOptions = callOptions.withCallCredentials(newCallCredentials);	1✔
130	}
131	}
132	catch (Exception e) {	×
133	// If we fail to attach CallCredentials due to any reason, return a FailingClientCall
134	return new FailingClientCall<>(Status.UNAUTHENTICATED	×
135	.withDescription("Failed to attach CallCredentials.")	×
136	.withCause(e));	×
137	}	1✔
138	return next.newCall(method, callOptions);	1✔
139	}
140	};
141	}
142
143	private CallCredentials getCallCredentials(LruCache<String, CallCredentials> cache,
144	String audience, ComputeEngineCredentials credentials) {
145
146	synchronized (cache) {	1✔
147	return cache.getOrInsert(audience, key -> {	1✔
148	IdTokenCredentials creds = IdTokenCredentials.newBuilder()	1✔
149	.setIdTokenProvider(credentials)	1✔
150	.setTargetAudience(audience)	1✔
151	.build();	1✔
152	return MoreCallCredentials.from(creds);	1✔
153	});
154	}
155	}
156
157	static final class GcpAuthenticationConfig implements FilterConfig {
158
159	private final int cacheSize;
160
161	public GcpAuthenticationConfig(int cacheSize) {	1✔
162	this.cacheSize = cacheSize;	1✔
163	}	1✔
164
165	public int getCacheSize() {
166	return cacheSize;	1✔
167	}
168
169	@Override
170	public String typeUrl() {
171	return GcpAuthenticationFilter.TYPE_URL;	×
172	}
173	}
174
175	/** An implementation of {@link ClientCall} that fails when started. */
176	private static final class FailingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
177
178	private final Status error;
179
180	public FailingClientCall(Status error) {	×
181	this.error = error;	×
182	}	×
183
184	@Override
185	public void start(ClientCall.Listener<RespT> listener, Metadata headers) {
186	listener.onClose(error, new Metadata());	×
187	}	×
188
189	@Override
190	public void request(int numMessages) {}	×
191
192	@Override
193	public void cancel(String message, Throwable cause) {}	×
194
195	@Override
196	public void halfClose() {}	×
197
198	@Override
199	public void sendMessage(ReqT message) {}	×
200	}
201
202	private static final class LruCache<K, V> {
203
204	private final Map<K, V> cache;
205
206	LruCache(int maxSize) {	1✔
207	this.cache = new LinkedHashMap<K, V>(	1✔
208	maxSize,
209	0.75f,
210	true) {	1✔
211	@Override
212	protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
213	return size() > maxSize;	1✔
214	}
215	};
216	}	1✔
217
218	V getOrInsert(K key, Function<K, V> create) {
219	return cache.computeIfAbsent(key, create);	1✔
220	}
221	}
222
223	static class AudienceMetadataParser implements MetadataValueParser {	1✔
224
225	@Override
226	public String getTypeUrl() {
227	return "type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.Audience";	1✔
228	}
229
230	@Override
231	public String parse(Any any) throws InvalidProtocolBufferException {
232	Audience audience = any.unpack(Audience.class);	1✔
233	String url = audience.getUrl();	1✔
234	if (url.isEmpty()) {	1✔
235	throw new InvalidProtocolBufferException(	×
236	"Audience URL is empty. Metadata value must contain a valid URL.");
237	}
238	return url;	1✔
239	}
240	}
241	}

grpc / grpc-java / #19671

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous