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.
diff --git a/xds/src/main/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserver.java b/xds/src/main/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserver.java new file mode 100644 index 0000000..88091cd --- /dev/null +++ b/xds/src/main/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserver.java
@@ -0,0 +1,172 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import com.google.common.collect.ImmutableList; +import io.envoyproxy.envoy.service.auth.v3.CheckResponse; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.Context; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.internal.DelayedClientCall; +import io.grpc.stub.StreamObserver; +import io.grpc.xds.internal.grpcservice.HeaderValue; +import io.grpc.xds.internal.headermutations.HeaderMutations; +import io.grpc.xds.internal.headermutations.HeaderMutator; +import io.grpc.xds.internal.headermutations.HeaderValueOption; +import java.util.concurrent.Executor; + +/** + * A {@link StreamObserver} that handles the response from the external authorization service, + * transitioning the delayed gRPC call based on the authorization decision. + */ +final class AuthzCallbackObserver<ReqT, RespT> implements StreamObserver<CheckResponse> { + + private static final Metadata.Key<String> X_ENVOY_AUTH_FAILURE_MODE_ALLOWED = + Metadata.Key.of("x-envoy-auth-failure-mode-allowed", Metadata.ASCII_STRING_MARSHALLER); + + private final DelayedClientCall<ReqT, RespT> delayedCall; + private final Channel next; + private final MethodDescriptor<ReqT, RespT> method; + private final CallOptions callOptions; + private final Executor callExecutor; + private final CheckResponseHandler responseHandler; + private final HeaderMutator headerMutator; + private final ExtAuthzConfig config; + private final Context.CancellableContext authzContext; + + AuthzCallbackObserver( + DelayedClientCall<ReqT, RespT> delayedCall, + Channel next, + MethodDescriptor<ReqT, RespT> method, + CallOptions callOptions, + Executor callExecutor, + CheckResponseHandler responseHandler, + HeaderMutator headerMutator, + ExtAuthzConfig config, + Context.CancellableContext authzContext) { + this.delayedCall = delayedCall; + this.next = next; + this.method = method; + this.callOptions = callOptions; + this.callExecutor = callExecutor; + this.responseHandler = responseHandler; + this.headerMutator = headerMutator; + this.config = config; + this.authzContext = authzContext; + } + + @Override + public void onNext(CheckResponse value) { + // Note on exception safety: handleResponse() internally catches + // HeaderMutationDisallowedException and converts it to a DENY response. + // Remaining calls (newCall, setCallAndDrain) use battle-tested + // infrastructure with protobuf-guaranteed non-null defaults. An + // uncaught RuntimeException here would propagate to the gRPC stub + // framework, which handles observer stream cleanup. The authzContext + // would remain uncancelled, but will be GC'd when this observer is + // collected — acceptable since the authz RPC would have already + // completed (we're inside onNext). + AuthzResponse authzResponse = responseHandler.handleResponse(value); + if (authzResponse.decision() == AuthzResponse.Decision.ALLOW) { + ClientCall<ReqT, RespT> delegate = next.newCall(method, callOptions); + ClientCall<ReqT, RespT> finalCall; + if (isHeaderMutationsEmpty(authzResponse.requestHeaderMutations()) + && isHeaderMutationsEmpty(authzResponse.responseHeaderMutations())) { + finalCall = delegate; + } else { + finalCall = new MutatingClientCall<>( + delegate, + authzResponse.requestHeaderMutations(), + authzResponse.responseHeaderMutations(), + headerMutator); + } + setCallAndDrain(finalCall); + } else { + // Defensive programming: status is always present on DENY branches since every + // deny() factory method requires a Status parameter. This orElseThrow satisfies + // static analysis and documents the invariant. + Status status = authzResponse.status().orElseThrow( + () -> new IllegalArgumentException("DENY response missing status")); + ClientCall<ReqT, RespT> failingCall; + if (isHeaderMutationsEmpty(authzResponse.responseHeaderMutations())) { + failingCall = new FailingClientCall<>(status); + } else { + failingCall = new FailingCallWithTrailerMutations<>( + status, + authzResponse.responseHeaderMutations(), + headerMutator); + } + setCallAndDrain(failingCall); + } + } + + @Override + public void onError(Throwable t) { + if (config.failureModeAllow()) { + ClientCall<ReqT, RespT> delegate = next.newCall(method, callOptions); + ClientCall<ReqT, RespT> finalCall; + if (config.failureModeAllowHeaderAdd()) { + HeaderMutations requestMutations = HeaderMutations.create( + ImmutableList.of(HeaderValueOption.create( + HeaderValue.create(X_ENVOY_AUTH_FAILURE_MODE_ALLOWED.name(), "true"), + HeaderValueOption.HeaderAppendAction.APPEND_IF_EXISTS_OR_ADD)), + ImmutableList.of()); + // TODO(sauravz): Consider using a static EMPTY_MUTATIONS constant + // to avoid repeated allocation of empty HeaderMutations. + HeaderMutations responseMutations = + HeaderMutations.create(ImmutableList.of(), ImmutableList.of()); + finalCall = new MutatingClientCall<>( + delegate, requestMutations, responseMutations, headerMutator); + } else { + finalCall = delegate; + } + setCallAndDrain(finalCall); + authzContext.close(); + } else { + Status statusToReturn = config.statusOnError().withCause(t); + setCallAndDrain(new FailingClientCall<>(statusToReturn)); + authzContext.close(); + } + } + + @Override + public void onCompleted() { + // Defensive: if the authz server completed without sending a CheckResponse + // (buggy server), resolve the DelayedClientCall with a failure and close + // the authzContext. If onNext() already called setCall(), this is a no-op + // since DelayedClientCall.setCall() ignores subsequent calls. + Status status = config.statusOnError() + .withDescription("Authz server completed without sending CheckResponse"); + setCallAndDrain(new FailingClientCall<>(status)); + authzContext.close(); + } + + private void setCallAndDrain(ClientCall<ReqT, RespT> call) { + Runnable drain = delayedCall.setCall(call); + if (drain != null) { + callExecutor.execute(drain); + } + } + + private static boolean isHeaderMutationsEmpty(HeaderMutations mutations) { + return mutations.headers().isEmpty() && mutations.headersToRemove().isEmpty(); + } +}
diff --git a/xds/src/main/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCall.java b/xds/src/main/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCall.java new file mode 100644 index 0000000..b9280de --- /dev/null +++ b/xds/src/main/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCall.java
@@ -0,0 +1,127 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.util.Timestamps; +import io.envoyproxy.envoy.service.auth.v3.AuthorizationGrpc; +import io.envoyproxy.envoy.service.auth.v3.CheckRequest; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.Context; +import io.grpc.Deadline; +import io.grpc.ForwardingClientCall; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.internal.DelayedClientCall; +import io.grpc.xds.internal.headermutations.HeaderMutator; +import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledExecutorService; +import javax.annotation.Nullable; + +/** + * A {@link ForwardingClientCall} that delegates to a lightweight {@link DelayedClientCall} + * to buffer RPC requests safely while waiting for the asynchronous authorization decision. + */ +public final class ExtAuthzClientCall<ReqT, RespT> extends ForwardingClientCall<ReqT, RespT> { + + private final Channel next; + private final MethodDescriptor<ReqT, RespT> method; + private final CallOptions callOptions; + private final AuthorizationGrpc.AuthorizationStub authzStub; + private final CheckRequestBuilder checkRequestBuilder; + private final CheckResponseHandler responseHandler; + private final HeaderMutator headerMutator; + private final ExtAuthzConfig config; + private final Executor callExecutor; + private final Context.CancellableContext authzContext; + + private final DelayedAuthzCall<ReqT, RespT> delegate; + + public ExtAuthzClientCall( + Executor executor, + ScheduledExecutorService scheduler, + CallOptions callOptions, + Channel next, + MethodDescriptor<ReqT, RespT> method, + AuthorizationGrpc.AuthorizationStub authzStub, + CheckRequestBuilder checkRequestBuilder, + CheckResponseHandler responseHandler, + HeaderMutator headerMutator, + ExtAuthzConfig config) { + this.callOptions = callOptions; + this.next = next; + this.method = method; + this.authzStub = authzStub; + this.checkRequestBuilder = checkRequestBuilder; + this.responseHandler = responseHandler; + this.headerMutator = headerMutator; + this.config = config; + this.callExecutor = executor; + this.authzContext = Context.current().withCancellation(); + this.delegate = new DelayedAuthzCall<>(executor, scheduler, callOptions.getDeadline()); + } + + @Override + protected ClientCall<ReqT, RespT> delegate() { + return delegate; + } + + @Override + public void start(Listener<RespT> responseListener, Metadata headers) { + // 1. Construct the check request FIRST before handoff (respects metadata thread safety) + CheckRequest request = checkRequestBuilder.buildRequest(method, headers, + Timestamps.fromMillis(System.currentTimeMillis())); + + // 2. Start the delayed call (buffers headers and outbound payloads) + super.start(responseListener, headers); + + // 3. Trigger the async authorization check under the cancellable context + authzContext.run(() -> { + authzStub.check(request, new AuthzCallbackObserver<>( + delegate, next, method, callOptions, callExecutor, responseHandler, headerMutator, + config, authzContext)); + }); + } + + @Override + public void cancel( + @Nullable String message, @Nullable Throwable cause) { + authzContext.cancel(cause); + super.cancel(message, cause); + } + + /** + * Returns the cancellable context used for the authorization check. + * Visible for testing. + */ + @VisibleForTesting + Context.CancellableContext getAuthzContextForTest() { + return authzContext; + } + + /** + * A lightweight package-private DelayedClientCall subclass to expose + * the protected constructor. + */ + private static final class DelayedAuthzCall<ReqT, RespT> extends DelayedClientCall<ReqT, RespT> { + DelayedAuthzCall(Executor executor, ScheduledExecutorService scheduler, Deadline deadline) { + super("ExtAuthzClientCall", executor, scheduler, deadline); + } + } +}
diff --git a/xds/src/main/java/io/grpc/xds/internal/extauthz/FailingCallWithTrailerMutations.java b/xds/src/main/java/io/grpc/xds/internal/extauthz/FailingCallWithTrailerMutations.java new file mode 100644 index 0000000..407cc50 --- /dev/null +++ b/xds/src/main/java/io/grpc/xds/internal/extauthz/FailingCallWithTrailerMutations.java
@@ -0,0 +1,61 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import io.grpc.ClientCall; +import io.grpc.Metadata; +import io.grpc.Status; +import io.grpc.xds.internal.headermutations.HeaderMutations; +import io.grpc.xds.internal.headermutations.HeaderMutator; + +/** + * A simple failing client call that lazily applies response mutations to trailers during start. + */ +final class FailingCallWithTrailerMutations<ReqT, RespT> extends ClientCall<ReqT, RespT> { + private final Status status; + private final HeaderMutations responseHeaderMutations; + private final HeaderMutator headerMutator; + + FailingCallWithTrailerMutations( + Status status, + HeaderMutations responseHeaderMutations, + HeaderMutator headerMutator) { + this.status = status; + this.responseHeaderMutations = responseHeaderMutations; + this.headerMutator = headerMutator; + } + + @Override + public void start(Listener<RespT> responseListener, Metadata headers) { + // Lazily allocate and apply response mutations to trailers copy on start! + Metadata trailers = new Metadata(); + headerMutator.applyMutations(responseHeaderMutations, trailers); + responseListener.onClose(status, trailers); + } + + @Override + public void request(int numMessages) {} + + @Override + public void cancel(String message, Throwable cause) {} + + @Override + public void halfClose() {} + + @Override + public void sendMessage(ReqT message) {} +}
diff --git a/xds/src/main/java/io/grpc/xds/internal/extauthz/FailingClientCall.java b/xds/src/main/java/io/grpc/xds/internal/extauthz/FailingClientCall.java new file mode 100644 index 0000000..9e0f2e6 --- /dev/null +++ b/xds/src/main/java/io/grpc/xds/internal/extauthz/FailingClientCall.java
@@ -0,0 +1,57 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import io.grpc.ClientCall; +import io.grpc.Metadata; +import io.grpc.Status; +import javax.annotation.Nullable; + +/** + * A {@link ClientCall} that fails immediately upon starting. + */ +public final class FailingClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> { + + private final Status error; + + /** + * Creates a new call that will fail with the given error. + */ + public FailingClientCall(Status error) { + this.error = error; + } + + /** + * Immediately fails the call by calling {@link Listener#onClose}. + */ + @Override + public void start(Listener<RespT> responseListener, Metadata headers) { + responseListener.onClose(error, new Metadata()); + } + + @Override + public void request(int numMessages) {} + + @Override + public void cancel(@Nullable String message, @Nullable Throwable cause) {} + + @Override + public void halfClose() {} + + @Override + public void sendMessage(ReqT message) {} +}
diff --git a/xds/src/main/java/io/grpc/xds/internal/extauthz/MutatingClientCall.java b/xds/src/main/java/io/grpc/xds/internal/extauthz/MutatingClientCall.java new file mode 100644 index 0000000..8503530 --- /dev/null +++ b/xds/src/main/java/io/grpc/xds/internal/extauthz/MutatingClientCall.java
@@ -0,0 +1,70 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import io.grpc.ClientCall; +import io.grpc.ForwardingClientCall; +import io.grpc.ForwardingClientCallListener; +import io.grpc.Metadata; +import io.grpc.xds.internal.headermutations.HeaderMutations; +import io.grpc.xds.internal.headermutations.HeaderMutator; + +/** + * A {@link ForwardingClientCall} that unifies both request and response header mutations + * symmetrically. + */ +final class MutatingClientCall<ReqT, RespT> + extends ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT> { + + private final HeaderMutations requestHeaderMutations; + private final HeaderMutations responseHeaderMutations; + private final HeaderMutator headerMutator; + + MutatingClientCall( + ClientCall<ReqT, RespT> delegate, + HeaderMutations requestHeaderMutations, + HeaderMutations responseHeaderMutations, + HeaderMutator headerMutator) { + super(delegate); + this.requestHeaderMutations = requestHeaderMutations; + this.responseHeaderMutations = responseHeaderMutations; + this.headerMutator = headerMutator; + } + + @Override + public void start(Listener<RespT> responseListener, Metadata headers) { + // 1. Apply allowed request header mutations lazily before forwarding start! + headerMutator.applyMutations(requestHeaderMutations, headers); + + if (responseHeaderMutations.headers().isEmpty() + && responseHeaderMutations.headersToRemove().isEmpty()) { + super.start(responseListener, headers); + return; + } + + Listener<RespT> wrappedListener = new ForwardingClientCallListener + .SimpleForwardingClientCallListener<RespT>(responseListener) { + @Override + public void onHeaders(Metadata headers) { + // 2. Apply allowed response header mutations when headers are returned by the server + headerMutator.applyMutations(responseHeaderMutations, headers); + super.onHeaders(headers); + } + }; + super.start(wrappedListener, headers); + } +}
diff --git a/xds/src/test/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserverTest.java b/xds/src/test/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserverTest.java new file mode 100644 index 0000000..166e68d --- /dev/null +++ b/xds/src/test/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserverTest.java
@@ -0,0 +1,720 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableList; +import com.google.common.util.concurrent.MoreExecutors; +import io.envoyproxy.envoy.config.core.v3.HeaderValue; +import io.envoyproxy.envoy.config.core.v3.HeaderValueOption; +import io.envoyproxy.envoy.service.auth.v3.AuthorizationGrpc; +import io.envoyproxy.envoy.service.auth.v3.CheckRequest; +import io.envoyproxy.envoy.service.auth.v3.CheckResponse; +import io.envoyproxy.envoy.service.auth.v3.DeniedHttpResponse; +import io.envoyproxy.envoy.service.auth.v3.OkHttpResponse; +import io.envoyproxy.envoy.type.v3.HttpStatus; +import io.envoyproxy.envoy.type.v3.StatusCode; +import io.grpc.CallOptions; +import io.grpc.ChannelCredentials; +import io.grpc.Context; +import io.grpc.Deadline; +import io.grpc.ManagedChannel; +import io.grpc.Metadata; +import io.grpc.Server; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import io.grpc.ServerInterceptor; +import io.grpc.ServerServiceDefinition; +import io.grpc.Status; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.internal.DelayedClientCall; +import io.grpc.stub.StreamObserver; +import io.grpc.testing.protobuf.SimpleRequest; +import io.grpc.testing.protobuf.SimpleResponse; +import io.grpc.testing.protobuf.SimpleServiceGrpc; +import io.grpc.xds.client.ConfiguredChannelCredentials; +import io.grpc.xds.internal.Matchers; +import io.grpc.xds.internal.extauthz.ExtAuthzTestHelper.CapturingListener; +import io.grpc.xds.internal.grpcservice.GrpcServiceConfig; +import io.grpc.xds.internal.headermutations.HeaderMutationFilter; +import io.grpc.xds.internal.headermutations.HeaderMutations; +import io.grpc.xds.internal.headermutations.HeaderMutator; +import java.util.Optional; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +/** + * Behavioral unit tests for {@link AuthzCallbackObserver}. + */ +@RunWith(JUnit4.class) +public class AuthzCallbackObserverTest { + + @Rule + public final MockitoRule mocks = MockitoJUnit.rule(); + + @Mock + private AuthorizationGrpc.AuthorizationImplBase authzService; + + private final CheckResponseHandler responseHandler = + new CheckResponseHandler(new HeaderMutationFilter(Optional.empty())); + private final HeaderMutator headerMutator = HeaderMutator.create(); + private final ScheduledExecutorService scheduler = + Executors.newSingleThreadScheduledExecutor(); + + private Server server; + private ManagedChannel channel; + private volatile Metadata capturedBackendHeaders; + private volatile SimpleRequest capturedBackendMessage; + + @Before + public void setUp() throws Exception { + String serverName = InProcessServerBuilder.generateName(); + server = InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(authzService) + .intercept(new ServerInterceptor() { + @Override + public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall( + ServerCall<ReqT, RespT> call, Metadata headers, + ServerCallHandler<ReqT, RespT> next) { + if (call.getMethodDescriptor().getServiceName().equals( + SimpleServiceGrpc.getServiceDescriptor().getName())) { + capturedBackendHeaders = headers; + } + return next.startCall(call, headers); + } + }) + .addService(ServerServiceDefinition.builder( + SimpleServiceGrpc.getUnaryRpcMethod().getServiceName()) + .addMethod(SimpleServiceGrpc.getUnaryRpcMethod(), + (ServerCallHandler<SimpleRequest, SimpleResponse>) (call, headers) -> { + call.request(2); + return new ServerCall.Listener<SimpleRequest>() { + @Override + public void onMessage(SimpleRequest message) { + capturedBackendMessage = message; + call.sendHeaders(new Metadata()); + call.sendMessage(SimpleResponse.getDefaultInstance()); + call.close(Status.OK, new Metadata()); + } + }; + }) + .build()) + .build().start(); + channel = InProcessChannelBuilder.forName(serverName) + .directExecutor().build(); + } + + @After + public void tearDown() { + if (channel != null) { + channel.shutdownNow(); + } + if (server != null) { + server.shutdownNow(); + } + scheduler.shutdownNow(); + } + + @Test + public void allow_proxiesCallAndMessageToBackend() { + capturedBackendHeaders = null; + capturedBackendMessage = null; + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onNext(CheckResponse.newBuilder() + .setStatus(com.google.rpc.Status.newBuilder().setCode(0).build()) + .setOkResponse(OkHttpResponse.getDefaultInstance()) + .build()); + obs.onCompleted(); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("allow-payload").build(); + Metadata headers = new Metadata(); + headers.put(Metadata.Key.of("x-request-id", Metadata.ASCII_STRING_MARSHALLER), "req-123"); + + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, headers); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(capturedBackendHeaders).isNotNull(); + assertThat(capturedBackendHeaders.get( + Metadata.Key.of("x-request-id", Metadata.ASCII_STRING_MARSHALLER))) + .isEqualTo("req-123"); + assertThat(capturedBackendMessage).isEqualTo(request); + } + + @Test + public void allow_withHeaderMutations_backendReceivesMutatedHeadersAndMessage() { + capturedBackendHeaders = null; + capturedBackendMessage = null; + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onNext(CheckResponse.newBuilder() + .setStatus(com.google.rpc.Status.newBuilder().setCode(0).build()) + .setOkResponse(OkHttpResponse.newBuilder() + .addHeaders(HeaderValueOption.newBuilder() + .setHeader(HeaderValue.newBuilder().setKey("x-custom").setValue("injected"))) + .build()) + .build()); + obs.onCompleted(); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("mutated-header-payload").build(); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, new Metadata()); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(capturedBackendHeaders).isNotNull(); + assertThat(capturedBackendHeaders.get( + Metadata.Key.of("x-custom", Metadata.ASCII_STRING_MARSHALLER))) + .isEqualTo("injected"); + assertThat(capturedBackendMessage).isEqualTo(request); + } + + @Test + public void deny_returnsPermissionDeniedWithoutContactingBackend() { + capturedBackendHeaders = null; + capturedBackendMessage = null; + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onNext(CheckResponse.newBuilder() + .setStatus(com.google.rpc.Status.newBuilder() + .setCode(com.google.rpc.Code.PERMISSION_DENIED_VALUE)) + .setDeniedResponse(DeniedHttpResponse.newBuilder() + .setStatus(HttpStatus.newBuilder().setCode(StatusCode.Forbidden))) + .build()); + obs.onCompleted(); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("deny-payload").build(); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, new Metadata()); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(listener.getCloseStatus().getCode()) + .isEqualTo(Status.Code.PERMISSION_DENIED); + assertThat(capturedBackendHeaders).isNull(); + assertThat(capturedBackendMessage).isNull(); + } + + @Test + public void deny_withTrailerMutations_clientReceivesMutatedTrailers() { + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onNext(CheckResponse.newBuilder() + .setStatus(com.google.rpc.Status.newBuilder() + .setCode(com.google.rpc.Code.PERMISSION_DENIED_VALUE)) + .setDeniedResponse(DeniedHttpResponse.newBuilder() + .setStatus(HttpStatus.newBuilder().setCode(StatusCode.Forbidden)) + .addHeaders(HeaderValueOption.newBuilder() + .setHeader(HeaderValue.newBuilder().setKey("x-deny-reason").setValue("policy")))) + .build()); + obs.onCompleted(); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("deny-trailers-payload").build(); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, new Metadata()); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(listener.getCloseStatus().getCode()) + .isEqualTo(Status.Code.PERMISSION_DENIED); + assertThat(listener.getCloseTrailers().get( + Metadata.Key.of("x-deny-reason", Metadata.ASCII_STRING_MARSHALLER))) + .isEqualTo("policy"); + } + + @Test + public void authzError_failClosed_returnsConfiguredStatus() { + capturedBackendHeaders = null; + capturedBackendMessage = null; + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onError(Status.UNAVAILABLE.asRuntimeException()); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("error-closed-payload").build(); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, new Metadata()); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(listener.getCloseStatus().getCode()) + .isEqualTo(Status.Code.PERMISSION_DENIED); + assertThat(capturedBackendHeaders).isNull(); + assertThat(capturedBackendMessage).isNull(); + } + + @Test + public void authzError_failOpen_proxiesToBackendWithMessage() { + capturedBackendHeaders = null; + capturedBackendMessage = null; + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onError(Status.UNAVAILABLE.asRuntimeException()); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, + failOpenConfig(/*headerAdd=*/false), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("error-open-payload").build(); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, new Metadata()); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(capturedBackendHeaders).isNotNull(); + assertThat(capturedBackendMessage).isEqualTo(request); + } + + @Test + public void authzError_failOpenWithHeaderAdd_backendGetsFailureModeHeader() { + capturedBackendHeaders = null; + capturedBackendMessage = null; + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onError(Status.UNAVAILABLE.asRuntimeException()); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, + failOpenConfig(/*headerAdd=*/true), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("error-header-payload").build(); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, new Metadata()); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(capturedBackendHeaders).isNotNull(); + assertThat(capturedBackendHeaders.get( + Metadata.Key.of("x-envoy-auth-failure-mode-allowed", + Metadata.ASCII_STRING_MARSHALLER))) + .isEqualTo("true"); + assertThat(capturedBackendMessage).isEqualTo(request); + } + + @Test + public void buggyServer_completesWithoutResponse_failsCall() { + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onCompleted(); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("buggy-server-payload").build(); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, new Metadata()); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(listener.getCloseStatus().isOk()).isFalse(); + assertThat(listener.getCloseStatus().getCode()) + .isEqualTo(Status.Code.PERMISSION_DENIED); + assertThat(listener.getCloseStatus().getCause()).isNotNull(); + assertThat(listener.getCloseStatus().getCause().getMessage()) + .contains("server cancelled stream"); + } + + @Test + public void allow_withResponseOnlyHeaderMutations_backendReceivesMutatedHeadersAndMessage() { + capturedBackendHeaders = null; + capturedBackendMessage = null; + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onNext(CheckResponse.newBuilder() + .setStatus(com.google.rpc.Status.newBuilder().setCode(0).build()) + .setOkResponse(OkHttpResponse.newBuilder() + .addResponseHeadersToAdd(HeaderValueOption.newBuilder() + .setHeader(HeaderValue.newBuilder().setKey("x-resp-header").setValue("val")))) + .build()); + obs.onCompleted(); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("resp-header-payload").build(); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, new Metadata()); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(capturedBackendHeaders).isNotNull(); + assertThat(capturedBackendMessage).isEqualTo(request); + } + + @Test + public void allow_withHeadersToRemoveOnly_backendReceivesMutatedHeaders() { + capturedBackendHeaders = null; + capturedBackendMessage = null; + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onNext(CheckResponse.newBuilder() + .setStatus(com.google.rpc.Status.newBuilder().setCode(0).build()) + .setOkResponse(OkHttpResponse.newBuilder() + .addHeadersToRemove("header-to-remove")) + .build()); + obs.onCompleted(); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + SimpleRequest request = + SimpleRequest.newBuilder().setRequestMessage("remove-header-payload").build(); + Metadata headers = new Metadata(); + headers.put(Metadata.Key.of("header-to-remove", Metadata.ASCII_STRING_MARSHALLER), "val"); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + delayedCall.start(listener, headers); + delayedCall.sendMessage(request); + delayedCall.halfClose(); + delayedCall.request(1); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(capturedBackendHeaders).isNotNull(); + assertThat(capturedBackendHeaders.containsKey( + Metadata.Key.of("header-to-remove", Metadata.ASCII_STRING_MARSHALLER))).isFalse(); + assertThat(capturedBackendMessage).isEqualTo(request); + } + + @Test(expected = IllegalArgumentException.class) + public void deny_withMissingStatus_throwsIllegalArgumentException() { + CheckResponseHandler mockHandler = mock(CheckResponseHandler.class); + AuthzResponse fakeAuthzResponse = new AuthzResponse() { + @Override + public Decision decision() { + return Decision.DENY; + } + + @Override + public Optional<Status> status() { + return Optional.empty(); + } + + @Override + public HeaderMutations requestHeaderMutations() { + return HeaderMutations.create(ImmutableList.of(), ImmutableList.of()); + } + + @Override + public HeaderMutations responseHeaderMutations() { + return HeaderMutations.create(ImmutableList.of(), ImmutableList.of()); + } + }; + when(mockHandler.handleResponse(any())).thenReturn(fakeAuthzResponse); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + mockHandler, headerMutator, failClosedConfig(), authzCtx); + + observer.onNext(CheckResponse.getDefaultInstance()); + } + + @Test + public void allow_whenDelayedCallNotStarted_setCallReturnsNull() { + doAnswer(invocation -> { + StreamObserver<CheckResponse> obs = invocation.getArgument(1); + obs.onNext(CheckResponse.newBuilder() + .setStatus(com.google.rpc.Status.newBuilder().setCode(0).build()) + .setOkResponse(OkHttpResponse.getDefaultInstance()) + .build()); + obs.onCompleted(); + return null; + }).when(authzService).check(any(), any()); + + TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall = + new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null); + Context.CancellableContext authzCtx = Context.current().withCancellation(); + AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer = + new AuthzCallbackObserver<>( + delayedCall, channel, + SimpleServiceGrpc.getUnaryRpcMethod(), + CallOptions.DEFAULT, + MoreExecutors.directExecutor(), + responseHandler, headerMutator, failClosedConfig(), authzCtx); + + authzCtx.run(() -> { + AuthorizationGrpc.newStub(channel) + .check(CheckRequest.getDefaultInstance(), observer); + }); + + assertThat(capturedBackendHeaders).isNull(); + } + + private static final class TestDelayedCall<ReqT, RespT> + extends DelayedClientCall<ReqT, RespT> { + TestDelayedCall( + java.util.concurrent.Executor executor, + ScheduledExecutorService scheduler, + Deadline deadline) { + super("TestDelayedCall", executor, scheduler, deadline); + } + } + + private static ExtAuthzConfig failClosedConfig() { + GrpcServiceConfig.GoogleGrpcConfig googleGrpc = + GrpcServiceConfig.GoogleGrpcConfig.builder() + .target("test-cluster") + .configuredChannelCredentials( + ConfiguredChannelCredentials.create( + mock(ChannelCredentials.class), + mock(ConfiguredChannelCredentials.ChannelCredsConfig.class))) + .build(); + GrpcServiceConfig grpcServiceConfig = + GrpcServiceConfig.builder() + .googleGrpc(googleGrpc) + .initialMetadata(ImmutableList.of()) + .build(); + return ExtAuthzConfig.builder() + .grpcService(grpcServiceConfig) + .failureModeAllow(false) + .failureModeAllowHeaderAdd(false) + .includePeerCertificate(false) + .denyAtDisable(false) + .filterEnabled(Matchers.FractionMatcher.create(100, 100)) + .statusOnError(Status.PERMISSION_DENIED) + .build(); + } + + private static ExtAuthzConfig failOpenConfig(boolean headerAdd) { + GrpcServiceConfig.GoogleGrpcConfig googleGrpc = + GrpcServiceConfig.GoogleGrpcConfig.builder() + .target("test-cluster") + .configuredChannelCredentials( + ConfiguredChannelCredentials.create( + mock(ChannelCredentials.class), + mock(ConfiguredChannelCredentials.ChannelCredsConfig.class))) + .build(); + GrpcServiceConfig grpcServiceConfig = + GrpcServiceConfig.builder() + .googleGrpc(googleGrpc) + .initialMetadata(ImmutableList.of()) + .build(); + return ExtAuthzConfig.builder() + .grpcService(grpcServiceConfig) + .failureModeAllow(true) + .failureModeAllowHeaderAdd(headerAdd) + .includePeerCertificate(false) + .denyAtDisable(false) + .filterEnabled(Matchers.FractionMatcher.create(100, 100)) + .statusOnError(Status.PERMISSION_DENIED) + .build(); + } +}
diff --git a/xds/src/test/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCallTest.java b/xds/src/test/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCallTest.java new file mode 100644 index 0000000..ffe0ea2 --- /dev/null +++ b/xds/src/test/java/io/grpc/xds/internal/extauthz/ExtAuthzClientCallTest.java
@@ -0,0 +1,630 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableList; +import io.envoyproxy.envoy.service.auth.v3.AuthorizationGrpc; +import io.envoyproxy.envoy.service.auth.v3.CheckRequest; +import io.envoyproxy.envoy.service.auth.v3.CheckResponse; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ChannelCredentials; +import io.grpc.Context; +import io.grpc.ManagedChannel; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Server; +import io.grpc.ServerCallHandler; +import io.grpc.ServerInterceptor; +import io.grpc.ServerServiceDefinition; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import io.grpc.testing.protobuf.SimpleRequest; +import io.grpc.testing.protobuf.SimpleResponse; +import io.grpc.testing.protobuf.SimpleServiceGrpc; +import io.grpc.xds.client.ConfiguredChannelCredentials; +import io.grpc.xds.internal.Matchers; +import io.grpc.xds.internal.extauthz.ExtAuthzTestHelper.CapturingListener; +import io.grpc.xds.internal.grpcservice.GrpcServiceConfig; +import io.grpc.xds.internal.grpcservice.HeaderValue; +import io.grpc.xds.internal.headermutations.HeaderMutations; +import io.grpc.xds.internal.headermutations.HeaderMutator; +import io.grpc.xds.internal.headermutations.HeaderValueOption; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.ArgumentMatchers; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +/** Unit tests for {@link ExtAuthzClientCall}. */ +@RunWith(JUnit4.class) +public class ExtAuthzClientCallTest { + + @Rule + public final MockitoRule mocks = MockitoJUnit.rule(); + + @Mock + private CheckRequestBuilder mockCheckRequestBuilder; + @Mock + private CheckResponseHandler mockResponseHandler; + @Mock + private HeaderMutator mockHeaderMutator; + @Mock + private AuthorizationGrpc.AuthorizationImplBase authzService; + + private final ScheduledExecutorService scheduler = + Executors.newSingleThreadScheduledExecutor(); + private Server server; + private ManagedChannel channel; + private volatile Metadata lastBackendHeaders; + private ExtAuthzConfig config; + private CallOptions callOptions; + + @Before + public void setUp() throws Exception { + // Single in-process server hosting both authz and backend services + String serverName = + InProcessServerBuilder.generateName(); + server = InProcessServerBuilder + .forName(serverName) + .directExecutor() + .intercept(new ServerInterceptor() { + @Override + public <ReqT, RespT> io.grpc.ServerCall.Listener<ReqT> interceptCall( + io.grpc.ServerCall<ReqT, RespT> call, Metadata headers, + ServerCallHandler<ReqT, RespT> next) { + lastBackendHeaders = headers; + return next.startCall(call, headers); + } + }) + .addService(authzService) + .addService(ServerServiceDefinition.builder( + SimpleServiceGrpc.getUnaryRpcMethod().getServiceName()) + .addMethod(SimpleServiceGrpc.getUnaryRpcMethod(), + (ServerCallHandler<SimpleRequest, SimpleResponse>) (call, headers) -> { + call.sendHeaders(new Metadata()); + call.close(Status.OK, new Metadata()); + return new io.grpc.ServerCall.Listener<SimpleRequest>() {}; + }) + .build()) + .build() + .start(); + channel = InProcessChannelBuilder + .forName(serverName) + .directExecutor() + .build(); + + config = buildConfig(); + callOptions = CallOptions.DEFAULT.withExecutor(scheduler); + when(mockCheckRequestBuilder.buildRequest( + any(MethodDescriptor.class), any(Metadata.class), + any(com.google.protobuf.Timestamp.class))) + .thenReturn(CheckRequest.getDefaultInstance()); + } + + @After + public void tearDown() { + if (server != null) { + server.shutdownNow(); + } + if (channel != null) { + channel.shutdownNow(); + } + scheduler.shutdownNow(); + } + + @Test + public void cancel_cancelsAuthzContext() { + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = + createCall(); + + Context.CancellableContext authzContext = + call.getAuthzContextForTest(); + assertThat(authzContext.isCancelled()).isFalse(); + + call.cancel("user cancelled", null); + + assertThat(authzContext.isCancelled()).isTrue(); + } + + @Test + public void cancel_whileCheckInFlight_cancelsContext() { + CountDownLatch checkCalled = new CountDownLatch(1); + doAnswer(invocation -> { + checkCalled.countDown(); + // Don't respond to simulate in-flight check + return null; + }).when(authzService) + .check(any(CheckRequest.class), + ArgumentMatchers.any()); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = + createCall(); + + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + call.start(listener, new Metadata()); + + try { + assertThat( + checkCalled.await(5, TimeUnit.SECONDS)).isTrue(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + + Context.CancellableContext authzContext = + call.getAuthzContextForTest(); + assertThat(authzContext.isCancelled()).isFalse(); + + call.cancel("client cancelled", null); + + assertThat(authzContext.isCancelled()).isTrue(); + } + + @Test + public void start_runsCheckUnderAuthzContext() { + Context[] capturedContext = new Context[1]; + doAnswer(invocation -> { + capturedContext[0] = Context.current(); + StreamObserver<CheckResponse> observer = + invocation.getArgument(1); + observer.onNext(CheckResponse.getDefaultInstance()); + observer.onCompleted(); + return null; + }).when(authzService) + .check(any(CheckRequest.class), + ArgumentMatchers.any()); + + HeaderMutations emptyMutations = HeaderMutations.create( + ImmutableList.of(), ImmutableList.of()); + AuthzResponse authzResponse = + AuthzResponse.allow(emptyMutations) + .setResponseHeaderMutations(emptyMutations) + .build(); + when(mockResponseHandler.handleResponse( + any(CheckResponse.class))).thenReturn(authzResponse); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = + createCall(); + + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + call.start(listener, new Metadata()); + + // The check was called under the authzContext + assertThat(capturedContext[0]).isNotNull(); + } + + @Test + public void start_triggersCheckRequest() { + doAnswer(invocation -> { + return null; + }).when(authzService) + .check(any(CheckRequest.class), + ArgumentMatchers.any()); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = + createCall(); + + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + call.start(listener, new Metadata()); + + verify(authzService) + .check(any(CheckRequest.class), + ArgumentMatchers.any()); + } + + @Test + public void authzCheckCompletes_contextIsCancelled() + throws InterruptedException { + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = + invocation.getArgument(1); + observer.onNext(CheckResponse.getDefaultInstance()); + observer.onCompleted(); + checkDone.countDown(); + return null; + }).when(authzService) + .check(any(CheckRequest.class), + ArgumentMatchers.any()); + + HeaderMutations emptyMutations = HeaderMutations.create( + ImmutableList.of(), ImmutableList.of()); + AuthzResponse authzResponse = + AuthzResponse.allow(emptyMutations) + .setResponseHeaderMutations(emptyMutations) + .build(); + when(mockResponseHandler.handleResponse( + any(CheckResponse.class))).thenReturn(authzResponse); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = + createCall(); + + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + call.start(listener, new Metadata()); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + + // After observer.onNext the context should be cancelled + Context.CancellableContext authzContext = + call.getAuthzContextForTest(); + assertThat(authzContext.isCancelled()).isTrue(); + } + + @Test + public void authzCheckError_contextIsCancelled() + throws InterruptedException { + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = + invocation.getArgument(1); + observer.onError( + Status.UNAVAILABLE.asRuntimeException()); + checkDone.countDown(); + return null; + }).when(authzService) + .check(any(CheckRequest.class), + ArgumentMatchers.any()); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = + createCall(); + + CapturingListener<SimpleResponse> listener = + new CapturingListener<>(); + call.start(listener, new Metadata()); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + + // After observer.onError the context should be cancelled + Context.CancellableContext authzContext = + call.getAuthzContextForTest(); + assertThat(authzContext.isCancelled()).isTrue(); + } + + @Test + public void authzCheckCompletes_allow_forwardsCall() throws Exception { + lastBackendHeaders = null; + + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = invocation.getArgument(1); + observer.onNext(CheckResponse.getDefaultInstance()); + observer.onCompleted(); + checkDone.countDown(); + return null; + }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any()); + + HeaderMutations emptyMutations = HeaderMutations.create( + ImmutableList.of(), ImmutableList.of()); + AuthzResponse authzResponse = AuthzResponse.allow(emptyMutations) + .setResponseHeaderMutations(emptyMutations) + .build(); + when(mockResponseHandler.handleResponse(any(CheckResponse.class))).thenReturn(authzResponse); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = createCall( + com.google.common.util.concurrent.MoreExecutors.directExecutor(), + channel, config); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + Metadata headers = new Metadata(); + call.start(listener, headers); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(lastBackendHeaders).isNotNull(); + verify(mockHeaderMutator, org.mockito.Mockito.never()).applyMutations(any(), any()); + } + + @Test + public void authzCheckCompletes_allowWithMutations_wrapsCall() throws Exception { + lastBackendHeaders = null; + + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = invocation.getArgument(1); + observer.onNext(CheckResponse.getDefaultInstance()); + observer.onCompleted(); + checkDone.countDown(); + return null; + }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any()); + + HeaderMutations requestMutations = HeaderMutations.create( + ImmutableList.of(HeaderValueOption.create( + HeaderValue.create("foo", "bar"), + HeaderValueOption.HeaderAppendAction.APPEND_IF_EXISTS_OR_ADD)), + ImmutableList.of()); + HeaderMutations responseMutations = HeaderMutations.create( + ImmutableList.of(), ImmutableList.of()); + AuthzResponse authzResponse = AuthzResponse.allow(requestMutations) + .setResponseHeaderMutations(responseMutations) + .build(); + when(mockResponseHandler.handleResponse(any(CheckResponse.class))).thenReturn(authzResponse); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = createCall( + com.google.common.util.concurrent.MoreExecutors.directExecutor(), + channel, config); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + Metadata headers = new Metadata(); + call.start(listener, headers); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(lastBackendHeaders).isNotNull(); + verify(mockHeaderMutator).applyMutations(ArgumentMatchers.eq(requestMutations), any()); + } + + @Test + public void authzCheckCompletes_deny_failsCall() throws Exception { + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = invocation.getArgument(1); + observer.onNext(CheckResponse.getDefaultInstance()); + observer.onCompleted(); + checkDone.countDown(); + return null; + }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any()); + + HeaderMutations emptyMutations = HeaderMutations.create( + ImmutableList.of(), ImmutableList.of()); + AuthzResponse authzResponse = AuthzResponse.deny(Status.PERMISSION_DENIED) + .setResponseHeaderMutations(emptyMutations) + .build(); + when(mockResponseHandler.handleResponse(any(CheckResponse.class))).thenReturn(authzResponse); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = createCall( + com.google.common.util.concurrent.MoreExecutors.directExecutor(), channel, config); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + call.start(listener, new Metadata()); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(listener.getCloseStatus()).isEqualTo(Status.PERMISSION_DENIED); + verify(mockHeaderMutator, org.mockito.Mockito.never()).applyMutations(any(), any()); + } + + @Test + public void authzCheckCompletes_denyWithMutations_failsCallWithTrailers() throws Exception { + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = invocation.getArgument(1); + observer.onNext(CheckResponse.getDefaultInstance()); + observer.onCompleted(); + checkDone.countDown(); + return null; + }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any()); + + HeaderMutations responseMutations = HeaderMutations.create( + ImmutableList.of(HeaderValueOption.create( + HeaderValue.create("foo", "bar"), + HeaderValueOption.HeaderAppendAction.APPEND_IF_EXISTS_OR_ADD)), + ImmutableList.of()); + AuthzResponse authzResponse = AuthzResponse.deny(Status.PERMISSION_DENIED) + .setResponseHeaderMutations(responseMutations) + .build(); + when(mockResponseHandler.handleResponse(any(CheckResponse.class))).thenReturn(authzResponse); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = createCall( + com.google.common.util.concurrent.MoreExecutors.directExecutor(), channel, config); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + call.start(listener, new Metadata()); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + verify(mockHeaderMutator).applyMutations(ArgumentMatchers.eq(responseMutations), any()); + assertThat(listener.getCloseStatus()).isEqualTo(Status.PERMISSION_DENIED); + } + + @Test + public void authzCheckError_failClosed_failsCall() throws Exception { + ExtAuthzConfig failClosedConfig = ExtAuthzConfig.builder() + .grpcService(config.grpcService()) + .failureModeAllow(false) + .failureModeAllowHeaderAdd(false) + .includePeerCertificate(false) + .denyAtDisable(false) + .filterEnabled(Matchers.FractionMatcher.create(100, 100)) + .statusOnError(Status.PERMISSION_DENIED) + .build(); + + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = invocation.getArgument(1); + observer.onError(Status.UNAVAILABLE.asRuntimeException()); + checkDone.countDown(); + return null; + }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any()); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = createCall( + com.google.common.util.concurrent.MoreExecutors.directExecutor(), channel, + failClosedConfig); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + call.start(listener, new Metadata()); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(listener.getCloseStatus().getCode()).isEqualTo(Status.Code.PERMISSION_DENIED); + assertThat(listener.getCloseStatus().getCause()).isInstanceOf(StatusRuntimeException.class); + assertThat( + ((StatusRuntimeException) listener.getCloseStatus().getCause()) + .getStatus().getCode()) + .isEqualTo(Status.Code.UNAVAILABLE); + } + + @Test + public void authzCheckError_failOpen_forwardsCall() throws Exception { + ExtAuthzConfig failOpenConfig = ExtAuthzConfig.builder() + .grpcService(config.grpcService()) + .failureModeAllow(true) + .failureModeAllowHeaderAdd(false) + .includePeerCertificate(false) + .denyAtDisable(false) + .filterEnabled(Matchers.FractionMatcher.create(100, 100)) + .statusOnError(Status.PERMISSION_DENIED) + .build(); + + lastBackendHeaders = null; + + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = invocation.getArgument(1); + observer.onError(Status.UNAVAILABLE.asRuntimeException()); + checkDone.countDown(); + return null; + }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any()); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = createCall( + com.google.common.util.concurrent.MoreExecutors.directExecutor(), + channel, failOpenConfig); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + Metadata headers = new Metadata(); + call.start(listener, headers); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(lastBackendHeaders).isNotNull(); + verify(mockHeaderMutator, org.mockito.Mockito.never()).applyMutations(any(), any()); + } + + @Test + public void authzCheckError_failOpenWithHeaderAdd_wrapsCall() throws Exception { + ExtAuthzConfig failOpenWithHeaderConfig = ExtAuthzConfig.builder() + .grpcService(config.grpcService()) + .failureModeAllow(true) + .failureModeAllowHeaderAdd(true) + .includePeerCertificate(false) + .denyAtDisable(false) + .filterEnabled(Matchers.FractionMatcher.create(100, 100)) + .statusOnError(Status.PERMISSION_DENIED) + .build(); + + lastBackendHeaders = null; + + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = invocation.getArgument(1); + observer.onError(Status.UNAVAILABLE.asRuntimeException()); + checkDone.countDown(); + return null; + }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any()); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = createCall( + com.google.common.util.concurrent.MoreExecutors.directExecutor(), + channel, failOpenWithHeaderConfig); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + Metadata headers = new Metadata(); + call.start(listener, headers); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(lastBackendHeaders).isNotNull(); + } + + @Test + public void drain_executesOnCallExecutor() throws Exception { + java.util.concurrent.Executor mockExecutor = mock(java.util.concurrent.Executor.class); + doAnswer(invocation -> { + Runnable r = invocation.getArgument(0); + r.run(); + return null; + }).when(mockExecutor).execute(any(Runnable.class)); + + lastBackendHeaders = null; + + CountDownLatch checkDone = new CountDownLatch(1); + doAnswer(invocation -> { + StreamObserver<CheckResponse> observer = invocation.getArgument(1); + observer.onNext(CheckResponse.getDefaultInstance()); + observer.onCompleted(); + checkDone.countDown(); + return null; + }).when(authzService).check(any(CheckRequest.class), ArgumentMatchers.any()); + + HeaderMutations emptyMutations = HeaderMutations.create( + ImmutableList.of(), ImmutableList.of()); + AuthzResponse authzResponse = AuthzResponse.allow(emptyMutations) + .setResponseHeaderMutations(emptyMutations) + .build(); + when(mockResponseHandler.handleResponse(any(CheckResponse.class))).thenReturn(authzResponse); + + ExtAuthzClientCall<SimpleRequest, SimpleResponse> call = + createCall(mockExecutor, channel, config); + CapturingListener<SimpleResponse> listener = new CapturingListener<>(); + call.start(listener, new Metadata()); + + assertThat(checkDone.await(5, TimeUnit.SECONDS)).isTrue(); + verify(mockExecutor, org.mockito.Mockito.atLeastOnce()).execute(any(Runnable.class)); + } + + private ExtAuthzClientCall<SimpleRequest, SimpleResponse> + createCall() { + return createCall(channel, config); + } + + private ExtAuthzClientCall<SimpleRequest, SimpleResponse> + createCall(Channel next, ExtAuthzConfig config) { + return createCall(callOptions.getExecutor(), next, config); + } + + private ExtAuthzClientCall<SimpleRequest, SimpleResponse> + createCall( + java.util.concurrent.Executor executor, Channel next, ExtAuthzConfig config) { + return new ExtAuthzClientCall<>( + executor, scheduler, callOptions, + next, + SimpleServiceGrpc.getUnaryRpcMethod(), + AuthorizationGrpc.newStub(channel), + mockCheckRequestBuilder, mockResponseHandler, + mockHeaderMutator, config); + } + + private static ExtAuthzConfig buildConfig() { + GrpcServiceConfig.GoogleGrpcConfig googleGrpc = + GrpcServiceConfig.GoogleGrpcConfig.builder() + .target("test-cluster") + .configuredChannelCredentials( + ConfiguredChannelCredentials.create( + mock(ChannelCredentials.class), + mock(ConfiguredChannelCredentials + .ChannelCredsConfig.class))) + .build(); + GrpcServiceConfig grpcServiceConfig = + GrpcServiceConfig.builder() + .googleGrpc(googleGrpc) + .initialMetadata(ImmutableList.of()) + .build(); + return ExtAuthzConfig.builder() + .grpcService(grpcServiceConfig) + .failureModeAllow(false) + .failureModeAllowHeaderAdd(false) + .includePeerCertificate(false) + .denyAtDisable(false) + .filterEnabled( + Matchers.FractionMatcher.create(100, 100)) + .statusOnError(Status.PERMISSION_DENIED) + .build(); + } +}
diff --git a/xds/src/test/java/io/grpc/xds/internal/extauthz/FailingCallWithTrailerMutationsTest.java b/xds/src/test/java/io/grpc/xds/internal/extauthz/FailingCallWithTrailerMutationsTest.java new file mode 100644 index 0000000..0be19e5 --- /dev/null +++ b/xds/src/test/java/io/grpc/xds/internal/extauthz/FailingCallWithTrailerMutationsTest.java
@@ -0,0 +1,108 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.verify; + +import com.google.common.collect.ImmutableList; +import io.grpc.Metadata; +import io.grpc.Status; +import io.grpc.xds.internal.extauthz.ExtAuthzTestHelper.CapturingListener; +import io.grpc.xds.internal.headermutations.HeaderMutations; +import io.grpc.xds.internal.headermutations.HeaderMutator; +import io.grpc.xds.internal.headermutations.HeaderValueOption; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +/** + * Unit tests for {@link FailingCallWithTrailerMutations}. + */ +@RunWith(JUnit4.class) +public class FailingCallWithTrailerMutationsTest { + + @Rule + public final MockitoRule mocks = MockitoJUnit.rule(); + + @Mock + private HeaderMutator mockHeaderMutator; + + private HeaderMutations responseMutations; + private Status failStatus; + private FailingCallWithTrailerMutations<Void, Void> failingCall; + + @Before + public void setUp() { + HeaderValueOption option = HeaderValueOption.create( + io.grpc.xds.internal.grpcservice.HeaderValue.create("test-key", "test-value"), + HeaderValueOption.HeaderAppendAction.APPEND_IF_EXISTS_OR_ADD); + responseMutations = HeaderMutations.create(ImmutableList.of(option), ImmutableList.of()); + failStatus = Status.PERMISSION_DENIED.withDescription("authz denied"); + + failingCall = new FailingCallWithTrailerMutations<>( + failStatus, + responseMutations, + mockHeaderMutator); + } + + @Test + public void start_appliesTrailersMutationsAndClosesListener() { + Metadata headers = new Metadata(); + CapturingListener<Void> listener = new CapturingListener<>(); + + // Trigger start on the failing call + failingCall.start(listener, headers); + + // 1. Verify that trailers are lazily allocated and mutations are applied to them + verify(mockHeaderMutator).applyMutations(eq(responseMutations), any(Metadata.class)); + + // 2. Verify that the listener was closed with the correct status + assertThat(listener.getCloseStatus()).isEqualTo(failStatus); + assertThat(listener.getCloseTrailers()).isNotNull(); + } + + @Test + public void outboundMethods_areNoOpsAndSafe() { + // Verify that calling any outbound transport methods does not throw any exceptions + failingCall.sendMessage(null); + failingCall.request(1); + failingCall.halfClose(); + failingCall.cancel("cancel", null); + } + + @Test + public void start_calledTwice_firesOnCloseTwice() { + CapturingListener<Void> firstListener = new CapturingListener<>(); + Metadata headers1 = new Metadata(); + failingCall.start(firstListener, headers1); + assertThat(firstListener.getCloseStatus()).isEqualTo(failStatus); + + // Call start again with a different listener + CapturingListener<Void> secondListener = new CapturingListener<>(); + Metadata headers2 = new Metadata(); + failingCall.start(secondListener, headers2); + assertThat(secondListener.getCloseStatus()).isEqualTo(failStatus); + } +}
diff --git a/xds/src/test/java/io/grpc/xds/internal/extauthz/FailingClientCallTest.java b/xds/src/test/java/io/grpc/xds/internal/extauthz/FailingClientCallTest.java new file mode 100644 index 0000000..72439c3 --- /dev/null +++ b/xds/src/test/java/io/grpc/xds/internal/extauthz/FailingClientCallTest.java
@@ -0,0 +1,88 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import static com.google.common.truth.Truth.assertThat; + +import io.grpc.Metadata; +import io.grpc.NoopClientCall; +import io.grpc.Status; +import javax.annotation.Nullable; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Unit tests for {@link FailingClientCall}. */ +@RunWith(JUnit4.class) +public class FailingClientCallTest { + + private final CapturingListener<Object> listener = new CapturingListener<>(); + + @Test + public void startCallsOnClose() { + Status error = Status.UNAVAILABLE.withDescription("test error"); + FailingClientCall<Object, Object> call = new FailingClientCall<>(error); + Metadata metadata = new Metadata(); + call.start(listener, metadata); + + assertThat(listener.getCloseStatus()).isEqualTo(error); + assertThat(listener.getCloseTrailers()).isNotNull(); + assertThat(listener.getCloseTrailers().keys()).isEmpty(); + } + + @Test + public void otherMethodsAreNoOps() { + Status error = Status.UNAVAILABLE.withDescription("test error"); + FailingClientCall<Object, Object> call = new FailingClientCall<>(error); + Metadata metadata = new Metadata(); + + call.start(listener, metadata); // Must call start first + + call.request(1); + call.cancel("message", new RuntimeException("cause")); + call.halfClose(); + call.sendMessage(new Object()); + + // Only one onClose should have been called (from start), no additional callbacks. + assertThat(listener.getCloseStatus()).isEqualTo(error); + assertThat(listener.getCloseTrailers()).isNotNull(); + assertThat(listener.getCloseTrailers().keys()).isEmpty(); + } + + /** A capturing listener that records the status and trailers from {@link #onClose}. */ + private static final class CapturingListener<T> + extends NoopClientCall.NoopClientCallListener<T> { + @Nullable private Status closeStatus; + @Nullable private Metadata closeTrailers; + + @Override + public void onClose(Status status, Metadata trailers) { + this.closeStatus = status; + this.closeTrailers = trailers; + } + + @Nullable + Status getCloseStatus() { + return closeStatus; + } + + @Nullable + Metadata getCloseTrailers() { + return closeTrailers; + } + } +}
diff --git a/xds/src/test/java/io/grpc/xds/internal/extauthz/MutatingClientCallTest.java b/xds/src/test/java/io/grpc/xds/internal/extauthz/MutatingClientCallTest.java new file mode 100644 index 0000000..45cfbb5 --- /dev/null +++ b/xds/src/test/java/io/grpc/xds/internal/extauthz/MutatingClientCallTest.java
@@ -0,0 +1,186 @@ +/* + * Copyright 2025 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.internal.extauthz; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.verify; + +import com.google.common.collect.ImmutableList; +import io.grpc.Metadata; +import io.grpc.Status; +import io.grpc.xds.internal.extauthz.ExtAuthzTestHelper.CapturingClientCall; +import io.grpc.xds.internal.extauthz.ExtAuthzTestHelper.CapturingListener; +import io.grpc.xds.internal.headermutations.HeaderMutations; +import io.grpc.xds.internal.headermutations.HeaderMutator; +import io.grpc.xds.internal.headermutations.HeaderValueOption; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +/** + * Unit tests for {@link MutatingClientCall}. + */ +@RunWith(JUnit4.class) +public class MutatingClientCallTest { + + @Rule + public final MockitoRule mocks = MockitoJUnit.rule(); + + private final CapturingClientCall<Void, Void> mockDelegateCall = new CapturingClientCall<>(); + @Mock + private HeaderMutator mockHeaderMutator; + private final CapturingListener<Void> mockResponseListener = new CapturingListener<>(); + + private HeaderMutations requestMutations; + private HeaderMutations responseMutations; + private MutatingClientCall<Void, Void> mutatingCall; + + @Before + public void setUp() { + HeaderValueOption option = HeaderValueOption.create( + io.grpc.xds.internal.grpcservice.HeaderValue.create("test-key", "test-value"), + HeaderValueOption.HeaderAppendAction.APPEND_IF_EXISTS_OR_ADD); + requestMutations = HeaderMutations.create(ImmutableList.of(option), ImmutableList.of()); + responseMutations = HeaderMutations.create(ImmutableList.of(option), ImmutableList.of()); + + mutatingCall = new MutatingClientCall<>( + mockDelegateCall, + requestMutations, + responseMutations, + mockHeaderMutator); + } + + @Test + public void start_appliesRequestMutationsSymmetrically() { + Metadata headers = new Metadata(); + + // Trigger start + mutatingCall.start(mockResponseListener, headers); + + // 1. Verify request mutations are applied lazily to headers before forwarding start + verify(mockHeaderMutator).applyMutations(eq(requestMutations), eq(headers)); + + // 2. Verify delegate start was triggered with the mutated headers and captured listener + assertThat(mockDelegateCall.isStarted()).isTrue(); + assertThat(mockDelegateCall.getHeaders()).isSameInstanceAs(headers); + assertThat(mockDelegateCall.getListener()).isNotNull(); + } + + @Test + public void onHeaders_appliesResponseMutationsSymmetrically() { + Metadata headers = new Metadata(); + mutatingCall.start(mockResponseListener, headers); + + // Get the wrapped listener that was passed to delegate + assertThat(mockDelegateCall.getListener()).isNotNull(); + + // Trigger onHeaders on captured listener + Metadata responseHeaders = new Metadata(); + mockDelegateCall.getListener().onHeaders(responseHeaders); + + // 1. Verify response mutations are applied lazily to response headers + verify(mockHeaderMutator).applyMutations(eq(responseMutations), eq(responseHeaders)); + + // 2. Verify response headers were forwarded to real listener + assertThat(mockResponseListener.getHeaders()).isSameInstanceAs(responseHeaders); + } + + @Test + public void listenerCallbacks_areDelegatedCorrectly() { + Metadata headers = new Metadata(); + mutatingCall.start(mockResponseListener, headers); + + assertThat(mockDelegateCall.getListener()).isNotNull(); + + // Test onMessage delegation + mockDelegateCall.getListener().onMessage(null); + assertThat(mockResponseListener.getMessages()).containsExactly((Void) null); + + // Test onReady delegation + mockDelegateCall.getListener().onReady(); + assertThat(mockResponseListener.isOnReadyCalled()).isTrue(); + + // Test onClose delegation + Status expectedStatus = Status.OK; + Metadata expectedTrailers = new Metadata(); + mockDelegateCall.getListener().onClose(expectedStatus, expectedTrailers); + assertThat(mockResponseListener.getCloseStatus()).isEqualTo(expectedStatus); + assertThat(mockResponseListener.getCloseTrailers()).isSameInstanceAs(expectedTrailers); + } + + @Test + public void emptyRequestMutations_nonEmptyResponseMutations_wrapsListener() { + HeaderMutations emptyRequest = + HeaderMutations.create(ImmutableList.of(), ImmutableList.of()); + CapturingClientCall<Void, Void> delegate = new CapturingClientCall<>(); + MutatingClientCall<Void, Void> call = new MutatingClientCall<>( + delegate, emptyRequest, responseMutations, mockHeaderMutator); + + Metadata headers = new Metadata(); + CapturingListener<Void> listener = new CapturingListener<>(); + call.start(listener, headers); + + // Request mutations should still be called (even if empty) + verify(mockHeaderMutator).applyMutations(eq(emptyRequest), eq(headers)); + + // Verify delegate was started with a wrapped listener + assertThat(delegate.isStarted()).isTrue(); + assertThat(delegate.getHeaders()).isSameInstanceAs(headers); + assertThat(delegate.getListener()).isNotNull(); + + // Trigger onHeaders and verify response mutations are applied + Metadata responseHeaders = new Metadata(); + delegate.getListener().onHeaders(responseHeaders); + verify(mockHeaderMutator).applyMutations( + eq(responseMutations), eq(responseHeaders)); + assertThat(listener.getHeaders()).isSameInstanceAs(responseHeaders); + } + + @Test + public void start_emptyResponseMutations_headersPassThroughUnmodified() { + HeaderMutations emptyResponseMutations = + HeaderMutations.create(ImmutableList.of(), ImmutableList.of()); + CapturingClientCall<Void, Void> delegate = new CapturingClientCall<>(); + MutatingClientCall<Void, Void> call = new MutatingClientCall<>( + delegate, requestMutations, emptyResponseMutations, mockHeaderMutator); + + CapturingListener<Void> listener = new CapturingListener<>(); + Metadata headers = new Metadata(); + call.start(listener, headers); + + // Request mutations should be applied + verify(mockHeaderMutator).applyMutations(eq(requestMutations), eq(headers)); + + // Simulate receiving response headers from the server + Metadata.Key<String> responseKey = + Metadata.Key.of("x-response-header", Metadata.ASCII_STRING_MARSHALLER); + Metadata responseHeaders = new Metadata(); + responseHeaders.put(responseKey, "original-value"); + delegate.getListener().onHeaders(responseHeaders); + + // Verify: headers pass through UNMODIFIED — no response mutation applied + assertThat(listener.getHeaders().get(responseKey)).isEqualTo("original-value"); + // headerMutator should NOT have been called again for response mutations + org.mockito.Mockito.verifyNoMoreInteractions(mockHeaderMutator); + } +}