#19715

Committed 06 Mar 2025 08:10AM UTC coverage: 88.479% (-0.04%) from 88.515%

Build # #19715

Build Type

push

github

Committed by

web-flow

Commit Message

xds: xDS-based HTTP CONNECT configuration (#11861)

Run Details

34489 of 38980 relevant lines covered (88.48%)

0.88 hits per line

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

69.74

/../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.MetadataRegistry.MetadataValueParser;
import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
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 {

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

  static final class Provider implements Filter.Provider {
    @Override
    public String[] typeUrls() {
      return new String[]{TYPE_URL};
    }

    @Override
    public boolean isClientFilter() {
      return true;
    }

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

    @Override
    public ConfigOrError<GcpAuthenticationConfig> 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<GcpAuthenticationConfig> 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 ResourceInvalidException {
      Audience audience;
      try {
        audience = any.unpack(Audience.class);
      } catch (InvalidProtocolBufferException ex) {
        throw new ResourceInvalidException("Invalid Resource in address proto", ex);
      }
      String url = audience.getUrl();
      if (url.isEmpty()) {
        throw new ResourceInvalidException(
            "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.MetadataRegistry.MetadataValueParser;
39	import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
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 {	1✔
51
52	static final String TYPE_URL =
53	"type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.GcpAuthnFilterConfig";
54
55	static final class Provider implements Filter.Provider {	1✔
56	@Override
57	public String[] typeUrls() {
58	return new String[]{TYPE_URL};	×
59	}
60
61	@Override
62	public boolean isClientFilter() {
63	return true;	1✔
64	}
65
66	@Override
67	public GcpAuthenticationFilter newInstance() {
68	return new GcpAuthenticationFilter();	×
69	}
70
71	@Override
72	public ConfigOrError<GcpAuthenticationConfig> parseFilterConfig(Message rawProtoMessage) {
73	GcpAuthnFilterConfig gcpAuthnProto;
74	if (!(rawProtoMessage instanceof Any)) {	1✔
75	return ConfigOrError.fromError("Invalid config type: " + rawProtoMessage.getClass());	1✔
76	}
77	Any anyMessage = (Any) rawProtoMessage;	1✔
78
79	try {
80	gcpAuthnProto = anyMessage.unpack(GcpAuthnFilterConfig.class);	1✔
81	} catch (InvalidProtocolBufferException e) {	×
82	return ConfigOrError.fromError("Invalid proto: " + e);	×
83	}	1✔
84
85	long cacheSize = 10;	1✔
86	// Validate cache_config
87	if (gcpAuthnProto.hasCacheConfig()) {	1✔
88	TokenCacheConfig cacheConfig = gcpAuthnProto.getCacheConfig();	1✔
89	cacheSize = cacheConfig.getCacheSize().getValue();	1✔
90	if (cacheSize == 0) {	1✔
91	return ConfigOrError.fromError(	1✔
92	"cache_config.cache_size must be greater than zero");
93	}
94	// LruCache's size is an int and briefly exceeds its maximum size before evicting entries
95	cacheSize = UnsignedLongs.min(cacheSize, Integer.MAX_VALUE - 1);	1✔
96	}
97
98	GcpAuthenticationConfig config = new GcpAuthenticationConfig((int) cacheSize);	1✔
99	return ConfigOrError.fromConfig(config);	1✔
100	}
101
102	@Override
103	public ConfigOrError<GcpAuthenticationConfig> parseFilterConfigOverride(
104	Message rawProtoMessage) {
105	return parseFilterConfig(rawProtoMessage);	×
106	}
107	}
108
109	@Nullable
110	@Override
111	public ClientInterceptor buildClientInterceptor(FilterConfig config,
112	@Nullable FilterConfig overrideConfig, ScheduledExecutorService scheduler) {
113
114	ComputeEngineCredentials credentials = ComputeEngineCredentials.create();	1✔
115	LruCache<String, CallCredentials> callCredentialsCache =	1✔
116	new LruCache<>(((GcpAuthenticationConfig) config).getCacheSize());	1✔
117	return new ClientInterceptor() {	1✔
118	@Override
119	public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
120	MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
121
122	/*String clusterName = callOptions.getOption(XdsAttributes.ATTR_CLUSTER_NAME);
123	if (clusterName == null) {
124	return next.newCall(method, callOptions);
125	}*/
126
127	// TODO: Fetch the CDS resource for the cluster.
128	// If the CDS resource is not available, fail the RPC with Status.UNAVAILABLE.
129
130	// TODO: Extract the audience from the CDS resource metadata.
131	// If the audience is not found or is in the wrong format, fail the RPC.
132	String audience = "TEST_AUDIENCE";	1✔
133
134	try {
135	CallCredentials existingCallCredentials = callOptions.getCredentials();	1✔
136	CallCredentials newCallCredentials =	1✔
137	getCallCredentials(callCredentialsCache, audience, credentials);	1✔
138	if (existingCallCredentials != null) {	1✔
139	callOptions = callOptions.withCallCredentials(	×
140	new CompositeCallCredentials(existingCallCredentials, newCallCredentials));
141	} else {
142	callOptions = callOptions.withCallCredentials(newCallCredentials);	1✔
143	}
144	}
145	catch (Exception e) {	×
146	// If we fail to attach CallCredentials due to any reason, return a FailingClientCall
147	return new FailingClientCall<>(Status.UNAUTHENTICATED	×
148	.withDescription("Failed to attach CallCredentials.")	×
149	.withCause(e));	×
150	}	1✔
151	return next.newCall(method, callOptions);	1✔
152	}
153	};
154	}
155
156	private CallCredentials getCallCredentials(LruCache<String, CallCredentials> cache,
157	String audience, ComputeEngineCredentials credentials) {
158
159	synchronized (cache) {	1✔
160	return cache.getOrInsert(audience, key -> {	1✔
161	IdTokenCredentials creds = IdTokenCredentials.newBuilder()	1✔
162	.setIdTokenProvider(credentials)	1✔
163	.setTargetAudience(audience)	1✔
164	.build();	1✔
165	return MoreCallCredentials.from(creds);	1✔
166	});
167	}
168	}
169
170	static final class GcpAuthenticationConfig implements FilterConfig {
171
172	private final int cacheSize;
173
174	public GcpAuthenticationConfig(int cacheSize) {	1✔
175	this.cacheSize = cacheSize;	1✔
176	}	1✔
177
178	public int getCacheSize() {
179	return cacheSize;	1✔
180	}
181
182	@Override
183	public String typeUrl() {
184	return GcpAuthenticationFilter.TYPE_URL;	×
185	}
186	}
187
188	/** An implementation of {@link ClientCall} that fails when started. */
189	private static final class FailingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
190
191	private final Status error;
192
193	public FailingClientCall(Status error) {	×
194	this.error = error;	×
195	}	×
196
197	@Override
198	public void start(ClientCall.Listener<RespT> listener, Metadata headers) {
199	listener.onClose(error, new Metadata());	×
200	}	×
201
202	@Override
203	public void request(int numMessages) {}	×
204
205	@Override
206	public void cancel(String message, Throwable cause) {}	×
207
208	@Override
209	public void halfClose() {}	×
210
211	@Override
212	public void sendMessage(ReqT message) {}	×
213	}
214
215	private static final class LruCache<K, V> {
216
217	private final Map<K, V> cache;
218
219	LruCache(int maxSize) {	1✔
220	this.cache = new LinkedHashMap<K, V>(	1✔
221	maxSize,
222	0.75f,
223	true) {	1✔
224	@Override
225	protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
226	return size() > maxSize;	1✔
227	}
228	};
229	}	1✔
230
231	V getOrInsert(K key, Function<K, V> create) {
232	return cache.computeIfAbsent(key, create);	1✔
233	}
234	}
235
236	static class AudienceMetadataParser implements MetadataValueParser {	1✔
237
238	@Override
239	public String getTypeUrl() {
240	return "type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.Audience";	1✔
241	}
242
243	@Override
244	public String parse(Any any) throws ResourceInvalidException {
245	Audience audience;
246	try {
247	audience = any.unpack(Audience.class);	1✔
248	} catch (InvalidProtocolBufferException ex) {	×
249	throw new ResourceInvalidException("Invalid Resource in address proto", ex);	×
250	}	1✔
251	String url = audience.getUrl();	1✔
252	if (url.isEmpty()) {	1✔
253	throw new ResourceInvalidException(	×
254	"Audience URL is empty. Metadata value must contain a valid URL.");
255	}
256	return url;	1✔
257	}
258	}
259	}

grpc / grpc-java / #19715

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