Skip to content

Commit 08fa030

Browse files
sauravzgsauravzg
committed
xds: Add ExtAuthzClientInterceptor and call buffering
Part 3 of the client-side ext_authz filter. Sits on top of the response handling PR. Introduces the ClientInterceptor that performs async ext_authz checks on outgoing RPCs. Because the authorization call is asynchronous, outgoing sendMessage() and halfClose() frames are buffered in ExtAuthzClientCall until a decision arrives. On allow, buffered operations are replayed with any header mutations applied. On deny, the call is terminated immediately. Key classes: - AuthzCallbackObserver — async listener on the authz gRPC stream that signals the buffering call on completion. - ExtAuthzClientCall — buffers outgoing frames while authz is pending; drains or cancels based on the decision. - MutatingClientCall — wraps the real call to inject header mutations from the authz response. - FailingClientCall / FailingCallWithTrailerMutations — immediate termination paths for denied or disabled-filter scenarios.
1 parent 9e6dfc4 commit 08fa030

10 files changed

Lines changed: 2219 additions & 0 deletions

‎xds/src/main/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserver.java‎

Lines changed: 172 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,172 @@
1+
/*
2+
* Copyright 2025 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.internal.extauthz;
18+
19+
import com.google.common.collect.ImmutableList;
20+
import io.envoyproxy.envoy.service.auth.v3.CheckResponse;
21+
import io.grpc.CallOptions;
22+
import io.grpc.Channel;
23+
import io.grpc.ClientCall;
24+
import io.grpc.Context;
25+
import io.grpc.Metadata;
26+
import io.grpc.MethodDescriptor;
27+
import io.grpc.Status;
28+
import io.grpc.internal.DelayedClientCall;
29+
import io.grpc.stub.StreamObserver;
30+
import io.grpc.xds.internal.grpcservice.HeaderValue;
31+
import io.grpc.xds.internal.headermutations.HeaderMutations;
32+
import io.grpc.xds.internal.headermutations.HeaderMutator;
33+
import io.grpc.xds.internal.headermutations.HeaderValueOption;
34+
import java.util.concurrent.Executor;
35+
36+
/**
37+
* A {@link StreamObserver} that handles the response from the external authorization service,
38+
* transitioning the delayed gRPC call based on the authorization decision.
39+
*/
40+
final class AuthzCallbackObserver<ReqT, RespT> implements StreamObserver<CheckResponse> {
41+
42+
private static final Metadata.Key<String> X_ENVOY_AUTH_FAILURE_MODE_ALLOWED =
43+
Metadata.Key.of("x-envoy-auth-failure-mode-allowed", Metadata.ASCII_STRING_MARSHALLER);
44+
45+
private final DelayedClientCall<ReqT, RespT> delayedCall;
46+
private final Channel next;
47+
private final MethodDescriptor<ReqT, RespT> method;
48+
private final CallOptions callOptions;
49+
private final Executor callExecutor;
50+
private final CheckResponseHandler responseHandler;
51+
private final HeaderMutator headerMutator;
52+
private final ExtAuthzConfig config;
53+
private final Context.CancellableContext authzContext;
54+
55+
AuthzCallbackObserver(
56+
DelayedClientCall<ReqT, RespT> delayedCall,
57+
Channel next,
58+
MethodDescriptor<ReqT, RespT> method,
59+
CallOptions callOptions,
60+
Executor callExecutor,
61+
CheckResponseHandler responseHandler,
62+
HeaderMutator headerMutator,
63+
ExtAuthzConfig config,
64+
Context.CancellableContext authzContext) {
65+
this.delayedCall = delayedCall;
66+
this.next = next;
67+
this.method = method;
68+
this.callOptions = callOptions;
69+
this.callExecutor = callExecutor;
70+
this.responseHandler = responseHandler;
71+
this.headerMutator = headerMutator;
72+
this.config = config;
73+
this.authzContext = authzContext;
74+
}
75+
76+
@Override
77+
public void onNext(CheckResponse value) {
78+
// Note on exception safety: handleResponse() internally catches
79+
// HeaderMutationDisallowedException and converts it to a DENY response.
80+
// Remaining calls (newCall, setCallAndDrain) use battle-tested
81+
// infrastructure with protobuf-guaranteed non-null defaults. An
82+
// uncaught RuntimeException here would propagate to the gRPC stub
83+
// framework, which handles observer stream cleanup. The authzContext
84+
// would remain uncancelled, but will be GC'd when this observer is
85+
// collected — acceptable since the authz RPC would have already
86+
// completed (we're inside onNext).
87+
AuthzResponse authzResponse = responseHandler.handleResponse(value);
88+
if (authzResponse.decision() == AuthzResponse.Decision.ALLOW) {
89+
ClientCall<ReqT, RespT> delegate = next.newCall(method, callOptions);
90+
ClientCall<ReqT, RespT> finalCall;
91+
if (isHeaderMutationsEmpty(authzResponse.requestHeaderMutations())
92+
&& isHeaderMutationsEmpty(authzResponse.responseHeaderMutations())) {
93+
finalCall = delegate;
94+
} else {
95+
finalCall = new MutatingClientCall<>(
96+
delegate,
97+
authzResponse.requestHeaderMutations(),
98+
authzResponse.responseHeaderMutations(),
99+
headerMutator);
100+
}
101+
setCallAndDrain(finalCall);
102+
} else {
103+
// Defensive programming: status is always present on DENY branches since every
104+
// deny() factory method requires a Status parameter. This orElseThrow satisfies
105+
// static analysis and documents the invariant.
106+
Status status = authzResponse.status().orElseThrow(
107+
() -> new IllegalArgumentException("DENY response missing status"));
108+
ClientCall<ReqT, RespT> failingCall;
109+
if (isHeaderMutationsEmpty(authzResponse.responseHeaderMutations())) {
110+
failingCall = new FailingClientCall<>(status);
111+
} else {
112+
failingCall = new FailingCallWithTrailerMutations<>(
113+
status,
114+
authzResponse.responseHeaderMutations(),
115+
headerMutator);
116+
}
117+
setCallAndDrain(failingCall);
118+
}
119+
}
120+
121+
@Override
122+
public void onError(Throwable t) {
123+
if (config.failureModeAllow()) {
124+
ClientCall<ReqT, RespT> delegate = next.newCall(method, callOptions);
125+
ClientCall<ReqT, RespT> finalCall;
126+
if (config.failureModeAllowHeaderAdd()) {
127+
HeaderMutations requestMutations = HeaderMutations.create(
128+
ImmutableList.of(HeaderValueOption.create(
129+
HeaderValue.create(X_ENVOY_AUTH_FAILURE_MODE_ALLOWED.name(), "true"),
130+
HeaderValueOption.HeaderAppendAction.APPEND_IF_EXISTS_OR_ADD)),
131+
ImmutableList.of());
132+
// TODO(sauravz): Consider using a static EMPTY_MUTATIONS constant
133+
// to avoid repeated allocation of empty HeaderMutations.
134+
HeaderMutations responseMutations =
135+
HeaderMutations.create(ImmutableList.of(), ImmutableList.of());
136+
finalCall = new MutatingClientCall<>(
137+
delegate, requestMutations, responseMutations, headerMutator);
138+
} else {
139+
finalCall = delegate;
140+
}
141+
setCallAndDrain(finalCall);
142+
authzContext.close();
143+
} else {
144+
Status statusToReturn = config.statusOnError().withCause(t);
145+
setCallAndDrain(new FailingClientCall<>(statusToReturn));
146+
authzContext.close();
147+
}
148+
}
149+
150+
@Override
151+
public void onCompleted() {
152+
// Defensive: if the authz server completed without sending a CheckResponse
153+
// (buggy server), resolve the DelayedClientCall with a failure and close
154+
// the authzContext. If onNext() already called setCall(), this is a no-op
155+
// since DelayedClientCall.setCall() ignores subsequent calls.
156+
Status status = config.statusOnError()
157+
.withDescription("Authz server completed without sending CheckResponse");
158+
setCallAndDrain(new FailingClientCall<>(status));
159+
authzContext.close();
160+
}
161+
162+
private void setCallAndDrain(ClientCall<ReqT, RespT> call) {
163+
Runnable drain = delayedCall.setCall(call);
164+
if (drain != null) {
165+
callExecutor.execute(drain);
166+
}
167+
}
168+
169+
private static boolean isHeaderMutationsEmpty(HeaderMutations mutations) {
170+
return mutations.headers().isEmpty() && mutations.headersToRemove().isEmpty();
171+
}
172+
}

