diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderServerTransport.java b/binder/src/main/java/io/grpc/binder/internal/BinderServerTransport.java index 784d833bdf5..379183f45c5 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderServerTransport.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderServerTransport.java @@ -127,6 +127,11 @@ void notifyShutdown(Status status) { // Nothing to do. } + @Override + void notifyTerminatedUnlocked() { + BinderTransportSecurity.notifyTerminatedUnlocked(getAttributes()); + } + @Override @GuardedBy("this") void notifyTerminated() { diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java b/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java index 2b7aa97bfd9..c70cadc8749 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java @@ -251,6 +251,8 @@ final boolean isReady() { @GuardedBy("this") abstract void notifyTerminated(); + void notifyTerminatedUnlocked() {} + void releaseExecutors() { executorServicePool.returnObject(scheduledExecutorService); } @@ -335,6 +337,8 @@ final void shutdownInternal(Status shutdownStatus, boolean forceTerminate) { future.cancel(false); // No effect if already isDone(). } + notifyTerminatedUnlocked(); + synchronized (this) { notifyTerminated(); } diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java index 6f95ef8a83c..7215bbee669 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransportSecurity.java @@ -16,10 +16,14 @@ package io.grpc.binder.internal; +import static com.google.common.util.concurrent.Futures.nonCancellationPropagating; + +import com.google.common.annotations.VisibleForTesting; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.SettableFuture; import com.google.errorprone.annotations.CheckReturnValue; import io.grpc.Attributes; import io.grpc.Internal; @@ -83,6 +87,24 @@ public static void attachAuthAttrs( .set(GrpcAttributes.ATTR_SECURITY_LEVEL, SecurityLevel.PRIVACY_AND_INTEGRITY); } + /** + * Informs this module that the transport with the specified 'attributes' is terminating. + * + *

Any resources allocated by this module will be released and no further resources will be + * allocated. Any ongoing background work will be canceled and no further background work will be + * initiated. + * + *

This method completes futures and therefore may execute arbitrary listener code on a + * potentially direct Executor. To avoid deadlock, callers must not hold any locks. + */ + @Internal + public static void notifyTerminatedUnlocked(Attributes attributes) { + TransportAuthorizationState state = attributes.get(TRANSPORT_AUTHORIZATION_STATE); + if (state != null) { + state.notifyTerminatedUnlocked(); + } + } + /** * Intercepts server calls and ensures they're authorized before allowing them to proceed. * Authentication state is fetched from the call attributes, inherited from the transport. @@ -161,12 +183,16 @@ public void onFailure(Throwable t) { * Maintains the authorization state for a single transport instance. This class lives for the * lifetime of a single transport. */ - private static final class TransportAuthorizationState { + @VisibleForTesting + static final class TransportAuthorizationState { private final int uid; private final ServerPolicyChecker serverPolicyChecker; + // Holds *all* pending policy check futures and *certain* complete ones that we want to cache. private final ConcurrentHashMap> serviceAuthorization; private final Executor executor; + private volatile boolean isTerminated; + /** * @param executor used for calling into the application. Must outlive the transport. */ @@ -182,43 +208,76 @@ private static final class TransportAuthorizationState { @CheckReturnValue ListenableFuture checkAuthorization(MethodDescriptor method) { String serviceName = method.getServiceName(); - // Only cache decisions if the method can be sampled for tracing, - // which is true for all generated methods. Otherwise, programmatically - // created methods could cause this cache to grow unbounded. - boolean useCache = method.isSampledToLocalTracing(); - if (useCache) { - @Nullable ListenableFuture authorization = serviceAuthorization.get(serviceName); - if (authorization != null) { - // Authorization check exists and is a pending or successful future (even if for a - // failed authorization). - return authorization; + @Nullable + ListenableFuture pendingOrCachedAuthResult = serviceAuthorization.get(serviceName); + if (pendingOrCachedAuthResult != null) { + return nonCancellationPropagating(pendingOrCachedAuthResult); + } + + SettableFuture newPendingAuthResult = SettableFuture.create(); + ListenableFuture checkThenActRaceWinner = + serviceAuthorization.putIfAbsent(serviceName, newPendingAuthResult); + if (checkThenActRaceWinner != null) { + // Another thread running this method must have also just saw no entry for serviceName, then + // beat us to calling putIfAbsent(). We can only track one check at a time so share theirs. + return nonCancellationPropagating(checkThenActRaceWinner); + } + + // We only check isTerminated *after* the new future is visible to other threads in + // serviceAuthorization. In case of a race with a simultaneous call to notifyTerminated(), + // better to harmlessly cancel the new future twice rather than not cancel it at all. + if (!isTerminated) { + // If notifyTerminated() already cancelled newPendingAuthResult, setFuture() will forward + // that cancellation to its argument so it's safe to ignore the return value here. + try { + newPendingAuthResult.setFuture( + serverPolicyChecker.checkAuthorizationForServiceAsync(uid, serviceName)); + } catch (Exception e) { // Not just RuntimeException! Handle the "sneaky" checked case too. + newPendingAuthResult.setException(e); } + } else { + newPendingAuthResult.cancel(false); } - // Under high load, this may trigger a large number of concurrent authorization checks that - // perform essentially the same work and have the potential of exhausting the resources they - // depend on. This was a non-issue in the past with synchronous policy checks due to the - // fixed-size nature of the thread pool this method runs under. - // - // TODO(10669): evaluate if there should be at most a single pending authorization check per - // (uid, serviceName) pair at any given time. - ListenableFuture authorization = - serverPolicyChecker.checkAuthorizationForServiceAsync(uid, serviceName); - if (useCache) { - serviceAuthorization.putIfAbsent(serviceName, authorization); - Futures.addCallback( - authorization, - new FutureCallback() { - @Override - public void onSuccess(Status result) {} - - @Override - public void onFailure(Throwable t) { - serviceAuthorization.remove(serviceName, authorization); + + Futures.addCallback( + newPendingAuthResult, + new FutureCallback() { + @Override + public void onSuccess(Status result) { + // Auth checks can be expensive so we want to cache the results. But programmatically + // created service names could cause the cache to grow without bound. Conservatively, + // we only cache results for codegen service names as there can't be too many of them. + if (!method.isSampledToLocalTracing()) { + serviceAuthorization.remove(serviceName, newPendingAuthResult); } - }, - MoreExecutors.directExecutor()); + } + + @Override + public void onFailure(Throwable t) { + // Not simply a non-OK auth result but a failure to return any decision at all. Never + // cache these so that if the caller retries, we'll retry the auth check as well. + serviceAuthorization.remove(serviceName, newPendingAuthResult); + } + }, + MoreExecutors.directExecutor()); + + return nonCancellationPropagating(newPendingAuthResult); + } + + /** + * After this method returns, every future ever returned by a prior or concurrent call to + * checkAuthorization() is guaranteed to be complete. Every future returned by subsequent call + * to checkAuthorization() is also guaranteed to be complete. + */ + void notifyTerminatedUnlocked() { + // Any entries added to serviceAuthorization *after* this assignment will be immediately + // canceled by the adding thread in checkAuthorization(). + isTerminated = true; + + // Cancel any entries added to serviceAuthorization *before* the volatile assignment above. + for (ListenableFuture authResult : serviceAuthorization.values()) { + authResult.cancel(false); // No-op for cached results (already done). } - return authorization; } } diff --git a/binder/src/test/java/io/grpc/binder/RobolectricBinderSecurityTest.java b/binder/src/test/java/io/grpc/binder/RobolectricBinderSecurityTest.java index ffd1d89e69c..19c187bf2eb 100644 --- a/binder/src/test/java/io/grpc/binder/RobolectricBinderSecurityTest.java +++ b/binder/src/test/java/io/grpc/binder/RobolectricBinderSecurityTest.java @@ -19,6 +19,8 @@ import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.truth.Truth.assertThat; import static com.google.common.util.concurrent.MoreExecutors.directExecutor; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.fail; import static org.robolectric.Shadows.shadowOf; import android.app.Application; @@ -48,6 +50,7 @@ import io.grpc.stub.ServerCalls; import java.io.IOException; import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.CancellationException; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -154,25 +157,43 @@ public void tearDown() { @Test public void testAsyncServerSecurityPolicy_failed_returnsFailureStatus() throws Exception { ListenableFuture status = makeCall(); - statusesToSet.take().set(Status.ALREADY_EXISTS); + awaitNext(statusesToSet).set(Status.ALREADY_EXISTS); - assertThat(status.get().getCode()).isEqualTo(Status.Code.ALREADY_EXISTS); + assertThat(awaitResult(status).getCode()).isEqualTo(Status.Code.ALREADY_EXISTS); } @Test public void testAsyncServerSecurityPolicy_failedFuture_failsWithCodeInternal() throws Exception { ListenableFuture status = makeCall(); - statusesToSet.take().setException(new IllegalStateException("oops")); + awaitNext(statusesToSet).setException(new IllegalStateException("oops")); - assertThat(status.get().getCode()).isEqualTo(Status.Code.INTERNAL); + assertThat(awaitResult(status).getCode()).isEqualTo(Status.Code.INTERNAL); } @Test public void testAsyncServerSecurityPolicy_allowed_returnsOkStatus() throws Exception { ListenableFuture status = makeCall(); - statusesToSet.take().set(Status.OK); + awaitNext(statusesToSet).set(Status.OK); - assertThat(status.get().getCode()).isEqualTo(Status.Code.OK); + assertThat(awaitResult(status).getCode()).isEqualTo(Status.Code.OK); + } + + @Test + public void testAsyncServerSecurityPolicy_shutdownNow_cancelsAuthFutures() throws Exception { + ListenableFuture callResult = makeCall(); + SettableFuture authResultFuture = awaitNext(statusesToSet); + + channel.shutdownNow(); + boolean terminationResult = channel.awaitTermination(10, SECONDS); + assertThat(terminationResult).isTrue(); + + try { + Status authResult = awaitResult(authResultFuture); + fail("Expected authResultFuture cancellation but got " + authResult); + } catch (CancellationException expected) { + } + assertThat(authResultFuture.isCancelled()).isTrue(); + assertThat(awaitResult(callResult).getCode()).isEqualTo(Status.Code.UNAVAILABLE); } private ListenableFuture makeCall() { @@ -187,6 +208,18 @@ private ListenableFuture makeCall() { directExecutor()); } + private static T awaitNext(java.util.concurrent.BlockingQueue queue) throws Exception { + T item = queue.poll(10, SECONDS); + if (item == null) { + throw new java.util.concurrent.TimeoutException("Queue timed out waiting for item"); + } + return item; + } + + private static T awaitResult(java.util.concurrent.Future future) throws Exception { + return future.get(10, SECONDS); + } + private static MethodDescriptor getMethodDescriptor() { MethodDescriptor.Marshaller marshaller = ProtoLiteUtils.marshaller(Empty.getDefaultInstance()); diff --git a/binder/src/test/java/io/grpc/binder/internal/TransportAuthorizationStateTest.java b/binder/src/test/java/io/grpc/binder/internal/TransportAuthorizationStateTest.java new file mode 100644 index 00000000000..f08235b7801 --- /dev/null +++ b/binder/src/test/java/io/grpc/binder/internal/TransportAuthorizationStateTest.java @@ -0,0 +1,230 @@ +/* + * Copyright 2024 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.binder.internal; + +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.SettableFuture; +import com.google.protobuf.Empty; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.binder.internal.BinderTransportSecurity.ServerPolicyChecker; +import io.grpc.binder.internal.BinderTransportSecurity.TransportAuthorizationState; +import io.grpc.protobuf.lite.ProtoLiteUtils; +import java.util.NoSuchElementException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.robolectric.RobolectricTestRunner; + +@RunWith(RobolectricTestRunner.class) +public final class TransportAuthorizationStateTest { + + private static final int UID = 12345; + private static final String NONCODEGEN_SERVICE_NAME = "test.noncodegen.service"; + private static final MethodDescriptor NONCODEGEN_METHOD = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName(NONCODEGEN_SERVICE_NAME + "/NonCodegenMethod") + .setRequestMarshaller(ProtoLiteUtils.marshaller(Empty.getDefaultInstance())) + .setResponseMarshaller(ProtoLiteUtils.marshaller(Empty.getDefaultInstance())) + .setSampledToLocalTracing(false) + .build(); + + private static final String CODEGEN_SERVICE_NAME = "test.codegen.service"; + private static final MethodDescriptor CODEGEN_METHOD = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName(CODEGEN_SERVICE_NAME + "/CodegenMethod") + .setRequestMarshaller(ProtoLiteUtils.marshaller(Empty.getDefaultInstance())) + .setResponseMarshaller(ProtoLiteUtils.marshaller(Empty.getDefaultInstance())) + .setSampledToLocalTracing(true) + .build(); + + private ExecutorService executor; + private FakeServerPolicyChecker fakePolicyChecker; + private TransportAuthorizationState authState; + + @Before + public void setUp() { + executor = Executors.newSingleThreadExecutor(); + fakePolicyChecker = new FakeServerPolicyChecker(); + authState = new TransportAuthorizationState(UID, fakePolicyChecker, executor); + } + + @After + public void tearDown() throws Exception { + assertThat(executor.shutdownNow()).isEmpty(); + assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + } + + @Test + public void checkAuthorization_deduplicatesSimultaneousNonCodegenMethods() throws Exception { + ListenableFuture authResult1 = authState.checkAuthorization(NONCODEGEN_METHOD); + assertThat(authResult1.isDone()).isFalse(); + + ListenableFuture authResult2 = authState.checkAuthorization(NONCODEGEN_METHOD); + assertThat(authResult2.isDone()).isFalse(); + + // The fake policy checker was only invoked ONCE thanks to deduplication. + // Completing that single underlying auth check satisfies both futures. + fakePolicyChecker.takeNextAuthRequestOrDie().set(Status.OK); + + assertThat(authResult1.get()).isEqualTo(Status.OK); + assertThat(authResult2.get()).isEqualTo(Status.OK); + assertThat(fakePolicyChecker.statusesToSet).isEmpty(); + + // Because it's a non-codegen method, the auth result should not be cached. + ListenableFuture authResult3 = authState.checkAuthorization(NONCODEGEN_METHOD); + assertThat(authResult3.isDone()).isFalse(); + + fakePolicyChecker.takeNextAuthRequestOrDie().set(Status.PERMISSION_DENIED); + assertThat(authResult3.get()).isEqualTo(Status.PERMISSION_DENIED); + } + + @Test + public void checkAuthorization_cachesSimultaneousCodegenMethods() throws Exception { + ListenableFuture authResult1 = authState.checkAuthorization(CODEGEN_METHOD); + assertThat(authResult1.isDone()).isFalse(); + + ListenableFuture authResult2 = authState.checkAuthorization(CODEGEN_METHOD); + assertThat(authResult2.isDone()).isFalse(); + + // The fake policy checker was only invoked ONCE thanks to deduplication. + // Completing that single underlying auth check satisfies both futures. + fakePolicyChecker.takeNextAuthRequestOrDie().set(Status.OK); + + assertThat(authResult1.get()).isEqualTo(Status.OK); + assertThat(authResult2.get()).isEqualTo(Status.OK); + assertThat(fakePolicyChecker.statusesToSet).isEmpty(); + + // Because it's a codegen method, the auth result should be cached for the life of the object. + ListenableFuture authResult3 = authState.checkAuthorization(CODEGEN_METHOD); + assertThat(authResult3.isDone()).isTrue(); + assertThat(authResult3.get()).isEqualTo(Status.OK); + assertThat(fakePolicyChecker.statusesToSet).isEmpty(); + } + + @Test + public void checkAuthorization_cancellation_doesNotPropagateToUnderlyingCheck() throws Exception { + ListenableFuture authResult1 = authState.checkAuthorization(CODEGEN_METHOD); + ListenableFuture authResult2 = authState.checkAuthorization(CODEGEN_METHOD); + + // Cancel the first future. + authResult1.cancel(true); + + assertThat(authResult1.isCancelled()).isTrue(); + // The second future should NOT be cancelled, because the cancellation shouldn't propagate + // to the underlying shared future. + assertThat(authResult2.isCancelled()).isFalse(); + + // Completing the underlying auth check satisfies the non-cancelled future. + fakePolicyChecker.takeNextAuthRequestOrDie().set(Status.OK); + + assertThat(authResult2.get()).isEqualTo(Status.OK); + } + + @Test + public void checkAuthorization_failedFuture_notCached() throws Exception { + ListenableFuture authResult1 = authState.checkAuthorization(CODEGEN_METHOD); + assertThat(authResult1.isDone()).isFalse(); + + fakePolicyChecker.takeNextAuthRequestOrDie().setException(new IllegalStateException("oops")); + + ExecutionException exception = assertThrows(ExecutionException.class, () -> authResult1.get()); + assertThat(exception).hasCauseThat().isInstanceOf(IllegalStateException.class); + assertThat(fakePolicyChecker.statusesToSet).isEmpty(); + + // Failed futures must not be cached, even for codegen methods. + ListenableFuture authResult2 = authState.checkAuthorization(CODEGEN_METHOD); + assertThat(authResult2.isDone()).isFalse(); + + fakePolicyChecker.takeNextAuthRequestOrDie().set(Status.OK); + assertThat(authResult2.get()).isEqualTo(Status.OK); + } + + @Test + public void notifyTerminatedUnlocked_cancelsPendingAuthFutures() { + ListenableFuture authResult = authState.checkAuthorization(CODEGEN_METHOD); + assertThat(authResult.isDone()).isFalse(); + + authState.notifyTerminatedUnlocked(); + + assertThat(authResult.isCancelled()).isTrue(); + } + + @Test + public void checkAuthorization_afterTermination_returnsCancelledFuture() { + authState.notifyTerminatedUnlocked(); + + ListenableFuture authResult = authState.checkAuthorization(CODEGEN_METHOD); + + assertThat(authResult.isCancelled()).isTrue(); + assertThat(fakePolicyChecker.statusesToSet).isEmpty(); // fakePolicyChecker never called. + } + + @Test + public void checkAuthorization_synchronousException_doesNotLeaveStrandedFuture() + throws Exception { + IllegalStateException syncException = new IllegalStateException("ouch"); + fakePolicyChecker.syncExceptionsToThrow.add(syncException); + + // The synchronous exception is safely returned as a failed future. + ListenableFuture authResult1 = authState.checkAuthorization(CODEGEN_METHOD); + ExecutionException ee = assertThrows(ExecutionException.class, authResult1::get); + assertThat(ee).hasCauseThat().isSameInstanceAs(syncException); + + // Ensure the failed check can be retried. + ListenableFuture authResult2 = authState.checkAuthorization(CODEGEN_METHOD); + assertThat(authResult2.isDone()).isFalse(); + + fakePolicyChecker.takeNextAuthRequestOrDie().set(Status.OK); + assertThat(authResult2.get()).isEqualTo(Status.OK); + } + + private static final class FakeServerPolicyChecker implements ServerPolicyChecker { + final LinkedBlockingQueue syncExceptionsToThrow = new LinkedBlockingQueue<>(); + final LinkedBlockingQueue> statusesToSet = new LinkedBlockingQueue<>(); + + @Override + public ListenableFuture checkAuthorizationForServiceAsync(int uid, String serviceName) { + RuntimeException syncException = syncExceptionsToThrow.poll(); + if (syncException != null) { + throw syncException; + } + SettableFuture pendingResult = SettableFuture.create(); + statusesToSet.add(pendingResult); + return pendingResult; + } + + SettableFuture takeNextAuthRequestOrDie() throws InterruptedException { + SettableFuture item = statusesToSet.poll(10, java.util.concurrent.TimeUnit.SECONDS); + if (item == null) { + throw new NoSuchElementException("Queue timed out waiting for item"); + } + return item; + } + } +}