‎xds/src/main/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCall.java‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
/*
2+
* Copyright 2025 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.internal.extauthz;
18+
19+
import com.google.common.annotations.VisibleForTesting;
20+
import com.google.protobuf.util.Timestamps;
21+
import io.envoyproxy.envoy.service.auth.v3.AuthorizationGrpc;
22+
import io.envoyproxy.envoy.service.auth.v3.CheckRequest;
23+
import io.grpc.CallOptions;
24+
import io.grpc.Channel;
25+
import io.grpc.ClientCall;
26+
import io.grpc.Context;
27+
import io.grpc.Deadline;
28+
import io.grpc.ForwardingClientCall;
29+
import io.grpc.Metadata;
30+
import io.grpc.MethodDescriptor;
31+
import io.grpc.internal.DelayedClientCall;
32+
import io.grpc.xds.internal.headermutations.HeaderMutator;
33+
import java.util.concurrent.Executor;
34+
import java.util.concurrent.ScheduledExecutorService;
35+
import javax.annotation.Nullable;
36+
37+
/**
38+
* A {@link ForwardingClientCall} that delegates to a lightweight {@link DelayedClientCall}
39+
* to buffer RPC requests safely while waiting for the asynchronous authorization decision.
40+
*/
41+
public final class ExtAuthzClientCall<ReqT, RespT> extends ForwardingClientCall<ReqT, RespT> {
42+
43+
private final Channel next;
44+
private final MethodDescriptor<ReqT, RespT> method;
45+
private final CallOptions callOptions;
46+
private final AuthorizationGrpc.AuthorizationStub authzStub;
47+
private final CheckRequestBuilder checkRequestBuilder;
48+
private final CheckResponseHandler responseHandler;
49+
private final HeaderMutator headerMutator;
50+
private final ExtAuthzConfig config;
51+
private final Executor callExecutor;
52+
private final Context.CancellableContext authzContext;
53+
54+
private final DelayedAuthzCall<ReqT, RespT> delegate;
55+
56+
public ExtAuthzClientCall(
57+
Executor executor,
58+
ScheduledExecutorService scheduler,
59+
CallOptions callOptions,
60+
Channel next,
61+
MethodDescriptor<ReqT, RespT> method,
62+
AuthorizationGrpc.AuthorizationStub authzStub,
63+
CheckRequestBuilder checkRequestBuilder,
64+
CheckResponseHandler responseHandler,
65+
HeaderMutator headerMutator,
66+
ExtAuthzConfig config) {
67+
this.callOptions = callOptions;
68+
this.next = next;
69+
this.method = method;
70+
this.authzStub = authzStub;
71+
this.checkRequestBuilder = checkRequestBuilder;
72+
this.responseHandler = responseHandler;
73+
this.headerMutator = headerMutator;
74+
this.config = config;
75+
this.callExecutor = executor;
76+
this.authzContext = Context.current().withCancellation();
77+
this.delegate = new DelayedAuthzCall<>(executor, scheduler, callOptions.getDeadline());
78+
}
79+
80+
@Override
81+
protected ClientCall<ReqT, RespT> delegate() {
82+
return delegate;
83+
}
84+
85+
@Override
86+
public void start(Listener<RespT> responseListener, Metadata headers) {
87+
// 1. Construct the check request FIRST before handoff (respects metadata thread safety)
88+
CheckRequest request = checkRequestBuilder.buildRequest(method, headers,
89+
Timestamps.fromMillis(System.currentTimeMillis()));
90+
91+
// 2. Start the delayed call (buffers headers and outbound payloads)
92+
super.start(responseListener, headers);
93+
94+
// 3. Trigger the async authorization check under the cancellable context
95+
authzContext.run(() -> {
96+
authzStub.check(request, new AuthzCallbackObserver<>(
97+
delegate, next, method, callOptions, callExecutor, responseHandler, headerMutator,
98+
config, authzContext));
99+
});
100+
}
101+
102+
@Override
103+
public void cancel(
104+
@Nullable String message, @Nullable Throwable cause) {
105+
authzContext.cancel(cause);
106+
super.cancel(message, cause);
107+
}
108+
109+
/**
110+
* Returns the cancellable context used for the authorization check.
111+
* Visible for testing.
112+
*/
113+
@VisibleForTesting
114+
Context.CancellableContext getAuthzContextForTest() {
115+
return authzContext;
116+
}
117+
118+
/**
119+
* A lightweight package-private DelayedClientCall subclass to expose
120+
* the protected constructor.
121+
*/
122+
private static final class DelayedAuthzCall<ReqT, RespT> extends DelayedClientCall<ReqT, RespT> {
123+
DelayedAuthzCall(Executor executor, ScheduledExecutorService scheduler, Deadline deadline) {
124+
super("ExtAuthzClientCall", executor, scheduler, deadline);
125+
}
126+
}
127+
}

‎xds/src/main/java/io/grpc/xds/internal/extauthz/FailingCallWithTrailerMutations.java‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
/*
2+
* Copyright 2025 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.internal.extauthz;
18+
19+
import io.grpc.ClientCall;
20+
import io.grpc.Metadata;
21+
import io.grpc.Status;
22+
import io.grpc.xds.internal.headermutations.HeaderMutations;
23+
import io.grpc.xds.internal.headermutations.HeaderMutator;
24+
25+
/**
26+
* A simple failing client call that lazily applies response mutations to trailers during start.
27+
*/
28+
final class FailingCallWithTrailerMutations<ReqT, RespT> extends ClientCall<ReqT, RespT> {
29+
private final Status status;
30+
private final HeaderMutations responseHeaderMutations;
31+
private final HeaderMutator headerMutator;
32+
33+
FailingCallWithTrailerMutations(
34+
Status status,
35+
HeaderMutations responseHeaderMutations,
36+
HeaderMutator headerMutator) {
37+
this.status = status;
38+
this.responseHeaderMutations = responseHeaderMutations;
39+
this.headerMutator = headerMutator;
40+
}
41+
42+
@Override
43+
public void start(Listener<RespT> responseListener, Metadata headers) {
44+
// Lazily allocate and apply response mutations to trailers copy on start!
45+
Metadata trailers = new Metadata();
46+
headerMutator.applyMutations(responseHeaderMutations, trailers);
47+
responseListener.onClose(status, trailers);
48+
}
49+
50+
@Override
51+
public void request(int numMessages) {}
52+
53+
@Override
54+
public void cancel(String message, Throwable cause) {}
55+
56+
@Override
57+
public void halfClose() {}
58+
59+
@Override
60+
public void sendMessage(ReqT message) {}
61+
}

0 commit comments

Comments
 (0)